/**
 * IM消息协议包
 * 所有WebSocket消息都用这个格式（JSON序列化）
 */
@Data
@Builder
public class ImMessage {
    
    private String msgId;       // 客户端生成的唯一ID（UUID），用于去重
    private Long msgSeq;        // 服务端分配的序列号，用于排序和ACK
    private Integer msgType;    // 消息类型（见MessageType枚举）
    private String fromUserId;  // 发送者ID
    private String toId;        // 接收者ID（单聊=用户ID，群聊=群ID）
    private String sessionType; // 会话类型："single"/"group"
    private Object content;     // 消息内容（根据msgType不同结构不同）
    private Long timestamp;     // 发送时间戳（毫秒）
    private String ackId;       // ACK时对应的msgSeq（ACK类型消息使用）
    
    public enum MessageType {
        TEXT(1),         // 文本消息
        IMAGE(2),        // 图片消息
        FILE(3),         // 文件消息
        AUDIO(4),        // 语音消息
        HEARTBEAT(10),   // 心跳
        ACK(11),         // 消息确认
        READ_RECEIPT(12),// 已读回执
        SYSTEM(20);      // 系统消息
        
        private final int code;
        MessageType(int code) { this.code = code; }
    }
}

三、系统架构设计

四、消息可靠性保障（核心难题）

消息可靠性是IM系统最难的部分，要处理三个场景：

消息发出后，服务端没收到（网络丢包）
服务端收到后，推送给接收方时，接收方掉线了（消息丢失）
消息发出后，网络不稳定，重发导致重复（消息重复）

解决方案：客户端重传 + 服务端去重 + ACK机制

五、关键代码实现

5.1 WebSocket消息处理器

@Component
@Slf4j
public class ImWebSocketHandler extends TextWebSocketHandler {

    @Autowired
    private ImMessageService messageService;
    
    @Autowired
    private ConnectionManager connectionManager;
    
    @Autowired
    private MessageDeduplicator deduplicator;

    @Override
    public void afterConnectionEstablished(WebSocketSession session) {
        String userId = extractUserId(session);
        connectionManager.onConnect(userId, session);
        
        // 连接成功后，拉取离线期间的消息
        pullOfflineMessages(userId, session);
    }

    @Override
    protected void handleTextMessage(
            WebSocketSession session, TextMessage message) {
        try {
            ImMessage imMsg = JsonUtils.fromJson(
                message.getPayload(), ImMessage.class);
            
            String userId = extractUserId(session);
            imMsg.setFromUserId(userId);
            imMsg.setTimestamp(System.currentTimeMillis());
            
            switch (imMsg.getMsgType()) {
                case 10: // 心跳
                    handleHeartbeat(userId, session);
                    break;
                case 11: // ACK
                    handleAck(userId, imMsg);
                    break;
                case 12: // 已读回执
                    handleReadReceipt(userId, imMsg);
                    break;
                default: // 业务消息
                    handleBusinessMessage(userId, imMsg, session);
            }
        } catch (Exception e) {
            log.error("消息处理异常, session={}", session.getId(), e);
        }
    }

    private void handleBusinessMessage(
            String fromUserId, ImMessage msg, WebSocketSession session) {
        
        // 1. 服务端去重：同一个msgId只处理一次
        if (!deduplicator.isNew(msg.getMsgId())) {
            log.debug("重复消息，忽略。msgId={}", msg.getMsgId());
            // 即使重复也要回复ACK，告知客户端已收到
            sendAck(session, msg.getMsgId(), 0L);
            return;
        }
        
        // 2. 存储消息并分配序列号
        Long msgSeq = messageService.saveAndRoute(msg);
        
        // 3. 回复发送者ACK（告知消息已被服务端接收，附带服务端序列号）
        sendAck(session, msg.getMsgId(), msgSeq);
    }
    
    private void sendAck(WebSocketSession session, String msgId, Long msgSeq) {
        ImMessage ack = ImMessage.builder()
            .msgType(ImMessage.MessageType.ACK.getCode())
            .ackId(msgId)
            .msgSeq(msgSeq)
            .timestamp(System.currentTimeMillis())
            .build();
        try {
            session.sendMessage(new TextMessage(JsonUtils.toJson(ack)));
        } catch (IOException e) {
            log.warn("ACK发送失败, msgId={}", msgId);
        }
    }
    
    private void pullOfflineMessages(String userId, WebSocketSession session) {
        // 用户上线时，推送离线期间的消息
        Long lastReceivedSeq = getLastReceivedSeq(userId);
        List<ImMessage> offlineMessages = messageService.getOfflineMessages(
            userId, lastReceivedSeq);
        
        offlineMessages.forEach(msg -> {
            try {
                session.sendMessage(new TextMessage(JsonUtils.toJson(msg)));
            } catch (IOException e) {
                log.warn("离线消息推送失败, userId={}, msgSeq={}", 
                    userId, msg.getMsgSeq());
            }
        });
    }
}

5.2 消息存储与路由服务

@Service
@Slf4j
public class ImMessageService {

    @Autowired
    private MessageMapper messageMapper;
    
    @Autowired
    private SessionMapper sessionMapper;
    
    @Autowired
    private ConnectionManager connectionManager;
    
    @Autowired
    private StringRedisTemplate redisTemplate;
    
    @Autowired
    private KafkaTemplate<String, ImMessage> kafkaTemplate;

    /**
     * 保存消息并路由推送
     * @return 消息序列号
     */
    @Transactional
    public Long saveAndRoute(ImMessage msg) {
        // 1. 生成消息序列号（全局单调递增）
        Long msgSeq = generateMsgSeq(msg.getToId());
        msg.setMsgSeq(msgSeq);
        
        // 2. 持久化消息到MySQL
        messageMapper.insert(buildMessageEntity(msg));
        
        // 3. 更新会话信息（最新消息、未读数）
        updateSession(msg);
        
        // 4. 确定接收者列表
        List<String> receivers = getReceivers(msg);
        
        // 5. 路由推送消息
        for (String receiverId : receivers) {
            if (!receiverId.equals(msg.getFromUserId())) {
                routeAndPush(receiverId, msg);
            }
        }
        
        return msgSeq;
    }
    
    /**
     * 消息路由：在线直推，离线存消息盒子并走APNs/FCM
     */
    private void routeAndPush(String userId, ImMessage msg) {
        // 查询用户在哪台WebSocket服务器上
        String onlineServerId = redisTemplate.opsForValue()
            .get("im:online:" + userId);
        
        if (onlineServerId == null) {
            // 用户离线：存消息盒子，发APNs推送
            kafkaTemplate.send("im-offline-push", userId, msg);
        } else if (onlineServerId.equals(getLocalServerId())) {
            // 用户在本机：直接推送
            connectionManager.pushToLocal(userId, JsonUtils.toJson(msg));
        } else {
            // 用户在其他服务器：通过Redis Pub/Sub转发
            String channel = "im:push:" + onlineServerId;
            ImPushCommand cmd = new ImPushCommand(userId, msg);
            redisTemplate.convertAndSend(channel, JsonUtils.toJson(cmd));
        }
    }
    
    /**
     * 生成消息序列号
     * 使用Redis自增保证单调递增（按会话维度）
     */
    private Long generateMsgSeq(String sessionId) {
        String key = "im:seq:" + sessionId;
        return redisTemplate.opsForValue().increment(key);
    }
    
    /**
     * 获取离线消息（用户上线时拉取）
     * @param userId 用户ID
     * @param lastSeq 客户端本地最大序列号
     */
    public List<ImMessage> getOfflineMessages(String userId, Long lastSeq) {
        if (lastSeq == null) lastSeq = 0L;
        return messageMapper.findAfterSeq(userId, lastSeq, 100); // 最多拉100条
    }
    
    private void updateSession(ImMessage msg) {
        // 更新发件人会话
        sessionMapper.upsertSession(msg.getFromUserId(), msg.getToId(), 
            msg.getSessionType(), msg);
        
        // 更新收件人会话（单聊：收件人=toId；群聊：所有群成员）
        List<String> receivers = getReceivers(msg);
        receivers.forEach(userId -> {
            if (!userId.equals(msg.getFromUserId())) {
                sessionMapper.upsertSession(userId, msg.getToId(), 
                    msg.getSessionType(), msg);
                // 未读数+1
                redisTemplate.opsForValue().increment(
                    "im:unread:" + userId + ":" + msg.getToId()
                );
            }
        });
    }
    
    private List<String> getReceivers(ImMessage msg) {
        if ("single".equals(msg.getSessionType())) {
            return Collections.singletonList(msg.getToId());
        } else {
            // 群聊：查询群成员列表
            return groupMemberMapper.findMemberIds(msg.getToId());
        }
    }
}

5.3 消息去重器

@Component
public class MessageDeduplicator {

    @Autowired
    private StringRedisTemplate redisTemplate;
    
    private static final long DEDUP_TTL_SECONDS = 300; // 5分钟去重窗口

    /**
     * 检查消息是否是新消息（首次接收）
     * 用Redis的SETNX实现，同一个msgId在5分钟内只处理一次
     */
    public boolean isNew(String msgId) {
        String key = "im:dedup:" + msgId;
        Boolean isNew = redisTemplate.opsForValue()
            .setIfAbsent(key, "1", DEDUP_TTL_SECONDS, TimeUnit.SECONDS);
        return Boolean.TRUE.equals(isNew);
    }
}