IMS系统WebSocket实时通信完全指南

/* WebSocket 服务 */
class WebSocketServer {
    constructor(server) {
        this.wss = new WebSocketServer({ server });
        this.clients = new Map();  // clientId -> WebSocket
        this.userConnections = new Map();  // userId -> Set

        this.wss.on('connection', (ws, req) => this.handleConnection(ws, req));
    }

    handleConnection(ws, req) {
        /* 解析 token 认证 */
        const token = parseToken(req.url);
        const user = verifyToken(token);

        if (!user) {
            ws.close(4001, 'Unauthorized');
            return;
        }

        /* 生成客户端 ID */
        const clientId = generateUUID();
        ws.clientId = clientId;
        ws.userId = user.id;

        /* 存储连接 */
        this.clients.set(clientId, ws);

        if (!this.userConnections.has(user.id)) {
            this.userConnections.set(user.id, new Set());
        }
        this.userConnections.get(user.id).add(clientId);

        /* 绑定事件 */
        ws.on('message', (msg) => this.handleMessage(ws, msg));
        ws.on('close', () => this.handleClose(ws));
        ws.on('error', (err) => this.handleError(ws, err));

        /* 发送连接成功消息 */
        this.send(ws, { type: 'connected', clientId });
    }

    handleClose(ws) {
        this.clients.delete(ws.clientId);

        const userConns = this.userConnections.get(ws.userId);
        if (userConns) {
            userConns.delete(ws.clientId);
            if (userConns.size === 0) {
                this.userConnections.delete(ws.userId);
            }
        }
    }
}

/* 心跳管理器 */
class HeartbeatManager {
    constructor(wsServer) {
        this.wsServer = wsServer;
        this.pingInterval = 30000;  // 30 秒 ping 一次
        this.pongTimeout = 10000;    // 10 秒未收到 pong 断开
        this.timers = new Map();  // clientId -> timer
    }

    start() {
        setInterval(() => this.pingAll(), this.pingInterval);
    }

    pingAll() {
        for (const [clientId, ws] of this.wsServer.clients) {
            if (ws.readyState === WebSocket.OPEN) {
                this.ping(ws, clientId);
            }
        }
    }

    ping(ws, clientId) {
        /* 发送 ping */
        ws.send(JSON.stringify({ type: 'ping', timestamp: Date.now() }));

        /* 设置超时 */
        const timer = setTimeout(() => {
            if (ws.readyState === WebSocket.OPEN) {
                console.log(`Client ${clientId} pong timeout, closing`);
                ws.close(4002, 'Pong timeout');
            }
        }, this.pongTimeout);

        this.timers.set(clientId, timer);
    }

    handlePong(clientId) {
        const timer = this.timers.get(clientId);
        if (timer) {
            clearTimeout(timer);
            this.timers.delete(clientId);
        }
    }
}

/* 消息推送服务 */
class MessagePushService {
    constructor(wsServer) {
        this.wsServer = wsServer;
        this.redis = new Redis();
        this.subscriber = this.redis.duplicate();
    }

    init() {
        /* 订阅消息队列 */
        this.subscriber.subscribe('ims:push');
        this.subscriber.on('message', (channel, msg) => {
            this.handlePushMessage(msg);
        });
    }

    handlePushMessage(msg) {
        const data = JSON.parse(msg);
        const { targetType, target, payload } = data;

        switch (targetType) {
            case 'user':
                this.pushToUser(target, payload);
                break;
            case 'role':
                this.pushToRole(target, payload);
                break;
            case 'all':
                this.pushToAll(payload);
                break;
        }
    }

    pushToUser(userId, payload) {
        const connections = this.wsServer.userConnections.get(userId);
        if (!connections) return;

        const message = JSON.stringify(payload);

        for (const clientId of connections) {
            const ws = this.wsServer.clients.get(clientId);
            if (ws && ws.readyState === WebSocket.OPEN) {
                ws.send(message);
            }
        }
    }

    pushToRole(roleId, payload) {
        /* 获取角色下的所有用户 */
        getUsersByRole(roleId).then(users => {
            users.forEach(user => this.pushToUser(user.id, payload));
        });
    }
}

/* 客户端重连逻辑 */
class WebSocketClient {
    constructor(url) {
        this.url = url;
        this.reconnectDelay = 1000;
        this.maxReconnectDelay = 30000;
        this.maxReconnectAttempts = 10;
    }

    connect() {
        this.ws = new WebSocket(this.url);

        this.ws.onclose = (event) => {
            if (!event.wasClean) {
                this.scheduleReconnect();
            }
        };

        this.ws.onerror = () => {
            this.scheduleReconnect();
        };
    }

    scheduleReconnect() {
        if (this.reconnectAttempts >= this.maxReconnectAttempts) {
            console.error('Max reconnect attempts reached');
            return;
        }

        setTimeout(() => {
            this.reconnectAttempts++;
            this.reconnectDelay = Math.min(
                this.reconnectDelay * 2,
                this.maxReconnectDelay
            );
            this.connect();
        }, this.reconnectDelay);
    }
}

/* 消息协议 */
const MESSAGE_TYPES = {
    /* 系统消息 */
    PING: 'ping',
    PONG: 'pong',
    CONNECTED: 'connected',
    ERROR: 'error',

    /* 业务消息 */
    NOTIFICATION: 'notification',  // 通知
    ORDER_UPDATE: 'order_update',  // 订单更新
    MESSAGE: 'message',          // 站内信
    ONLINE_STATUS: 'online_status', // 在线状态
    COLLABORATION: 'collaboration', // 实时协作
};

/* 消息格式 */
interface WSMessage {
    type: string;
    id?: string;
    timestamp: number;
    payload: object;
}