From 99121a25e0ea9e526a3db42dba1455c7075600e0 Mon Sep 17 00:00:00 2001 From: YunaiV Date: Tue, 21 Apr 2026 20:02:47 +0800 Subject: [PATCH] ✨ feat(message): 优化群聊的 pullGroupMessageList 方法，解决空分页的风险(pro 增强） --- yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/service/message/ImGroupMessageServiceImpl.java | 98 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++---------------------------------------- yudao-module-im/yudao-module-im-biz/src/test/java/cn/iocoder/yudao/module/im/service/message/ImGroupMessageServiceImplTest.java | 109 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++----- 2 files changed, 162 insertions(+), 45 deletions(-) diff --git a/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/service/message/ImGroupMessageServiceImpl.java b/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/service/message/ImGroupMessageServiceImpl.java index 9882aad..e41b718 100644 --- a/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/service/message/ImGroupMessageServiceImpl.java +++ b/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/service/message/ImGroupMessageServiceImpl.java @@ -51,8 +51,8 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService { /** * 仅用于规避群消息 pull 的"假空页"： - * 首批 raw 消息可能因入群时间或定向接收过滤后变成空列表，但后续更大的 id 仍然存在可见消息。 - * 因此仅在过滤结果为空时，按本轮 raw 最大 id 向后再试几次。 + * 首批消息可能因入群时间或定向接收过滤后变成空列表，但后续更大的 id 仍然存在可见消息。 + * 因此仅在过滤结果为空时，按本轮消息最大 id 向后再试几次。 */ private static final int PULL_GROUP_MESSAGE_EMPTY_RETRY_TIMES = 3; @@ -111,56 +111,71 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService { // 0. 拉取时间窗：超过窗口的老消息不再通过离线通道推送 LocalDateTime minSendTime = LocalDateTime.now().minusDays(MESSAGE_GROUP_PULL_MAX_DAYS); - // 1. 拉取一批原始消息（包含当前在群、退群前补齐两部分），并过滤得到当前用户可见的结果 - List result = Collections.emptyList(); - Long currentMinId = minId; // 当前轮次的下限消息编号；初始为用户传入的 minId，后续会根据拉取结果更新 + // 1.1 主查询：仅用"当前仍在群"的成员记录驱动；若首批消息过滤后为空，则允许内部重试 + List activeMembers = groupMemberService.getActiveGroupMemberListByUserId(userId); + Map memberMap = convertMap(activeMembers, ImGroupMemberDO::getGroupId); + List messages = new ArrayList<>(); + if (CollUtil.isNotEmpty(activeMembers)) { + List groupIds = convertList(activeMembers, ImGroupMemberDO::getGroupId); + messages.addAll(pullActiveGroupMessageList(userId, groupIds, minId, size, minSendTime, memberMap)); + } + // 1.2 补齐"退群前"的消息： + // - 继续基于用户原始 minId 单次查询，不能被主查询内部探测游标带着前进； + // - 若 minId > 0 且能查到对应消息，则进一步把下限抬到该消息的 sendTime，避免把客户端已拥有的老消息再次推送。 + messages.addAll(pullQuitGroupMessageList(userId, minId, size, minSendTime, memberMap)); + + // 2. 合并后统一过滤，得到当前用户可见的结果 + List result = filterGroupMessageList(messages, memberMap, userId, size); + + // 3. 按当前用户补齐：消息已读态（相对 Redis 已读游标）、本人发送的回执消息的已读人数 + appendMessageStatusAndReceipt(userId, result); + log.info("[pullGroupMessageList][userId({}) minId({}) size({}) result({})]", userId, minId, size, result.size()); + return result; + } + + /** + * 拉取当前仍在群主路径的消息。 + * + * 仅当首批消息过滤后仍无可见消息时，才按消息最大 id 继续向后探测， + * 直到命中可见消息或确认该来源已耗尽。 + */ + private List pullActiveGroupMessageList(Long userId, List groupIds, Long minId, + Integer size, LocalDateTime minSendTime, + Map memberMap) { + // 1. 主查询内部探测游标：仅用于向后探测，不代表客户端真实已送达边界 + Long activeMinId = minId; for (int retryCount = 0; retryCount <= PULL_GROUP_MESSAGE_EMPTY_RETRY_TIMES; retryCount++) { - // 1.1 查询一批原始消息（包含当前在群、退群前补齐两部分） - Map memberMap = new HashMap<>(); // 群编号 → 群成员记录（包含当前在群、退群前补齐两部分） - List messages = getGroupMessageList(userId, currentMinId, size, minSendTime, memberMap); + // 2. 查询本轮消息；若已无更多消息，则当前主路径直接结束 + List messages = groupMessageMapper.selectListByMinId(groupIds, activeMinId, + minSendTime, size); if (CollUtil.isEmpty(messages)) { - break; + return Collections.emptyList(); } - // 1.2 过滤得到当前用户可见的结果（按 id 升序） - result = filterGroupMessageList(messages, memberMap, userId, size); - // 已拿到可见消息，或原始消息已不足一页（说明没有继续重试的必要） - if (CollUtil.isNotEmpty(result) || messages.size() < size) { - break; + boolean hasVisibleMessage = CollUtil.anyMatch(messages, + message -> isMessageVisible(message, memberMap.get(message.getGroupId()), userId)); + boolean sourceExhausted = messages.size() < size; + // 3. 本轮已命中可见消息，或主查询来源已耗尽，直接返回这一轮消息 + if (hasVisibleMessage || sourceExhausted) { + return messages; } - // 1.3 更新 currentMinId，进行下一轮查询；如果本轮查询的最大消息编号没有前进，则说明后续没有更多消息了，直接 break + // 4. 按本轮消息最大 id 推进内部游标，跳过这段不可见区间；若游标未前进则直接停止 Long maxMessageId = getMaxValue(messages, ImGroupMessageDO::getId); - if (maxMessageId == null || maxMessageId <= currentMinId) { - break; + if (maxMessageId <= activeMinId) { + return Collections.emptyList(); } - currentMinId = maxMessageId; + activeMinId = maxMessageId; } - - // 2. 按当前用户补齐：消息已读态（相对 Redis 已读游标）、本人发送的回执消息的已读人数 - appendMessageStatusAndReceipt(userId, result); - log.info("[pullGroupMessageList][userId({}) minId({}) size({}) result({})]", userId, minId, size, result.size()); - return result; + return Collections.emptyList(); } /** - * 查询一批原始群消息（包含当前在群、退群前补齐两部分） + * 拉取已离开群路径的消息（退群前消息） */ - private List getGroupMessageList(Long userId, Long minId, Integer size, - LocalDateTime minSendTime, - Map memberMap) { - // 1. 主查询：仅用"当前仍在群"的成员记录驱动（已退群的群由步骤 2 补齐） - List activeMembers = groupMemberService.getActiveGroupMemberListByUserId(userId); - memberMap.putAll(convertMap(activeMembers, ImGroupMemberDO::getGroupId)); - List messages = new ArrayList<>(); - if (CollUtil.isNotEmpty(activeMembers)) { - List groupIds = convertList(activeMembers, ImGroupMemberDO::getGroupId); - messages.addAll(groupMessageMapper.selectListByMinId(groupIds, minId, minSendTime, size)); - } - - // 2. 补齐"退群前"的消息： - // - 退群时间早于窗口起点的群直接忽略（消息已不在拉取范围）； - // - 若 minId > 0 且能查到对应消息，则进一步把下限抬到该消息的 sendTime， - // 避免把客户端已拥有的老消息再次推送。 + private List pullQuitGroupMessageList(Long userId, Long minId, Integer size, + LocalDateTime minSendTime, + Map memberMap) { + // 1. 退群补齐始终基于用户原始 minId 计算时间边界，避免被主查询内部重试游标误伤 LocalDateTime minQuitTime = minSendTime; if (minId != null && minId > 0) { ImGroupMessageDO minMessage = groupMessageMapper.selectById(minId); @@ -169,8 +184,11 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService { minQuitTime = minMessage.getSendTime(); } } + // 2. 查询用户离开的群记录；若原始 minId 对应消息仍在窗口内，则用它的发送时间抬升退群筛选下限 + List messages = new ArrayList<>(); List quitMembers = groupMemberService.getQuitGroupMemberListByUserId(userId, minQuitTime); for (ImGroupMemberDO quitMember : quitMembers) { + // 3. 按原始 minId + 退群时间补齐该群退群前消息，并把成员记录写回 memberMap 供统一可见性过滤使用 List quitGroupMessages = groupMessageMapper.selectListByGroupIdAndMinIdAndQuitTimeBefore( quitMember.getGroupId(), minId, minSendTime, quitMember.getQuitTime(), size); if (CollUtil.isEmpty(quitGroupMessages)) { diff --git a/yudao-module-im/yudao-module-im-biz/src/test/java/cn/iocoder/yudao/module/im/service/message/ImGroupMessageServiceImplTest.java b/yudao-module-im/yudao-module-im-biz/src/test/java/cn/iocoder/yudao/module/im/service/message/ImGroupMessageServiceImplTest.java index 355dad7..12b94e1 100644 --- a/yudao-module-im/yudao-module-im-biz/src/test/java/cn/iocoder/yudao/module/im/service/message/ImGroupMessageServiceImplTest.java +++ b/yudao-module-im/yudao-module-im-biz/src/test/java/cn/iocoder/yudao/module/im/service/message/ImGroupMessageServiceImplTest.java @@ -185,6 +185,45 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest { } @Test + public void testPullMessages_activeQueryRetriesWhenFirstBatchInvisible() { + LocalDateTime now = LocalDateTime.now(); + LocalDateTime joinTime = now.minusHours(1); + ImGroupMemberDO member = ImGroupMemberDO.builder() + .groupId(10L).userId(1L) + .status(CommonStatusEnum.ENABLE.getStatus()) + .joinTime(joinTime).build(); + when(groupMemberService.getActiveGroupMemberListByUserId(1L)).thenReturn(List.of(member)); + + ImGroupMessageDO beforeJoin = ImGroupMessageDO.builder() + .id(1L).groupId(10L).senderId(2L) + .sendTime(now.minusHours(3)).build(); + ImGroupMessageDO directedMsg = ImGroupMessageDO.builder() + .id(2L).groupId(10L).senderId(2L) + .receiverUserIds(List.of(2L, 3L)) + .sendTime(now.minusHours(2)).build(); + ImGroupMessageDO visibleMsg = ImGroupMessageDO.builder() + .id(3L).groupId(10L).senderId(2L) + .sendTime(now.minusMinutes(30)).build(); + when(groupMessageMapper.selectListByMinId(eq(List.of(10L)), eq(0L), + any(LocalDateTime.class), eq(2))) + .thenReturn(List.of(beforeJoin, directedMsg)); + when(groupMessageMapper.selectListByMinId(eq(List.of(10L)), eq(2L), + any(LocalDateTime.class), eq(2))) + .thenReturn(List.of(visibleMsg)); + when(groupMemberService.getQuitGroupMemberListByUserId(eq(1L), any(LocalDateTime.class))) + .thenReturn(List.of()); + + List result = groupMessageService.pullGroupMessageList(1L, 0L, 2); + + assertEquals(1, result.size()); + assertEquals(3L, result.get(0).getId()); + verify(groupMessageMapper).selectListByMinId(eq(List.of(10L)), eq(0L), + any(LocalDateTime.class), eq(2)); + verify(groupMessageMapper).selectListByMinId(eq(List.of(10L)), eq(2L), + any(LocalDateTime.class), eq(2)); + } + + @Test public void testPullMessages_quitMemberLoadsPreQuitMessages() { // 准备：当前群（在群）+ 一个最近退群 LocalDateTime now = LocalDateTime.now(); @@ -232,8 +271,6 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest { // 准备：minId > 0，边界消息 sendTime 晚于窗口起点，应当被用作 minQuitTime LocalDateTime now = LocalDateTime.now(); when(groupMemberService.getActiveGroupMemberListByUserId(1L)).thenReturn(List.of()); - when(groupMessageMapper.selectListByMinId(anyList(), anyLong(), any(LocalDateTime.class), anyInt())) - .thenReturn(List.of()); LocalDateTime boundarySendTime = now.minusDays(2); ImGroupMessageDO boundary = ImGroupMessageDO.builder() @@ -251,6 +288,70 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest { } @Test + public void testPullMessages_quitQueryKeepsOriginalMinIdWhenActiveQueryRetries() { + LocalDateTime now = LocalDateTime.now(); + LocalDateTime joinTime = now.minusHours(1); + ImGroupMemberDO activeMember = ImGroupMemberDO.builder() + .groupId(10L).userId(1L) + .status(CommonStatusEnum.ENABLE.getStatus()) + .joinTime(joinTime).build(); + when(groupMemberService.getActiveGroupMemberListByUserId(1L)).thenReturn(List.of(activeMember)); + + ImGroupMessageDO beforeJoin = ImGroupMessageDO.builder() + .id(100L).groupId(10L).senderId(2L) + .sendTime(now.minusHours(3)).build(); + ImGroupMessageDO directedMsg = ImGroupMessageDO.builder() + .id(101L).groupId(10L).senderId(2L) + .receiverUserIds(List.of(2L, 3L)) + .sendTime(now.minusMinutes(50)).build(); + ImGroupMessageDO visibleActiveMsg = ImGroupMessageDO.builder() + .id(102L).groupId(10L).senderId(2L) + .sendTime(now.minusMinutes(30)).build(); + when(groupMessageMapper.selectListByMinId(eq(List.of(10L)), eq(8L), + any(LocalDateTime.class), eq(2))) + .thenReturn(List.of(beforeJoin, directedMsg)); + when(groupMessageMapper.selectListByMinId(eq(List.of(10L)), eq(101L), + any(LocalDateTime.class), eq(2))) + .thenReturn(List.of(visibleActiveMsg)); + + LocalDateTime boundarySendTime = now.minusDays(2); + ImGroupMessageDO boundary = ImGroupMessageDO.builder() + .id(8L).groupId(99L).senderId(2L).sendTime(boundarySendTime).build(); + when(groupMessageMapper.selectById(8L)).thenReturn(boundary); + + LocalDateTime quitTime = now.minusHours(12); + ImGroupMemberDO quitMember = ImGroupMemberDO.builder() + .groupId(20L).userId(1L) + .status(CommonStatusEnum.DISABLE.getStatus()) + .joinTime(now.minusDays(10)) + .quitTime(quitTime).build(); + when(groupMemberService.getQuitGroupMemberListByUserId(1L, boundarySendTime)) + .thenReturn(List.of(quitMember)); + + ImGroupMessageDO quitGroupMsg = ImGroupMessageDO.builder() + .id(50L).groupId(20L).senderId(99L) + .sendTime(now.minusDays(1)).build(); + when(groupMessageMapper.selectListByGroupIdAndMinIdAndQuitTimeBefore(eq(20L), eq(8L), + any(LocalDateTime.class), eq(quitTime), eq(2))) + .thenReturn(List.of(quitGroupMsg)); + + List result = groupMessageService.pullGroupMessageList(1L, 8L, 2); + + assertEquals(2, result.size()); + assertEquals(50L, result.get(0).getId()); + assertEquals(102L, result.get(1).getId()); + verify(groupMessageMapper).selectListByMinId(eq(List.of(10L)), eq(8L), + any(LocalDateTime.class), eq(2)); + verify(groupMessageMapper).selectListByMinId(eq(List.of(10L)), eq(101L), + any(LocalDateTime.class), eq(2)); + verify(groupMemberService).getQuitGroupMemberListByUserId(1L, boundarySendTime); + verify(groupMessageMapper).selectListByGroupIdAndMinIdAndQuitTimeBefore(eq(20L), eq(8L), + any(LocalDateTime.class), eq(quitTime), eq(2)); + verify(groupMessageMapper, never()).selectListByGroupIdAndMinIdAndQuitTimeBefore(eq(20L), eq(101L), + any(LocalDateTime.class), eq(quitTime), eq(2)); + } + + @Test public void testPullMessages_receiverUserIdsFilter() { // 准备 LocalDateTime now = LocalDateTime.now(); @@ -452,9 +553,7 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest { List members = List.of( selfMember, ImGroupMemberDO.builder().groupId(10L).userId(2L) - .status(CommonStatusEnum.ENABLE.getStatus()).build(), - ImGroupMemberDO.builder().groupId(10L).userId(3L) - .status(CommonStatusEnum.DISABLE.getStatus()).build() + .status(CommonStatusEnum.ENABLE.getStatus()).build() ); when(groupMemberService.getActiveGroupMemberListByGroupId(10L)).thenReturn(members); -- libgit2 0.21.4