|
|
|
@ -158,13 +158,17 @@ public class GroupMessageServiceImpl extends ServiceImpl<GroupMessageMapper, Gro |
|
|
|
this.sendLoadingMessage(false, session); |
|
|
|
return; |
|
|
|
} |
|
|
|
|
|
|
|
// 只能拉取最近3个月的,移动端只拉最近一个月
|
|
|
|
int months = session.getTerminal().equals(IMTerminalType.APP.code()) ? 1 : 3; |
|
|
|
Date minDate = DateUtils.addMonths(new Date(), -months); |
|
|
|
// 只拉最近一个月
|
|
|
|
Date minDate = DateUtils.addMonths(new Date(), -1); |
|
|
|
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); |
|
|
|
wrapper.gt(GroupMessage::getSendTime, minDate); |
|
|
|
wrapper.in(GroupMessage::getGroupId, groupIds); |
|
|
|
wrapper.orderByDesc(GroupMessage::getId); |
|
|
|
if (minId <= 0) { |
|
|
|
// 首次拉取限制消息数量大小，防止内存溢出
|
|
|
|
wrapper.last("limit 100000"); |
|
|
|
} |
|
|
|
List<GroupMessage> messages = this.list(wrapper); |
|
|
|
// 通过群聊对消息进行分组
|
|
|
|
Map<Long, List<GroupMessage>> messageGroupMap = |
|
|
|
@ -173,9 +177,11 @@ public class GroupMessageServiceImpl extends ServiceImpl<GroupMessageMapper, Gro |
|
|
|
List<GroupMember> quitMembers = groupMemberService.findQuitInMonth(session.getUserId()); |
|
|
|
for (GroupMember quitMember : quitMembers) { |
|
|
|
wrapper = Wrappers.lambdaQuery(); |
|
|
|
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); |
|
|
|
wrapper.gt(GroupMessage::getId, minId); |
|
|
|
wrapper.between(GroupMessage::getSendTime, minDate, quitMember.getQuitTime()); |
|
|
|
wrapper.eq(GroupMessage::getGroupId, quitMember.getGroupId()); |
|
|
|
wrapper.ne(GroupMessage::getStatus, MessageStatus.RECALL.code()); |
|
|
|
wrapper.orderByDesc(GroupMessage::getId); |
|
|
|
List<GroupMessage> groupMessages = this.list(wrapper); |
|
|
|
messageGroupMap.put(quitMember.getGroupId(), groupMessages); |
|
|
|
groupMemberMap.put(quitMember.getGroupId(), quitMember); |
|
|
|
@ -184,19 +190,28 @@ public class GroupMessageServiceImpl extends ServiceImpl<GroupMessageMapper, Gro |
|
|
|
// 开启加载中标志
|
|
|
|
this.sendLoadingMessage(true, session); |
|
|
|
// 推送消息
|
|
|
|
AtomicInteger sendCount = new AtomicInteger(); |
|
|
|
messageGroupMap.forEach((groupId, groupMessages) -> { |
|
|
|
// 第一次拉取时,一个群最多推送1w条消息，防止前端接收能力溢出导致卡顿
|
|
|
|
int sendCount = 0; |
|
|
|
for (Map.Entry<Long, List<GroupMessage>> entry : messageGroupMap.entrySet()) { |
|
|
|
Long groupId = entry.getKey(); |
|
|
|
List<GroupMessage> groupMessages = entry.getValue(); |
|
|
|
// 第一次拉取时,一个群最多推送3000条消息，防止前端接收能力溢出导致卡顿
|
|
|
|
List<GroupMessage> sendMessages = groupMessages; |
|
|
|
if (minId <= 0 && groupMessages.size() > 10000) { |
|
|
|
sendMessages = groupMessages.subList(groupMessages.size() - 10000, groupMessages.size()); |
|
|
|
if (minId <= 0 && groupMessages.size() > 3000) { |
|
|
|
sendMessages = groupMessages.subList(0, 3000); |
|
|
|
} |
|
|
|
// id从小到大排序
|
|
|
|
CollectionUtil.reverse(sendMessages); |
|
|
|
// 填充消息状态
|
|
|
|
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) { |
|
|
|
// 推送过程如果用户下线了，则不再推送
|
|
|
|
if (!imClient.isOnline(session.getUserId(), IMTerminalType.fromCode(session.getTerminal()))) { |
|
|
|
log.info("用户已下线，停止推送离线群聊消息,用户id:{}", session.getUserId()); |
|
|
|
return; |
|
|
|
} |
|
|
|
// 排除加群之前的消息
|
|
|
|
GroupMember member = groupMemberMap.get(m.getGroupId()); |
|
|
|
if (DateUtil.compare(member.getCreatedTime(), m.getSendTime()) > 0) { |
|
|
|
@ -231,12 +246,12 @@ public class GroupMessageServiceImpl extends ServiceImpl<GroupMessageMapper, Gro |
|
|
|
sendMessage.setSendToSelf(false); |
|
|
|
sendMessage.setData(vo); |
|
|
|
imClient.sendGroupMessage(sendMessage); |
|
|
|
sendCount.getAndIncrement(); |
|
|
|
sendCount++; |
|
|
|
} |
|
|
|
} |
|
|
|
}); |
|
|
|
// 关闭加载中标志
|
|
|
|
this.sendLoadingMessage(false, session); |
|
|
|
log.info("拉取离线群聊消息,用户id:{},数量:{}", session.getUserId(), sendCount.get()); |
|
|
|
log.info("拉取离线群聊消息,用户id:{},数量:{}", session.getUserId(), sendCount++); |
|
|
|
}); |
|
|
|
} |
|
|
|
|
|
|
|
|
blue
9 months ago
Gitee