Commit 0ce9eddedb339477335bdfcffec793a50ac97e3f
1 parent
bf8eb273
✨ feat(message): 优化群聊的 pullGroupMessageList 方法,解决空分页的风险
Showing
2 changed files
with
89 additions
and
13 deletions
yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/service/group/ImGroupServiceImpl.java
| ... | ... | @@ -4,12 +4,20 @@ import org.springframework.stereotype.Service; |
| 4 | 4 | import jakarta.annotation.Resource; |
| 5 | 5 | import org.springframework.validation.annotation.Validated; |
| 6 | 6 | |
| 7 | + | |
| 8 | +import cn.iocoder.yudao.framework.common.util.collection.CollectionUtils; | |
| 7 | 9 | import cn.iocoder.yudao.module.im.controller.admin.group.vo.*; |
| 8 | 10 | import cn.iocoder.yudao.module.im.dal.dataobject.group.ImGroupDO; |
| 11 | +import cn.iocoder.yudao.module.im.dal.dataobject.group.ImGroupMemberDO; | |
| 9 | 12 | import cn.iocoder.yudao.framework.common.pojo.PageResult; |
| 10 | 13 | import cn.iocoder.yudao.framework.common.util.object.BeanUtils; |
| 11 | 14 | |
| 12 | 15 | import cn.iocoder.yudao.module.im.dal.mysql.group.ImGroupMapper; |
| 16 | +import cn.iocoder.yudao.module.im.dal.mysql.group.ImGroupMemberMapper; | |
| 17 | + | |
| 18 | +import java.util.Collections; | |
| 19 | +import java.util.List; | |
| 20 | +import java.util.Set; | |
| 13 | 21 | |
| 14 | 22 | import static cn.iocoder.yudao.framework.common.exception.util.ServiceExceptionUtil.exception; |
| 15 | 23 | import static cn.iocoder.yudao.module.im.enums.ErrorCodeConstants.*; |
| ... | ... | @@ -26,6 +34,9 @@ public class ImGroupServiceImpl implements ImGroupService { |
| 26 | 34 | @Resource |
| 27 | 35 | private ImGroupMapper imGroupMapper; |
| 28 | 36 | |
| 37 | + @Resource | |
| 38 | + private ImGroupMemberMapper imGroupMemberMapper; | |
| 39 | + | |
| 29 | 40 | @Override |
| 30 | 41 | public Long createGroup(ImGroupSaveReqVO createReqVO) { |
| 31 | 42 | // 插入 |
| ... | ... | @@ -77,4 +88,21 @@ public class ImGroupServiceImpl implements ImGroupService { |
| 77 | 88 | return imGroupMapper.selectPage(pageReqVO); |
| 78 | 89 | } |
| 79 | 90 | |
| 91 | + @Override | |
| 92 | + public List<ImGroupDO> getMyGroupList(Long userId) { | |
| 93 | + // 1. 查用户所在的、仍有效的群成员记录(仅 ENABLE 状态) | |
| 94 | + List<ImGroupMemberDO> members = imGroupMemberMapper.selectEnabledListByUserId(userId); | |
| 95 | + if (members.isEmpty()) { | |
| 96 | + return Collections.emptyList(); | |
| 97 | + } | |
| 98 | + Set<Long> groupIds = CollectionUtils.convertSet(members, ImGroupMemberDO::getGroupId); | |
| 99 | + if (groupIds.isEmpty()) { | |
| 100 | + return Collections.emptyList(); | |
| 101 | + } | |
| 102 | + // 2. 查群并过滤掉已封禁/解散的 | |
| 103 | + List<ImGroupDO> groups = imGroupMapper.selectByIds(groupIds); | |
| 104 | + groups.removeIf(g -> Boolean.TRUE.equals(g.getBanned()) || Boolean.TRUE.equals(g.getDissolved())); | |
| 105 | + return groups; | |
| 106 | + } | |
| 107 | + | |
| 80 | 108 | } |
| 81 | 109 | \ No newline at end of file | ... | ... |
yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/service/message/ImGroupMessageServiceImpl.java
| ... | ... | @@ -49,6 +49,13 @@ import static cn.iocoder.yudao.module.im.enums.ImCommonConstants.*; |
| 49 | 49 | @Slf4j |
| 50 | 50 | public class ImGroupMessageServiceImpl implements ImGroupMessageService { |
| 51 | 51 | |
| 52 | + /** | |
| 53 | + * 仅用于规避群消息 pull 的"假空页": | |
| 54 | + * 首批 raw 消息可能因入群时间或定向接收过滤后变成空列表,但后续更大的 id 仍然存在可见消息。 | |
| 55 | + * 因此仅在过滤结果为空时,按本轮 raw 最大 id 向后再试几次。 | |
| 56 | + */ | |
| 57 | + private static final int PULL_GROUP_MESSAGE_EMPTY_RETRY_TIMES = 3; | |
| 58 | + | |
| 52 | 59 | @Resource |
| 53 | 60 | private ImGroupMessageMapper groupMessageMapper; |
| 54 | 61 | @Resource |
| ... | ... | @@ -100,12 +107,50 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService { |
| 100 | 107 | if (size > MESSAGE_MAX_PULL_SIZE) { |
| 101 | 108 | throw exception(MESSAGE_PULL_SIZE_EXCEEDED); |
| 102 | 109 | } |
| 110 | + | |
| 103 | 111 | // 0. 拉取时间窗:超过窗口的老消息不再通过离线通道推送 |
| 104 | 112 | LocalDateTime minSendTime = LocalDateTime.now().minusDays(MESSAGE_GROUP_PULL_MAX_DAYS); |
| 105 | 113 | |
| 114 | + // 1. 拉取一批原始消息(包含当前在群、退群前补齐两部分),并过滤得到当前用户可见的结果 | |
| 115 | + List<ImGroupMessageDO> result = Collections.emptyList(); | |
| 116 | + Long currentMinId = minId; // 当前轮次的下限消息编号;初始为用户传入的 minId,后续会根据拉取结果更新 | |
| 117 | + for (int retryCount = 0; retryCount <= PULL_GROUP_MESSAGE_EMPTY_RETRY_TIMES; retryCount++) { | |
| 118 | + // 1.1 查询一批原始消息(包含当前在群、退群前补齐两部分) | |
| 119 | + Map<Long, ImGroupMemberDO> memberMap = new HashMap<>(); // 群编号 → 群成员记录(包含当前在群、退群前补齐两部分) | |
| 120 | + List<ImGroupMessageDO> messages = getGroupMessageList(userId, currentMinId, size, minSendTime, memberMap); | |
| 121 | + if (CollUtil.isEmpty(messages)) { | |
| 122 | + break; | |
| 123 | + } | |
| 124 | + // 1.2 过滤得到当前用户可见的结果(按 id 升序) | |
| 125 | + result = filterGroupMessageList(messages, memberMap, userId, size); | |
| 126 | + // 已拿到可见消息,或原始消息已不足一页(说明没有继续重试的必要) | |
| 127 | + if (CollUtil.isNotEmpty(result) || messages.size() < size) { | |
| 128 | + break; | |
| 129 | + } | |
| 130 | + | |
| 131 | + // 1.3 更新 currentMinId,进行下一轮查询;如果本轮查询的最大消息编号没有前进,则说明后续没有更多消息了,直接 break | |
| 132 | + Long maxMessageId = getMaxValue(messages, ImGroupMessageDO::getId); | |
| 133 | + if (maxMessageId == null || maxMessageId <= currentMinId) { | |
| 134 | + break; | |
| 135 | + } | |
| 136 | + currentMinId = maxMessageId; | |
| 137 | + } | |
| 138 | + | |
| 139 | + // 2. 按当前用户补齐:消息已读态(相对 Redis 已读游标)、本人发送的回执消息的已读人数 | |
| 140 | + appendMessageStatusAndReceipt(userId, result); | |
| 141 | + log.info("[pullGroupMessageList][userId({}) minId({}) size({}) result({})]", userId, minId, size, result.size()); | |
| 142 | + return result; | |
| 143 | + } | |
| 144 | + | |
| 145 | + /** | |
| 146 | + * 查询一批原始群消息(包含当前在群、退群前补齐两部分) | |
| 147 | + */ | |
| 148 | + private List<ImGroupMessageDO> getGroupMessageList(Long userId, Long minId, Integer size, | |
| 149 | + LocalDateTime minSendTime, | |
| 150 | + Map<Long, ImGroupMemberDO> memberMap) { | |
| 106 | 151 | // 1. 主查询:仅用"当前仍在群"的成员记录驱动(已退群的群由步骤 2 补齐) |
| 107 | 152 | List<ImGroupMemberDO> activeMembers = groupMemberService.getActiveGroupMemberListByUserId(userId); |
| 108 | - Map<Long, ImGroupMemberDO> memberMap = convertMap(activeMembers, ImGroupMemberDO::getGroupId); | |
| 153 | + memberMap.putAll(convertMap(activeMembers, ImGroupMemberDO::getGroupId)); | |
| 109 | 154 | List<ImGroupMessageDO> messages = new ArrayList<>(); |
| 110 | 155 | if (CollUtil.isNotEmpty(activeMembers)) { |
| 111 | 156 | List<Long> groupIds = convertList(activeMembers, ImGroupMemberDO::getGroupId); |
| ... | ... | @@ -128,25 +173,28 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService { |
| 128 | 173 | for (ImGroupMemberDO quitMember : quitMembers) { |
| 129 | 174 | List<ImGroupMessageDO> quitGroupMessages = groupMessageMapper.selectListByGroupIdAndMinIdAndQuitTimeBefore( |
| 130 | 175 | quitMember.getGroupId(), minId, minSendTime, quitMember.getQuitTime(), size); |
| 131 | - if (CollUtil.isNotEmpty(quitGroupMessages)) { | |
| 132 | - messages.addAll(quitGroupMessages); | |
| 133 | - memberMap.put(quitMember.getGroupId(), quitMember); | |
| 176 | + if (CollUtil.isEmpty(quitGroupMessages)) { | |
| 177 | + continue; | |
| 134 | 178 | } |
| 179 | + messages.addAll(quitGroupMessages); | |
| 180 | + memberMap.put(quitMember.getGroupId(), quitMember); | |
| 135 | 181 | } |
| 182 | + return messages; | |
| 183 | + } | |
| 136 | 184 | |
| 137 | - // 3.1 按可见性过滤(入群前不可见、定向消息排除),按 id 升序后仅取本页 size 条, | |
| 138 | - // 避免「在群 + 退群前」多路合并时一次响应跨度过大、游标直接跳到全局最大 id 而漏拉中间消息 | |
| 139 | - List<ImGroupMessageDO> result = messages.stream() | |
| 185 | + /** | |
| 186 | + * 过滤一批原始群消息,得到当前用户可见的返回结果 | |
| 187 | + */ | |
| 188 | + private List<ImGroupMessageDO> filterGroupMessageList(List<ImGroupMessageDO> messages, | |
| 189 | + Map<Long, ImGroupMemberDO> memberMap, | |
| 190 | + Long userId, Integer size) { | |
| 191 | + // 按可见性过滤(入群前不可见、定向消息排除),按 id 升序后仅取本页 size 条, | |
| 192 | + // 避免「在群 + 退群前」多路合并时一次响应跨度过大、游标直接跳到全局最大 id 而漏拉中间消息 | |
| 193 | + return messages.stream() | |
| 140 | 194 | .filter(msg -> isMessageVisible(msg, memberMap.get(msg.getGroupId()), userId)) |
| 141 | 195 | .sorted(Comparator.comparing(ImGroupMessageDO::getId)) |
| 142 | 196 | .limit(size) |
| 143 | 197 | .toList(); |
| 144 | - // 3.2 按当前用户补齐:消息已读态(相对 Redis 已读游标)、本人发送的回执消息的已读人数 | |
| 145 | - appendMessageStatusAndReceipt(userId, result); | |
| 146 | - | |
| 147 | - log.info("[pullGroupMessageList][userId({}) minId({}) size({}) result({})]", | |
| 148 | - userId, minId, size, result.size()); | |
| 149 | - return result; | |
| 150 | 198 | } |
| 151 | 199 | |
| 152 | 200 | /** | ... | ... |