初始化项目

云

2026-06-30 24681c81c09022f584a57006f2534b5f74723414

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
package cn.iocoder.yudao.module.im.mq.consumer.friend;
 
import cn.hutool.core.collection.CollUtil;
import cn.iocoder.yudao.module.im.dal.dataobject.friend.ImFriendDO;
import cn.iocoder.yudao.module.im.enums.ImContentTypeEnum;
import cn.iocoder.yudao.module.im.enums.ImConversationTypeEnum;
import cn.iocoder.yudao.module.im.service.friend.ImFriendService;
import cn.iocoder.yudao.module.im.service.websocket.ImWebSocketService;
import cn.iocoder.yudao.module.im.service.websocket.notification.friend.FriendInfoUpdatedNotification;
import cn.iocoder.yudao.module.system.api.message.user.AdminUserProfileUpdateMessage;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import org.springframework.transaction.event.TransactionPhase;
import org.springframework.transaction.event.TransactionalEventListener;
 
import java.util.List;
 
/**
 * 监听 system 模块的 {@link AdminUserProfileUpdateMessage} 消息，向「资料被改的人」的所有好友推送 FRIEND_INFO_UPDATED 通知
 *
 * @author 芋道源码
 */
@Slf4j
@Component
public class AdminUserProfileUpdateConsumer {
 
    @Resource
    private ImFriendService friendService;
    @Resource
    private ImWebSocketService websocketService;
 
    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT, fallbackExecution = true)
    @Async // Spring Event 默认在 Producer 发送的线程，通过 @Async 实现异步；事务提交后触发，避免回滚误推幽灵通知 / Consumer 抢在 commit 前读旧值
    public void onMessage(AdminUserProfileUpdateMessage message) {
        try {
            log.info("[onMessage][消息内容({})]", message);
            // 1. 过滤双向有效好友
            if (message == null || message.getUserId() == null) {
                return;
            }
            Long userId = message.getUserId();
            List<ImFriendDO> friends = friendService.getMutualEnableFriendList(userId);
            if (CollUtil.isEmpty(friends)) {
                return;
            }
 
            // 2. 给每个好友的多端推 FRIEND_INFO_UPDATED；payload 里 operatorUserId / friendUserId 都是「资料被改的人」
            int successCount = 0;
            for (ImFriendDO friend : friends) {
                try {
                    FriendInfoUpdatedNotification payload = (FriendInfoUpdatedNotification) new FriendInfoUpdatedNotification()
                            .setOperatorUserId(userId).setFriendUserId(userId);
                    websocketService.sendNotificationAsync(friend.getFriendUserId(),
                            ImConversationTypeEnum.NONE.getType(),
                            ImContentTypeEnum.FRIEND_INFO_UPDATED.getType(), payload);
                    successCount++;
                } catch (Exception e) {
                    log.warn("[onMessage][userId({}) friendUserId({}) 推送失败]",
                            userId, friend.getFriendUserId(), e);
                }
            }
            log.info("[onMessage][userId({}) 推送 FRIEND_INFO_UPDATED 给 {} 位好友]", userId, successCount);
        } catch (Exception e) {
            log.error("[onMessage][消息内容({}) 处理失败]", message, e);
        }
    }
 
}