Browse Source

perf: batch conversation unread counts

master
Zhang Shun 1 month ago
parent
commit
b56f792a3e
  1. 2
      docs/backend-development-roadmap.md
  2. 48
      docs/r01-d3-conversation-unread-query-report.md
  3. 31
      mydreamjob-server/src/main/java/com/mydreamjob/messaging/MessageRepository.java
  4. 5
      mydreamjob-server/src/main/java/com/mydreamjob/messaging/MessagingService.java
  5. 70
      mydreamjob-server/src/test/java/com/mydreamjob/messaging/MessageListQueryBudgetIntegrationTest.java

2
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节为准,移出一期 |

48
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 验收中复核执行计划与结果。

31
mydreamjob-server/src/main/java/com/mydreamjob/messaging/MessageRepository.java

@ -20,6 +20,37 @@ public interface MessageRepository extends JpaRepository<Message, Long> {
List<Message> findByConversationIdInOrderByConversationIdAscCreatedAtDesc(List<Long> 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<ConversationUnreadCount> countUnreadByConversationIds(
@Param("conversationIds") List<Long> conversationIds,
@Param("readerUserId") Long readerUserId);
/**
* 条件更新未命中后的当前读,用于区分并发请求已完成撤回与真实冲突。
* PESSIMISTIC_WRITE 强制读取提交后的最新行,避免 MySQL REPEATABLE READ 使用旧快照。

5
mydreamjob-server/src/main/java/com/mydreamjob/messaging/MessagingService.java

@ -71,11 +71,14 @@ public class MessagingService {
if (memberships.isEmpty()) return List.of();
List<Long> convIds = memberships.stream().map(ConversationMember::getConversationId).toList();
List<Conversation> convs = conversationRepository.findAllById(convIds);
Map<Long, Long> 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<Map<String, Object>> result = new ArrayList<>();
for (Conversation c : convs) {
Map<String, Object> v = toView(c, userId);
v.put("unreadCount", countUnread(c.getId(), userId));
v.put("unreadCount", unreadByConversation.getOrDefault(c.getId(), 0L));
result.add(v);
}
return result;

70
mydreamjob-server/src/test/java/com/mydreamjob/messaging/MessageListQueryBudgetIntegrationTest.java

@ -120,6 +120,76 @@ class MessageListQueryBudgetIntegrationTest {
+ measuredStatements);
}
@Test
void listConversationsUsesFixedQueriesForOneTwentyAndFiftyConversations() {
List<Integer> 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<java.util.Map<String, Object>> 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<Message> 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<Message> 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<java.util.Map<String, Object>> 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))

Loading…
Cancel
Save