IMS消息推送系统设计

消息推送是企业信息系统中提升用户体验的关键功能。本文将介绍 IMS 系统中消息推送的整体设计方案。

消息推送的场景需求

IMS 系统中的消息推送主要覆盖以下场景：

• 审批通知 - 待审批事项自动推送给审批人
• 任务提醒 - 截止时间临近时提醒任务负责人
• 系统公告 - 重要通知全员推送
• 数据变更 - 关联数据更新时通知相关人员
• 即时通讯 - 消息提醒与未读计数

整体架构设计

消息推送系统采用分层架构：

• 接入层 - WebSocket 长连接 + HTTP 短连接
• 路由层 - 消息分发与用户路由
• 存储层 - 消息持久化与离线消息
• 推送层 - 实际推送通道（APP 推送、邮件、短信）

WebSocket 实时推送

实现基于 WebSocket 的实时消息推送：

// 消息推送服务
const WebSocket = require('ws');
const wss = new WebSocket.Server({ port: 8080 });

// 用户连接管理
const userConnections = new Map();  // userId -> WebSocket

wss.on('connection', (ws, req) => {
    // 从 URL 或 Token 获取用户 ID
    const userId = getUserIdFromRequest(req);
    userConnections.set(userId, ws);

    console.log(`用户 ${userId} 连接`);

    // 心跳保活
    ws.on('ping', () => ws.pong());

    // 断开连接
    ws.on('close', () => {
        userConnections.delete(userId);
        console.log(`用户 ${userId} 断开`);
    });

    // 接收客户端消息
    ws.on('message', (data) => {
        const msg = JSON.parse(data);
        handleClientMessage(userId, msg);
    });
});

// 推送消息给指定用户
function pushToUser(userId, message) {
    const ws = userConnections.get(userId);
    if (ws && ws.readyState === WebSocket.OPEN) {
        ws.send(JSON.stringify(message));
        return true;
    }
    return false;
}

// 广播消息给所有在线用户
function broadcast(message) {
    userConnections.forEach((ws, userId) => {
        pushToUser(userId, message);
    });
}

消息类型定义

统一消息格式，便于前端处理：

// 消息类型枚举
const MessageType = {
    NOTIFICATION: 'notification',  // 系统通知
    APPROVAL: 'approval',          // 审批提醒
    TASK: 'task',                  // 任务提醒
    CHAT: 'chat',                  // 即时消息
    SYSTEM: 'system'              // 系统公告
};

// 消息格式
const MessageSchema = {
    id: '消息唯一ID',
    type: '消息类型',
    title: '消息标题',
    content: '消息内容',
    sender: '发送者ID',
    receivers: ['接收者ID列表'],
    priority: '优先级: high/normal/low',
    action: {
        url: '点击跳转链接',
        method: 'POST/GET'
    },
    createdAt: '创建时间',
    expireAt: '过期时间'
};

消息存储设计

使用数据库存储消息，支持离线消息：

// 消息存储服务
class MessageStore {
    async save(message) {
        // 存储消息到数据库
        const sql = `
            INSERT INTO messages (id, type, title, content, sender, priority, created_at, expire_at)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?)
        `;
        await db.execute(sql, [
            message.id,
            message.type,
            message.title,
            message.content,
            message.sender,
            message.priority,
            message.createdAt,
            message.expireAt
        ]);

        // 存储接收者关系
        for (const receiverId of message.receivers) {
            await db.execute(
                `INSERT INTO message_receivers (message_id, user_id, is_read, read_at) VALUES (?, ?, 0, NULL)`,
                [message.id, receiverId]
            );
        }
    }

    async getUnreadCount(userId) {
        const sql = `
            SELECT COUNT(*) as count FROM message_receivers
            WHERE user_id = ? AND is_read = 0
        `;
        const [row] = await db.execute(sql, [userId]);
        return row.count;
    }

    async markAsRead(messageId, userId) {
        await db.execute(
            `UPDATE message_receivers SET is_read = 1, read_at = NOW() WHERE message_id = ? AND user_id = ?`,
            [messageId, userId]
        );
    }
}

消息推送策略

根据消息类型和用户设置选择推送通道：

// 推送策略
class PushStrategy {
    async send(message, userId) {
        // 获取用户通知设置
        const settings = await getUserSettings(userId);

        // 1. WebSocket 在线推送
        const pushed = pushToUser(userId, message);

        // 2. 离线消息存储
        if (!pushed) {
            await messageStore.save(message);
        }

        // 3. 根据设置决定是否发送邮件
        if (settings.emailNotify && message.priority === 'high') {
            await sendEmail(userId, message);
        }

        // 4. APP 推送（移动端）
        if (settings.appNotify) {
            await sendAppPush(userId, message);
        }
    }
}

前端接收处理

前端使用 EventSource 或 WebSocket 接收消息：

// 前端消息接收
class MessageClient {
    constructor() {
        this.ws = null;
        this.reconnectDelay = 3000;
    }

    connect() {
        this.ws = new WebSocket(`wss://${location.host}/ws`);

        this.ws.onmessage = (event) => {
            const message = JSON.parse(event.data);
            this.handleMessage(message);
        };

        this.ws.onclose = () => {
            setTimeout(()=>this.connect(), this.reconnectDelay);
        };
    }

    handleMessage(message) {
        // 更新未读计数
        updateBadgeCount();

        // 显示通知
        if (message.type === 'notification') {
            showToast(message.title, message.content);
        }

        // 更新消息列表
        prependToList(message);
    }
}

总结

IMS 消息推送系统通过 WebSocket 实现实时推送，结合离线消息存储和多通道推送策略，确保用户能够及时收到重要通知。同时需要做好消息去重、限流等安全措施。