BJ_ZTEITS-Developers / parking-cloud-yudao

Commit e4a6cc932c7ba94806f8a324cb3dfabb145e9a9e

Authored by YunaiV
1 parent 736a1a2f

✨ feat(websocket): 支持事务感知的异步消息推送 

实现了在事务中推送 WebSocket 消息的功能，确保在事务提交后再发送消息，避免客户端在查询数据库时看到未提交的变更。新增了 executeAfterTransaction 方法来处理事务同步，优化了私聊和群聊消息的发送逻辑。
Inline Side-by-side
Showing 1 changed file with 59 additions and 5 deletions
yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/service/websocket/ImWebSocketServiceImpl.java
  View file @e4a6cc9
1 1 package cn.iocoder.yudao.module.im.service.websocket;
2 2  
  3 +import cn.hutool.extra.spring.SpringUtil;
3 4 import cn.iocoder.yudao.framework.common.enums.UserTypeEnum;
4 5 import cn.iocoder.yudao.framework.websocket.core.sender.WebSocketMessageSender;
5 6 import cn.iocoder.yudao.module.im.service.websocket.dto.ImGroupMessageDTO;
... ... @@ -8,11 +9,17 @@ import jakarta.annotation.Resource;
8 9 import lombok.extern.slf4j.Slf4j;
9 10 import org.springframework.scheduling.annotation.Async;
10 11 import org.springframework.stereotype.Service;
  12 +import org.springframework.transaction.support.TransactionSynchronization;
  13 +import org.springframework.transaction.support.TransactionSynchronizationManager;
11 14  
12 15 import java.util.Collection;
13 16  
  17 +// TODO @芋艿：后续要分析下，是不是要拆分下 notify（类似好友、群的变更）
14 18 /**
15 19 * IM WebSocket 推送 Service 实现类
  20 + * <p>
  21 + * 当调用方处于事务中时，推送会延迟到事务提交后再异步执行，
  22 + * 避免客户端收到 WebSocket 消息时数据库变更尚未可见。
16 23 *
17 24 * @author 芋道源码
18 25 */
... ... @@ -23,28 +30,75 @@ public class ImWebSocketServiceImpl implements ImWebSocketService {
23 30 @Resource
24 31 private WebSocketMessageSender webSocketMessageSender;
25 32  
26   - @Async
27 33 @Override
28 34 public void sendPrivateMessageAsync(Long userId, ImPrivateMessageDTO dto) {
  35 + // 说明：通过 executeAfterCommitOrNow 保证事务提交后再推送，避免客户端收到消息后查询数据库时事务尚未提交
  36 + // 通过 getSelf() 获取 Spring 代理对象调用 @Async 方法，确保异步 AOP 生效（直接 this 调用会绕过代理）
  37 + executeAfterTransaction(() -> getSelf().doSendPrivateMessage(userId, dto));
  38 + }
  39 +
  40 + @Override
  41 + public void sendGroupMessageAsync(Collection<Long> userIds, ImGroupMessageDTO dto) {
  42 + executeAfterTransaction(() -> getSelf().doSendGroupMessage(userIds, dto));
  43 + }
  44 +
  45 + /**
  46 + * 异步发送私聊 WebSocket 消息
  47 + */
  48 + @Async
  49 + public void doSendPrivateMessage(Long userId, ImPrivateMessageDTO dto) {
29 50 try {
30 51 webSocketMessageSender.sendObject(UserTypeEnum.ADMIN.getValue(), userId,
31 52 ImPrivateMessageDTO.TYPE, dto);
32 53 } catch (Exception e) {
33   - log.error("[sendPrivateMessageAsync][userId({}) dto({}) 发送失败]", userId, dto, e);
  54 + log.error("[doSendPrivateMessage][userId({}) dto({}) 发送失败]", userId, dto, e);
34 55 }
35 56 }
36 57  
  58 + /**
  59 + * 异步发送群聊 WebSocket 消息
  60 + */
37 61 @Async
38   - @Override
39   - public void sendGroupMessageAsync(Collection<Long> userIds, ImGroupMessageDTO dto) {
  62 + public void doSendGroupMessage(Collection<Long> userIds, ImGroupMessageDTO dto) {
40 63 for (Long userId : userIds) {
41 64 try {
42 65 webSocketMessageSender.sendObject(UserTypeEnum.ADMIN.getValue(), userId,
43 66 ImGroupMessageDTO.TYPE, dto);
44 67 } catch (Exception e) {
45   - log.error("[sendGroupMessageAsync][userId({}) dto({}) 发送失败]", userId, dto, e);
  68 + log.error("[doSendGroupMessage][userId({}) dto({}) 发送失败]", userId, dto, e);
46 69 }
47 70 }
48 71 }
49 72  
  73 + /**
  74 + * 事务感知的任务调度：
  75 + * - 有事务：注册 afterCommit 回调，事务提交后再执行，防止客户端拿到消息去查库时数据还没落盘
  76 + * - 无事务：直接执行（如非 @Transactional 方法中的调用）
  77 + *
  78 + * @param task 待执行的推送任务（内部通过 getSelf() 走 @Async 异步执行）
  79 + */
  80 + private void executeAfterTransaction(Runnable task) {
  81 + // 情况一：没有事务，直接执行
  82 + if (!TransactionSynchronizationManager.isSynchronizationActive()) {
  83 + task.run();
  84 + return;
  85 + }
  86 + // 情况二：有事务，注册 afterCommit 事件，在事务提交后执行
  87 + TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
  88 +
  89 + @Override
  90 + public void afterCommit() {
  91 + task.run();
  92 + }
  93 +
  94 + });
  95 + }
  96 +
  97 + /**
  98 + * 获得自身的代理对象，解决 @Async AOP 代理问题
  99 + */
  100 + private ImWebSocketServiceImpl getSelf() {
  101 + return SpringUtil.getBean(getClass());
  102 + }
  103 +
50 104 }
... ...