La123123.
/
im-backed
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

1.拉取离线消息改为异步方式，防止请求超时 

2.前端消息列表优化，清除多余div
master
xsx 10 months ago
parent
96c267a714
commit
9be62faf38
3 changed files with 121 additions and 110 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 168
      im-platform/src/main/java/com/bx/implatform/service/impl/GroupMessageServiceImpl.java
  2. 42
      im-platform/src/main/java/com/bx/implatform/service/impl/PrivateMessageServiceImpl.java
  3. 21
      im-web/src/components/chat/ChatBox.vue

168
im-platform/src/main/java/com/bx/implatform/service/impl/GroupMessageServiceImpl.java
View File

@ -14,6 +14,7 @@ import com.bx.imcommon.enums.IMTerminalType;
import com.bx.imcommon.model.IMGroupMessage;
import com.bx.imcommon.model.IMUserInfo;
import com.bx.imcommon.util.CommaTextUtils;
import com.bx.imcommon.util.ThreadPoolExecutorFactory;
import com.bx.implatform.contant.Constant;
import com.bx.implatform.contant.RedisKey;
import com.bx.implatform.dto.GroupMessageDTO;
@ -40,19 +41,21 @@ import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import java.util.*;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
@Slf4j
@Service
@RequiredArgsConstructor
public class GroupMessageServiceImpl extends ServiceImpl<GroupMessageMapper, GroupMessage> implements
GroupMessageService {
public class GroupMessageServiceImpl extends ServiceImpl<GroupMessageMapper, GroupMessage>
implements GroupMessageService {
private final GroupService groupService;
private final GroupMemberService groupMemberService;
private final RedisTemplate<String, Object> redisTemplate;
private final IMClient imClient;
private final SensitiveFilterUtil sensitiveFilterUtil;
private static final ScheduledThreadPoolExecutor EXECUTOR = ThreadPoolExecutorFactory.getThreadPoolExecutor();
@Override
public GroupMessageVO sendMessage(GroupMessageDTO dto) {
@ -67,7 +70,8 @@ public class GroupMessageServiceImpl extends ServiceImpl<GroupMessageMapper, Gro
List<Long> userIds = groupMemberService.findUserIdsByGroupId(group.getId());
if (dto.getReceipt() && userIds.size() > Constant.MAX_LARGE_GROUP_MEMBER) {
// 大群的回执消息过于消耗资源，不允许发送
throw new GlobalException(String.format("当前群聊大于%s人,不支持发送回执消息", Constant.MAX_LARGE_GROUP_MEMBER));
throw new GlobalException(
String.format("当前群聊大于%s人,不支持发送回执消息", Constant.MAX_LARGE_GROUP_MEMBER));
}
// 不用发给自己
userIds = userIds.stream().filter(id -> !session.getUserId().equals(id)).collect(Collectors.toList());
@ -78,11 +82,10 @@ public class GroupMessageServiceImpl extends ServiceImpl<GroupMessageMapper, Gro
msg.setSendNickName(member.getShowNickName());
msg.setAtUserIds(CommaTextUtils.asText(dto.getAtUserIds()));
// 过滤内容中的敏感词
if(MessageType.TEXT.code().equals(dto.getType())){
if (MessageType.TEXT.code().equals(dto.getType())) {
msg.setContent(sensitiveFilterUtil.filter(dto.getContent()));
}
this.save(msg);
// 群发
GroupMessageVO msgInfo = BeanUtils.copyProperties(msg, GroupMessageVO.class);
msgInfo.setAtUserIds(dto.getAtUserIds());
@ -140,97 +143,101 @@ public class GroupMessageServiceImpl extends ServiceImpl<GroupMessageMapper, Gro
return msgInfo;
}
@Override
public void pullOfflineMessage(Long minId) {
UserSession session = SessionContext.getSession();
if(!imClient.isOnline(session.getUserId())){
if (!imClient.isOnline(session.getUserId())) {
throw new GlobalException("网络连接失败，无法拉取离线消息");
}
// 查询用户加入的群组
List<GroupMember> members = groupMemberService.findByUserId(session.getUserId());
Map<Long, GroupMember> groupMemberMap = CollStreamUtil.toIdentityMap(members, GroupMember::getGroupId);
Set<Long> groupIds = groupMemberMap.keySet();
if(CollectionUtil.isEmpty(groupIds)){
if (CollectionUtil.isEmpty(groupIds)) {
// 关闭加载中标志
this.sendLoadingMessage(false);
this.sendLoadingMessage(false, session);
return;
}
// 开启加载中标志
this.sendLoadingMessage(true);
// 只能拉取最近3个月的,最多拉取3000条
// 只能拉取最近3个月的,移动端只拉最近一个月
int months = session.getTerminal().equals(IMTerminalType.APP.code()) ? 1 : 3;
Date minDate = DateUtils.addMonths(new Date(), -months);
LambdaQueryWrapper<GroupMessage> wrapper = Wrappers.lambdaQuery();
wrapper.gt(GroupMessage::getId, minId)
.gt(GroupMessage::getSendTime, minDate)
.in(GroupMessage::getGroupId, groupIds)
.orderByAsc(GroupMessage::getId);
wrapper.gt(GroupMessage::getId, minId).gt(GroupMessage::getSendTime, minDate)
.in(GroupMessage::getGroupId, groupIds).orderByAsc(GroupMessage::getId);
List<GroupMessage> messages = this.list(wrapper);
// 通过群聊对消息进行分组
Map<Long, List<GroupMessage>> messageGroupMap = messages.stream().collect(Collectors.groupingBy(GroupMessage::getGroupId));
Map<Long, List<GroupMessage>> messageGroupMap =
messages.stream().collect(Collectors.groupingBy(GroupMessage::getGroupId));
// 退群前的消息
List<GroupMember> quitMembers = groupMemberService.findQuitInMonth(session.getUserId());
for(GroupMember quitMember: quitMembers){
for (GroupMember quitMember : quitMembers) {
wrapper = Wrappers.lambdaQuery();
wrapper.gt(GroupMessage::getId, minId)
.between(GroupMessage::getSendTime, minDate,quitMember.getQuitTime())
wrapper.gt(GroupMessage::getId, minId).between(GroupMessage::getSendTime, minDate, quitMember.getQuitTime())
.eq(GroupMessage::getGroupId, quitMember.getGroupId())
.ne(GroupMessage::getStatus, MessageStatus.RECALL.code())
.orderByAsc(GroupMessage::getId);
.ne(GroupMessage::getStatus, MessageStatus.RECALL.code()).orderByAsc(GroupMessage::getId);
List<GroupMessage> groupMessages = this.list(wrapper);
messageGroupMap.put(quitMember.getGroupId(),groupMessages);
groupMemberMap.put(quitMember.getGroupId(),quitMember);
messageGroupMap.put(quitMember.getGroupId(), groupMessages);
groupMemberMap.put(quitMember.getGroupId(), quitMember);
}
// 推送消息
AtomicInteger sendCount = new AtomicInteger();
messageGroupMap.forEach((groupId, groupMessages) -> {
// 填充消息状态
String key = StrUtil.join(":", RedisKey.IM_GROUP_READED_POSITION, groupId);
Object o = redisTemplate.opsForHash().get(key, session.getUserId().toString());
long readedMaxId = Objects.isNull(o) ? -1 : Long.parseLong(o.toString());
Map<Object, Object> maxIdMap = null;
for(GroupMessage m:groupMessages){
// 排除加群之前的消息
GroupMember member = groupMemberMap.get(m.getGroupId());
if(DateUtil.compare(member.getCreatedTime(), m.getSendTime()) > 0){
continue;
EXECUTOR.execute(() -> {
// 开启加载中标志
this.sendLoadingMessage(true, session);
// 推送消息
AtomicInteger sendCount = new AtomicInteger();
messageGroupMap.forEach((groupId, groupMessages) -> {
// 第一次拉取时,一个群最多推送1w条消息，防止前端接收能力溢出导致卡顿
List<GroupMessage> sendMessages = groupMessages;
if (minId <= 0 && groupMessages.size() > 10000) {
sendMessages = groupMessages.subList(groupMessages.size() - 10000, groupMessages.size());
}
// 排除不需要接收的消息
List<String> recvIds = CommaTextUtils.asList(m.getRecvIds());
if(!recvIds.isEmpty() && !recvIds.contains(session.getUserId().toString())){
continue;
}
// 组装vo
GroupMessageVO vo = BeanUtils.copyProperties(m, GroupMessageVO.class);
// 被@用户列表
List<String> atIds = CommaTextUtils.asList(m.getAtUserIds());
vo.setAtUserIds(atIds.stream().map(Long::parseLong).collect(Collectors.toList()));
// 填充状态
vo.setStatus(readedMaxId >= m.getId() ? MessageStatus.READED.code() : MessageStatus.UNSEND.code());
// 针对回执消息填充已读人数
if(m.getReceipt()){
if(Objects.isNull(maxIdMap)) {
maxIdMap = redisTemplate.opsForHash().entries(key);
// 填充消息状态
String key = StrUtil.join(":", RedisKey.IM_GROUP_READED_POSITION, groupId);
Object o = redisTemplate.opsForHash().get(key, session.getUserId().toString());
long readedMaxId = Objects.isNull(o) ? -1 : Long.parseLong(o.toString());
Map<Object, Object> maxIdMap = null;
for (GroupMessage m : sendMessages) {
// 排除加群之前的消息
GroupMember member = groupMemberMap.get(m.getGroupId());
if (DateUtil.compare(member.getCreatedTime(), m.getSendTime()) > 0) {
continue;
}
// 排除不需要接收的消息
List<String> recvIds = CommaTextUtils.asList(m.getRecvIds());
if (!recvIds.isEmpty() && !recvIds.contains(session.getUserId().toString())) {
continue;
}
// 组装vo
GroupMessageVO vo = BeanUtils.copyProperties(m, GroupMessageVO.class);
// 被@用户列表
List<String> atIds = CommaTextUtils.asList(m.getAtUserIds());
vo.setAtUserIds(atIds.stream().map(Long::parseLong).collect(Collectors.toList()));
// 填充状态
vo.setStatus(readedMaxId >= m.getId() ? MessageStatus.READED.code() : MessageStatus.UNSEND.code());
// 针对回执消息填充已读人数
if (m.getReceipt()) {
if (Objects.isNull(maxIdMap)) {
maxIdMap = redisTemplate.opsForHash().entries(key);
}
int count = getReadedUserIds(maxIdMap, m.getId(), m.getSendId()).size();
vo.setReadedCount(count);
}
int count = getReadedUserIds(maxIdMap, m.getId(),m.getSendId()).size();
vo.setReadedCount(count);
// 推送
IMGroupMessage<GroupMessageVO> sendMessage = new IMGroupMessage<>();
sendMessage.setSender(new IMUserInfo(m.getSendId(), IMTerminalType.WEB.code()));
sendMessage.setRecvIds(Arrays.asList(session.getUserId()));
sendMessage.setRecvTerminals(Arrays.asList(session.getTerminal()));
sendMessage.setSendResult(false);
sendMessage.setSendToSelf(false);
sendMessage.setData(vo);
imClient.sendGroupMessage(sendMessage);
sendCount.getAndIncrement();
}
// 推送
IMGroupMessage<GroupMessageVO> sendMessage = new IMGroupMessage<>();
sendMessage.setSender(new IMUserInfo(m.getSendId(), IMTerminalType.WEB.code()));
sendMessage.setRecvIds(Arrays.asList(session.getUserId()));
sendMessage.setRecvTerminals(Arrays.asList(session.getTerminal()));
sendMessage.setSendResult(false);
sendMessage.setSendToSelf(false);
sendMessage.setData(vo);
imClient.sendGroupMessage(sendMessage);
sendCount.getAndIncrement();
}
});
// 关闭加载中标志
this.sendLoadingMessage(false, session);
log.info("拉取离线群聊消息,用户id:{},数量:{}", session.getUserId(), sendCount.get());
});
// 关闭加载中标志
this.sendLoadingMessage(false);
log.info("拉取离线群聊消息,用户id:{},数量:{}",session.getUserId(),sendCount.get());
}
@Override
@ -238,10 +245,8 @@ public class GroupMessageServiceImpl extends ServiceImpl<GroupMessageMapper, Gro
UserSession session = SessionContext.getSession();
// 取出最后的消息id
LambdaQueryWrapper<GroupMessage> wrapper = Wrappers.lambdaQuery();
wrapper.eq(GroupMessage::getGroupId, groupId)
.orderByDesc(GroupMessage::getId)
.last("limit 1")
.select(GroupMessage::getId);
wrapper.eq(GroupMessage::getGroupId, groupId).orderByDesc(GroupMessage::getId).last("limit 1")
.select(GroupMessage::getId);
GroupMessage message = this.getOne(wrapper);
if (Objects.isNull(message)) {
return;
@ -276,9 +281,10 @@ public class GroupMessageServiceImpl extends ServiceImpl<GroupMessageMapper, Gro
List<Long> userIds = groupMemberService.findUserIdsByGroupId(groupId);
Map<Object, Object> maxIdMap = redisTemplate.opsForHash().entries(key);
for (GroupMessage receiptMessage : receiptMessages) {
Integer readedCount = getReadedUserIds(maxIdMap, receiptMessage.getId(),receiptMessage.getSendId()).size();
Integer readedCount =
getReadedUserIds(maxIdMap, receiptMessage.getId(), receiptMessage.getSendId()).size();
// 如果所有人都已读，记录回执消息完成标记
if(readedCount >= userIds.size() - 1){
if (readedCount >= userIds.size() - 1) {
receiptMessage.setReceiptOk(true);
this.updateById(receiptMessage);
}
@ -311,12 +317,12 @@ public class GroupMessageServiceImpl extends ServiceImpl<GroupMessageMapper, Gro
if (Objects.isNull(member) || member.getQuit()) {
throw new GlobalException("您已不在群聊里面");
}
// 已读位置key
// 已读位置key
String key = StrUtil.join(":", RedisKey.IM_GROUP_READED_POSITION, groupId);
// 一次获取所有用户的已读位置
Map<Object, Object> maxIdMap = redisTemplate.opsForHash().entries(key);
// 返回已读用户的id集合
return getReadedUserIds(maxIdMap, message.getId(),message.getSendId());
return getReadedUserIds(maxIdMap, message.getId(), message.getSendId());
}
@Override
@ -333,10 +339,11 @@ public class GroupMessageServiceImpl extends ServiceImpl<GroupMessageMapper, Gro
// 查询聊天记录，只查询加入群聊时间之后的消息
QueryWrapper<GroupMessage> wrapper = new QueryWrapper<>();
wrapper.lambda().eq(GroupMessage::getGroupId, groupId).gt(GroupMessage::getSendTime, member.getCreatedTime())
.ne(GroupMessage::getStatus, MessageStatus.RECALL.code()).orderByDesc(GroupMessage::getId).last("limit " + stIdx + "," + size);
.ne(GroupMessage::getStatus, MessageStatus.RECALL.code()).orderByDesc(GroupMessage::getId)
.last("limit " + stIdx + "," + size);
List<GroupMessage> messages = this.list(wrapper);
List<GroupMessageVO> messageInfos =
messages.stream().map(m -> BeanUtils.copyProperties(m, GroupMessageVO.class)).collect(Collectors.toList());
messages.stream().map(m -> BeanUtils.copyProperties(m, GroupMessageVO.class)).collect(Collectors.toList());
log.info("拉取群聊记录，用户id:{},群聊id:{}，数量:{}", userId, groupId, messageInfos.size());
return messageInfos;
}
@ -354,8 +361,7 @@ public class GroupMessageServiceImpl extends ServiceImpl<GroupMessageMapper, Gro
return userIds;
}
private void sendLoadingMessage(Boolean isLoadding){
UserSession session = SessionContext.getSession();
private void sendLoadingMessage(Boolean isLoadding, UserSession session) {
GroupMessageVO msgInfo = new GroupMessageVO();
msgInfo.setType(MessageType.LOADING.code());
msgInfo.setContent(isLoadding.toString());

42
im-platform/src/main/java/com/bx/implatform/service/impl/PrivateMessageServiceImpl.java
View File

@ -10,6 +10,7 @@ import com.bx.imcommon.contant.IMConstant;
import com.bx.imcommon.enums.IMTerminalType;
import com.bx.imcommon.model.IMPrivateMessage;
import com.bx.imcommon.model.IMUserInfo;
import com.bx.imcommon.util.ThreadPoolExecutorFactory;
import com.bx.implatform.dto.PrivateMessageDTO;
import com.bx.implatform.entity.PrivateMessage;
import com.bx.implatform.enums.MessageStatus;
@ -32,6 +33,7 @@ import org.springframework.transaction.annotation.Transactional;
import java.util.Date;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.stream.Collectors;
@Slf4j
@ -43,6 +45,7 @@ public class PrivateMessageServiceImpl extends ServiceImpl<PrivateMessageMapper,
private final FriendService friendService;
private final IMClient imClient;
private final SensitiveFilterUtil sensitiveFilterUtil;
private static final ScheduledThreadPoolExecutor EXECUTOR = ThreadPoolExecutorFactory.getThreadPoolExecutor();
@Override
public PrivateMessageVO sendMessage(PrivateMessageDTO dto) {
@ -135,8 +138,6 @@ public class PrivateMessageServiceImpl extends ServiceImpl<PrivateMessageMapper,
@Override
public void pullOfflineMessage(Long minId) {
UserSession session = SessionContext.getSession();
// 开启加载中标志
this.sendLoadingMessage(true);
// 获取当前用户的消息
LambdaQueryWrapper<PrivateMessage> wrapper = Wrappers.lambdaQuery();
// 只能拉取最近3个月的消息,移动端只拉取一个月消息
@ -148,21 +149,25 @@ public class PrivateMessageServiceImpl extends ServiceImpl<PrivateMessageMapper,
.eq(PrivateMessage::getRecvId, session.getUserId()));
wrapper.orderByAsc(PrivateMessage::getId);
List<PrivateMessage> messages = this.list(wrapper);
// 推送消息
for (PrivateMessage m : messages) {
PrivateMessageVO vo = BeanUtils.copyProperties(m, PrivateMessageVO.class);
IMPrivateMessage<PrivateMessageVO> sendMessage = new IMPrivateMessage<>();
sendMessage.setSender(new IMUserInfo(m.getSendId(), IMTerminalType.WEB.code()));
sendMessage.setRecvId(session.getUserId());
sendMessage.setRecvTerminals(List.of(session.getTerminal()));
sendMessage.setSendToSelf(false);
sendMessage.setData(vo);
sendMessage.setSendResult(true);
imClient.sendPrivateMessage(sendMessage);
}
// 关闭加载中标志
this.sendLoadingMessage(false);
log.info("拉取私聊消息，用户id:{},数量:{}", session.getUserId(), messages.size());
// 异步推送消息
EXECUTOR.execute(() -> {
// 开启加载中标志
this.sendLoadingMessage(true, session);
for (PrivateMessage m : messages) {
PrivateMessageVO vo = BeanUtils.copyProperties(m, PrivateMessageVO.class);
IMPrivateMessage<PrivateMessageVO> sendMessage = new IMPrivateMessage<>();
sendMessage.setSender(new IMUserInfo(m.getSendId(), IMTerminalType.WEB.code()));
sendMessage.setRecvId(session.getUserId());
sendMessage.setRecvTerminals(List.of(session.getTerminal()));
sendMessage.setSendToSelf(false);
sendMessage.setData(vo);
sendMessage.setSendResult(true);
imClient.sendPrivateMessage(sendMessage);
}
// 关闭加载中标志
this.sendLoadingMessage(false, session);
log.info("拉取私聊消息，用户id:{},数量:{}", session.getUserId(), messages.size());
});
}
@Transactional(rollbackFor = Exception.class)
@ -215,8 +220,7 @@ public class PrivateMessageServiceImpl extends ServiceImpl<PrivateMessageMapper,
return message.getId();
}
private void sendLoadingMessage(Boolean isLoadding) {
UserSession session = SessionContext.getSession();
private void sendLoadingMessage(Boolean isLoadding, UserSession session) {
PrivateMessageVO msgInfo = new PrivateMessageVO();
msgInfo.setType(MessageType.LOADING.code());
msgInfo.setContent(isLoadding.toString());

21
im-web/src/components/chat/ChatBox.vue
View File

@ -10,16 +10,13 @@
<el-container class="content-box">
<el-main class="im-chat-main" id="chatScrollBox" @scroll="onScroll">
<div class="im-chat-box">
<ul>
<li v-for="(msgInfo, idx) in chat.messages" :key="idx">
<chat-message-item v-if="idx >= showMinIdx" @call="onCall(msgInfo.type)"
:mine="msgInfo.sendId == mine.id" :headImage="headImage(msgInfo)"
:showName="showName(msgInfo)" :msgInfo="msgInfo"
:groupMembers="groupMembers" @delete="deleteMessage"
@recall="recallMessage">
</chat-message-item>
</li>
</ul>
<div v-for="(msgInfo, idx) in showMessages" :key="showMinIdx + idx">
<chat-message-item @call="onCall(msgInfo.type)"
:mine="msgInfo.sendId == mine.id" :headImage="headImage(msgInfo)"
:showName="showName(msgInfo)" :msgInfo="msgInfo" :groupMembers="groupMembers"
@delete="deleteMessage" @recall="recallMessage">
</chat-message-item>
</div>
</div>
</el-main>
<div v-if="!isInBottom" class="scroll-to-bottom" @click="scrollToBottom">
@ -684,6 +681,10 @@ export default {
unreadCount() {
return this.chat.unreadCount;
},
showMessages() {
console.log("this.chat.messages.slice(this.showMinIdx):",this.chat.messages.slice(this.showMinIdx))
return this.chat.messages.slice(this.showMinIdx)
},
messageSize() {
if (!this.chat || !this.chat.messages) {
return 0;

Write Preview
Loading…
Cancel
Save