From b56f792a3e5a907287e671864c7e631a4320ad48 Mon Sep 17 00:00:00 2001 From: Zhang Shun Date: Wed, 16 Sep 2026 09:52:11 +0800 Subject: [PATCH] perf: batch conversation unread counts --- docs/backend-development-roadmap.md | 2 +- ...r01-d3-conversation-unread-query-report.md | 48 +++++++++++++ .../messaging/MessageRepository.java | 31 ++++++++ .../messaging/MessagingService.java | 5 +- ...MessageListQueryBudgetIntegrationTest.java | 70 +++++++++++++++++++ 5 files changed, 154 insertions(+), 2 deletions(-) create mode 100644 docs/r01-d3-conversation-unread-query-report.md diff --git a/docs/backend-development-roadmap.md b/docs/backend-development-roadmap.md index 5d570f9..5107b03 100644 --- a/docs/backend-development-roadmap.md +++ b/docs/backend-development-roadmap.md @@ -15,7 +15,7 @@ | S02 Access/Refresh Token 隔离 | 已实现并完成自动化验证与干净构建，待独立代码复验及真实 MySQL/Docker 黑盒验收 | 类型隔离、独立密钥、轮换、重放处置、会话吊销已锁定；OpenAPI 175/213 | | S03 生产配置与简历隐私 | 已实现并完成自动化验证与干净构建，待独立代码复验及真实 MySQL/Docker 黑盒验收 | V43 禁用原始指纹种子账号；管理员初始化默认关闭且幂等；企业简历下载按隐私决策表授权；Actuator 隔离到管理端口；OpenAPI 176/214 | | Q01 岗位搜索真实查询覆盖（C02 证据补充） | 已完成实现、跨库一致性整改、独立复验及 Git 检查点（`77dc624`）。新增 54 条真实库测试：H2 Repository 21、H2 Service 9、H2 副作用 4、MySQL 8.4 Specification 15、MySQL 副作用 4、MySQL EXPLAIN 1；全量 1890/1890 | OpenAPI 保持 176/214；前导通配符、filesort 与真实规模执行计划移交 R01 | -| R01 结构与性能收束 | R01-A `0b5b849`、R01-B `15bd98b`、R01-C `b68519e`、R01-D1 `cf0c71e` 已建本地检查点；R01-D2 已完成个人数据导出 N+1 收束，待独立复验 | overview 与偏好口径已集中；百分位 O(n log n)；数据导出子集合改为固定批量查询；后续处理会话未读数和其余性能项 | +| R01 结构与性能收束 | R01-A `0b5b849`、R01-B `15bd98b`、R01-C `b68519e`、R01-D1 `cf0c71e`、R01-D2 `8c40fd4` 已建本地检查点；R01-D3 已完成会话列表未读数 N+1 收束，待独立复验 | overview 与偏好口径已集中；百分位 O(n log n)；数据导出与会话未读数均改为固定批量查询；后续处理其余性能项 | | 其余五个功能组 | 技能词云、入职清单/公司知识、薪资计算器/参考、诚信积分/激励、简历查重 | N03–N06、N08仍属一期，细节设计后编码 | | 上线保障 | 现有资产待逐项核验 | G01–G04，一期交付门槛 | | 后续候选池 | AI简历、AI模拟面试/预测、SHAP高级解释等 | 以清单第6节为准，移出一期 | diff --git a/docs/r01-d3-conversation-unread-query-report.md b/docs/r01-d3-conversation-unread-query-report.md new file mode 100644 index 0000000..80f5344 --- /dev/null +++ b/docs/r01-d3-conversation-unread-query-report.md @@ -0,0 +1,48 @@ +# R01-D3 会话未读数查询收束报告 + +## 结论 + +`MessagingService.listConversations` 已从逐会话查询最近消息和已读状态，收束为一次分组未读统计。会话数量从 1 增长到 20、50 时，消息相关业务 SQL 数保持固定；本次未改变公网接口、响应字段、权限、消息撤回或已读语义。 + +## 修复前问题 + +会话列表先查询成员关系和会话，再对每个会话调用 `countUnread`。每次调用都会查询最近 500 条消息及其已读记录，因此总查询数随会话数线性增长。 + +## 实现 + +- `MessageRepository.countUnreadByConversationIds` 使用一条分组 SQL 批量统计全部会话。 +- 使用 `ROW_NUMBER() OVER (PARTITION BY conversation_id ORDER BY created_at DESC)` 保留每个会话“最近 500 条”的历史窗口。 +- `RECALLED` 消息不计入未读，`status IS NULL` 仍按历史语义计入。 +- 存在当前用户 `message_reads` 的消息不计入未读。 +- 没有未读行的会话由 Service 映射为 `0`。 +- `getConversation` 的单会话路径未改动，避免扩大本批范围。 + +## 查询预算证据 + +`MessageListQueryBudgetIntegrationTest.listConversationsUsesFixedQueriesForOneTwentyAndFiftyConversations` 使用真实 Spring/JPA/H2 数据源代理统计业务 SQL，并混合 DIRECT、TALENT_INVITATION、SENT、RECALLED、已读和未读消息。 + +| 会话数 | 业务查询数 | 结果 | +| ---: | ---: | --- | +| 1 | 3 | PASS | +| 20 | 3 | PASS | +| 50 | 3 | PASS | + +三条查询分别为成员关系、会话批量读取、未读分组统计，查询数不随会话数增长。 + +## 语义边界 + +`groupedUnreadCountKeepsHistoricalLatestFiveHundredWindow` 构造 501 条顺序明确的消息，证明最旧消息位于窗口外时不被统计。测试首次编写时，实体的 `@CreatedDate` 覆盖夹具传入时间，造成 501 条消息时间并列；随后改为保存后按精确消息主键写入确定时间。该失败属于夹具问题，不是生产缺陷，也未通过放宽断言处理。 + +## 验证 + +- 定向边界测试：3/3，Failures=0、Errors=0、Skipped=0。 +- 关联消息测试：此前同批运行 42/42 通过。 +- `mvn clean verify`：2104 项，Failures=0、Errors=0、Skipped=20，BUILD SUCCESS；20 项均为既有环境条件测试。 +- 静态 OpenAPI：176 paths / 214 operations，校验通过。 + +## 边界 + +- 未新增或修改 Flyway、表、索引、ErrorCode、公网接口。 +- 未改变 DIRECT / TALENT_INVITATION 响应隐私语义。 +- 未改变消息发送、撤回、已读、分页和权限语义。 +- 原生窗口 SQL 仍需在独立真实 MySQL 验收中复核执行计划与结果。 diff --git a/mydreamjob-server/src/main/java/com/mydreamjob/messaging/MessageRepository.java b/mydreamjob-server/src/main/java/com/mydreamjob/messaging/MessageRepository.java index d67e3af..814d29d 100644 --- a/mydreamjob-server/src/main/java/com/mydreamjob/messaging/MessageRepository.java +++ b/mydreamjob-server/src/main/java/com/mydreamjob/messaging/MessageRepository.java @@ -20,6 +20,37 @@ public interface MessageRepository extends JpaRepository { List findByConversationIdInOrderByConversationIdAscCreatedAtDesc(List conversationIds); + interface ConversationUnreadCount { + Long getConversationId(); + Long getUnreadCount(); + } + + /** + * 批量统计各会话最近 500 条消息中的未读数，保持历史 countUnread 的窗口与撤回语义。 + */ + @Query(value = """ + SELECT ranked.conversation_id AS conversationId, COUNT(*) AS unreadCount + FROM ( + SELECT m.id, m.conversation_id, m.status, + ROW_NUMBER() OVER ( + PARTITION BY m.conversation_id + ORDER BY m.created_at DESC + ) AS row_no + FROM messages m + WHERE m.conversation_id IN (:conversationIds) + ) ranked + LEFT JOIN message_reads mr + ON mr.message_id = ranked.id + AND mr.reader_user_id = :readerUserId + WHERE ranked.row_no <= 500 + AND (ranked.status IS NULL OR ranked.status <> 'RECALLED') + AND mr.id IS NULL + GROUP BY ranked.conversation_id + """, nativeQuery = true) + List countUnreadByConversationIds( + @Param("conversationIds") List conversationIds, + @Param("readerUserId") Long readerUserId); + /** * 条件更新未命中后的当前读，用于区分并发请求已完成撤回与真实冲突。 * PESSIMISTIC_WRITE 强制读取提交后的最新行，避免 MySQL REPEATABLE READ 使用旧快照。 diff --git a/mydreamjob-server/src/main/java/com/mydreamjob/messaging/MessagingService.java b/mydreamjob-server/src/main/java/com/mydreamjob/messaging/MessagingService.java index a4164d4..7da0242 100644 --- a/mydreamjob-server/src/main/java/com/mydreamjob/messaging/MessagingService.java +++ b/mydreamjob-server/src/main/java/com/mydreamjob/messaging/MessagingService.java @@ -71,11 +71,14 @@ public class MessagingService { if (memberships.isEmpty()) return List.of(); List convIds = memberships.stream().map(ConversationMember::getConversationId).toList(); List convs = conversationRepository.findAllById(convIds); + Map unreadByConversation = new HashMap<>(); + messageRepository.countUnreadByConversationIds(convIds, userId).forEach(row -> + unreadByConversation.put(row.getConversationId(), row.getUnreadCount())); convs.sort((a, b) -> b.getUpdatedAt().compareTo(a.getUpdatedAt())); List> result = new ArrayList<>(); for (Conversation c : convs) { Map v = toView(c, userId); - v.put("unreadCount", countUnread(c.getId(), userId)); + v.put("unreadCount", unreadByConversation.getOrDefault(c.getId(), 0L)); result.add(v); } return result; diff --git a/mydreamjob-server/src/test/java/com/mydreamjob/messaging/MessageListQueryBudgetIntegrationTest.java b/mydreamjob-server/src/test/java/com/mydreamjob/messaging/MessageListQueryBudgetIntegrationTest.java index b8c20ea..4cbb58d 100644 --- a/mydreamjob-server/src/test/java/com/mydreamjob/messaging/MessageListQueryBudgetIntegrationTest.java +++ b/mydreamjob-server/src/test/java/com/mydreamjob/messaging/MessageListQueryBudgetIntegrationTest.java @@ -120,6 +120,76 @@ class MessageListQueryBudgetIntegrationTest { + measuredStatements); } + @Test + void listConversationsUsesFixedQueriesForOneTwentyAndFiftyConversations() { + List measuredStatements = new ArrayList<>(); + for (int count : List.of(1, 20, 50)) { + long reader = user("conversation-reader-" + count); + for (int i = 0; i < count; i++) { + long sender = user("conversation-sender-" + count + "-" + i); + long conversation = conversation(sender, reader, i % 2 == 0 ? "DIRECT" : "TALENT_INVITATION"); + Message unread = messageRepository.save(Message.builder() + .conversationId(conversation).senderUserId(sender).content("unread") + .status("SENT").version(0L).build()); + Message read = messageRepository.save(Message.builder() + .conversationId(conversation).senderUserId(sender).content("read") + .status("SENT").version(0L).build()); + messageRepository.save(Message.builder() + .conversationId(conversation).senderUserId(sender) + .content(MessageViewMapper.RECALLED_PLACEHOLDER) + .status("RECALLED").version(1L).recalledAt(LocalDateTime.now()).build()); + readRepository.save(MessageRead.builder().messageId(read.getId()).readerUserId(reader).build()); + assertTrue(unread.getId() > 0); + } + messageRepository.flush(); + readRepository.flush(); + + countingDataSource.reset(); + List> response = messagingService.listConversations(reader); + int statements = countingDataSource.messageStatements(); + measuredStatements.add(statements); + + assertEquals(count, response.size()); + assertTrue(response.stream().allMatch(item -> Long.valueOf(1L).equals(item.get("unreadCount"))), + "only the unread SENT message must be counted"); + assertTrue(statements <= 3, + "membership, conversations and grouped unread count must be the only query budget: " + + statements); + } + assertEquals(1, measuredStatements.stream().distinct().count(), + "conversation query count must not grow with conversation count: " + measuredStatements); + } + + @Test + void groupedUnreadCountKeepsHistoricalLatestFiveHundredWindow() { + long reader = user("window-reader"); + long sender = user("window-sender"); + long conversation = conversation(sender, reader, "DIRECT"); + LocalDateTime base = LocalDateTime.of(2026, 9, 16, 8, 0); + List messages = new ArrayList<>(); + for (int i = 0; i < 501; i++) { + boolean countedSent = i == 0 || i == 500; + messages.add(Message.builder() + .conversationId(conversation).senderUserId(sender) + .content(countedSent ? "sent" : MessageViewMapper.RECALLED_PLACEHOLDER) + .status(countedSent ? "SENT" : "RECALLED") + .version(countedSent ? 0L : 1L) + .recalledAt(countedSent ? null : base.plusSeconds(i)) + .createdAt(base.plusSeconds(i)).build()); + } + List saved = messageRepository.saveAllAndFlush(messages); + for (int i = 0; i < saved.size(); i++) { + jdbcTemplate.update("UPDATE messages SET created_at=? WHERE id=?", + java.sql.Timestamp.valueOf(base.plusSeconds(i)), saved.get(i).getId()); + } + + List> response = messagingService.listConversations(reader); + + assertEquals(1, response.size()); + assertEquals(1L, response.get(0).get("unreadCount"), + "oldest SENT message is outside the historical latest-500 window"); + } + private long user(String nickname) { User saved = userRepository.saveAndFlush(User.builder() .phone("139" + Math.abs(System.nanoTime() % 1_000_000_000L))