#腾讯即时通讯架构

// QQ 中转消息：最早期版本
public class QQMessageRelay {

    // 伪代码：服务器端消息中转逻辑
    public void relayMessage(Long senderUin, Long receiverUin, String content) {
        // 1. 检查接收方是否在线
        SessionInfo receiver = sessionManager.getSession(receiverUin);

        if (receiver != null && receiver.isOnline()) {
            // 接收方在线：直接推送
            PushService.push(receiver, content);
        } else {
            // 接收方不在线：存储到离线消息队列
            OfflineMessageStore.store(receiverUin, new Message(
                senderUin, content, System.currentTimeMillis()));
        }
    }
}

// 消息可靠送达：ACK 机制
public class MessageService {

    // 发送消息
    public SendResult sendMessage(Long senderUin, Long receiverUin, String content) {
        // 1. 生成全局唯一消息 ID
        String msgId = UUIDGenerator.next();

        // 2. 构造消息
        IMMessage message = new IMMessage();
        message.setMsgId(msgId);
        message.setSenderUin(senderUin);
        message.setReceiverUin(receiverUin);
        message.setContent(content);
        message.setTimestamp(System.currentTimeMillis());
        message.setStatus(MessageStatus.SENDING);

        // 3. 持久化消息（消息必须落盘才能算发送成功）
        messageStore.save(message);

        // 4. 尝试推送
        boolean pushed = pushService.push(receiverUin, message);

        if (!pushed) {
            // 接收方不在线：标记为待送达
            message.setStatus(MessageStatus.PENDING_DELIVERY);
            messageStore.update(message);
        }

        return new SendResult(msgId, pushed ? DeliveryStatus.DELIVERED : DeliveryStatus.PENDING);
    }

    // 接收方 ACK
    public void onMessageAck(Long msgId, Long receiverUin) {
        IMMessage message = messageStore.findByMsgId(msgId);
        message.setStatus(MessageStatus.DELIVERED);
        messageStore.update(message);

        // 通知发送方：消息已送达
        PushService.pushToClient(message.getSenderUin(),
            new DeliveryReceipt(msgId, receiverUin));
    }
}

// 消息存储：按接收者分片
public class MessageStore {

    // 分片策略：receiverUin % 分片数
    private int getShard(Long receiverUin) {
        return (int) (receiverUin % shardCount);
    }

    // 存储消息
    public void saveMessage(Long senderUin, Long receiverUin,
                           String content, long timestamp) {
        int shard = getShard(receiverUin);

        // Wide-Column 存储：
        // Row Key: receiverUin + "_" + timestamp
        // Column: senderUin -> content
        String rowKey = receiverUin + "_" + timestamp;

        wideColumnStore.put(shard, rowKey, senderUin.toString(), content);

        // 同时更新索引：按 sender 维度查询
        String indexRowKey = senderUin + "_" + receiverUin + "_" + timestamp;
        wideColumnStore.put(shard, indexRowKey, "msg", rowKey);
    }

    // 查询某会话的消息（分页）
    public List<IMMessage> getConversation(Long userA, Long userB,
                                         long beforeTimestamp, int limit) {
        // 1. 查询 A->B 的消息
        String startKeyA = userA + "_" + userB + "_" + beforeTimestamp;
        List<MessageEntry> messagesA = wideColumnStore.scan(
            getShard(userB), startKeyA, limit / 2);

        // 2. 查询 B->A 的消息
        String startKeyB = userB + "_" + userA + "_" + beforeTimestamp;
        List<MessageEntry> messagesB = wideColumnStore.scan(
            getShard(userA), startKeyB, limit / 2);

        // 3. 合并排序
        return mergeAndSort(messagesA, messagesB, limit);
    }
}

// WebSocket 心跳机制
public class HeartbeatService {

    private static final int HEARTBEAT_INTERVAL_SECONDS = 30;
    private static final int HEARTBEAT_TIMEOUT_SECONDS = 60;

    // 客户端：定时发送心跳
    @Scheduled(fixedRate = HEARTBEAT_INTERVAL_SECONDS * 1000)
    public void sendHeartbeat() {
        WebSocketSession session = getCurrentSession();
        if (session.isOpen()) {
            session.sendMessage(new HeartbeatMessage(
                getCurrentUserId(), System.currentTimeMillis()));
        }
    }

    // 服务器：检测心跳超时
    public void onHeartbeat(Long userId, long clientTimestamp) {
        connectionManager.updateHeartbeat(userId, clientTimestamp);
    }

    @Scheduled(fixedRate = HEARTBEAT_TIMEOUT_SECONDS * 1000)
    public void checkTimeout() {
        long threshold = System.currentTimeMillis()
            - HEARTBEAT_TIMEOUT_SECONDS * 1000L;

        // 找出心跳超时的连接
        for (Long userId : connectionManager.getOnlineUsers()) {
            if (connectionManager.getLastHeartbeat(userId) < threshold) {
                // 标记为离线，清理连接
                connectionManager.markOffline(userId);
                // 通知业务系统
                eventBus.publish(new UserOfflineEvent(userId));
            }
        }
    }
}

// 多端同步：消息漫游
public class MultiDeviceSyncService {

    // 用户设备列表
    // deviceType: MOBILE / PC / PAD / WEB
    public List<DeviceSession> getUserDevices(Long userId) {
        return deviceRepository.findByUserId(userId);
    }

    // 发送消息时，同步到所有在线设备
    public void syncMessageToDevices(Long userId, IMMessage message) {
        List<DeviceSession> devices = getUserDevices(userId);

        for (DeviceSession device : devices) {
            if (device.isOnline()) {
                // 推送到在线设备
                pushService.pushToDevice(device, message);
            } else {
                // 离线设备：消息漫游到服务器
                // 用户下次登录该设备时，拉取漫游消息
                offlineMessageStore.storeForDevice(
                    userId, device.getDeviceId(), message);
            }
        }
    }

    // 设备登录：拉取漫游消息
    public List<IMMessage> pullRoamingMessages(Long userId,
                                               String deviceId,
                                               long sinceTimestamp) {
        // 从离线消息存储中拉取该设备未同步的消息
        return offlineMessageStore.getForDevice(
            userId, deviceId, sinceTimestamp);
    }
}

// 就近接入：DNS 解析 + Anycast
public class AccessRouter {

    // 用户发起请求时，DNS 返回最近的接入机房 IP
    public InetSocketAddress resolveAccessPoint(Long userId) {
        // 1. 获取用户地理位置（通过 IP 库或 GPS）
        GeoLocation location = geoService.locateUser(userId);

        // 2. 根据地理位置选择最近的接入机房
        // 优先同机房 > 同城 > 同区域 > 跨区域
        AccessPoint ap = accessPointSelector.select(
            location, LoadMetric.RESPONSE_TIME);

        return new InetSocketAddress(ap.getPublicIP(), ap.getPort());
    }

    // 跨机房消息路由
    public void routeMessage(IMMessage message) {
        Long receiverId = message.getReceiverUin();

        if (isLocalUser(receiverId)) {
            // 本机房用户：直接投递
            deliverLocally(message);
        } else {
            // 跨机房用户：通过专线投递
            String targetDc = getUserDataCenter(receiverId);
            dcRouter.sendToDataCenter(targetDc, message);
        }
    }
}

// Snowflake 消息 ID 生成
public class SnowflakeIdGenerator {

    private static final long EPOCH = 1609459200000L; // 2021-01-01
    private static final long WORKER_ID_BITS = 10L;
    private static final long SEQUENCE_BITS = 12L;

    private final long workerId;
    private long sequence = 0L;
    private long lastTimestamp = -1L;

    public long nextId() {
        long timestamp = System.currentTimeMillis() - EPOCH;

        if (timestamp == lastTimestamp) {
            sequence = (sequence + 1) & ((1 << SEQUENCE_BITS) - 1);
            if (sequence == 0) timestamp = waitForNextMillis();
        } else {
            sequence = 0;
        }

        lastTimestamp = timestamp;

        return (timestamp << (WORKER_ID_BITS + SEQUENCE_BITS))
             | (workerId << SEQUENCE_BITS)
             | sequence;
    }
}