用户对于实时性的要求Yi经到了近乎苛刻的地步。想象一下当你正未免有杀鸡用牛刀之嫌。这时候，Server-Sent Events 就像是一股清流，凭借其基于HTTP的轻量级特性，成为了hen多开发者的首选。

但是SSE原生只支持服务器到客户端的单向通信。Ru果我们想要构建一个既Neng实时推送通知，又Neng让客户端主动上报状态、发送指令的“双向”系统，该怎么办呢？Zui近在公司项目中，我们就遇到了这样一个挑战。我们需要一个轻量级但功Neng完备的解决方案，既Neng像ChatGPT那样流畅地输出文本，又Neng处理复杂的业务指令。经过一番摸索，我们设计并实现了一套基于SSE的通用双向消息通知SDK。今天我就把这套方案的架构设计和实现细节毫无保留地分享出来。

一、 架构设计：打破单向的桎梏

在设计之初，我们必须明确一个核心概念：所谓的“SSE双向通信”，本质上是一种“混合模式”。SSE依然负责它Zui擅长的部分——服务器向客户端的实时数据流；而客户端向服务器的通信，则回归到Zui朴素的HTTP POST请求。这种组合拳打法，既保留了SSE穿透防火墙、自动重连的优势，又规避了引入复杂协议的必要性。

我们的设计目标非常明确：构建一个功Neng完备、易于使用的双向消息通知系统。它不仅要支持文本、JSON、系统通知等多种消息类型，还要Neng在高并发环境下稳如泰山。geng重要的是它必须具备“断线重连”和“消息确认”机制，确保在网络抖动这种不可控因素发生时关键数据绝不丢失。

1.1 消息模型定义

在代码实现之前，我们需要先定义好“语言”。所有的消息dou遵循一个统一的接口，这是SDK通用性的基础。我们使用TypeScript来定义这些类型，确保在编译阶段就Neng规避大部分低级错误。

/**
 * 消息类型枚举
 * 定义了SDK支持的所有消息类型
 */
export enum MessageType {
    /** 普通文本消息 */
    TEXT = 'text',
    /** JSON结构化消息 */
    JSON = 'json',
    /** 系统通知消息 */
    SYSTEM = 'system',
    /** 指令消息 */
    COMMAND = 'command',
    /** 心跳消息 */
    HEARTBEAT = 'heartbeat',
    /** 确认消息 */
    ACKNOWLEDGE = 'ack',
    /** 错误消息 */
    ERROR = 'error',
    /** 进度通知 */
    PROGRESS = 'progress'
}
/**
 * 消息优先级枚举
 * 用于消息处理的优先级调度
 */
export enum MessagePriority {
    LOW = 0,
    NORMAL = 1,
    HIGH = 2,
    CRITICAL = 3
}
/**
 * 消息接口定义
 * 所有消息的基类接口
 */
export interface Message {
    /** 消息唯一标识符 */
    id: string;
    /** 消息类型 */
    type: MessageType;
    /** 消息优先级 */
    priority: MessagePriority;
    /** 消息创建时间戳 */
    timestamp: number;
    /** 消息来源标识 */
    source: string;
    /** 消息目标标识 */
    target?: string;
    /** 消息载荷数据 */
    payload: any;
    /** 消息元数据 */
    metadata?: Record;
}

1.2 核心组件划分

为了保持代码的整洁和可维护性，我们将SDK拆分为几个核心模块：

连接管理器 ： 负责SSE连接的建立、维护、状态监控以及断线重连。它是SDK的“心脏”。

消息队列管理器 ： 负责消息的缓存、优先级排序和可靠投递。当网络不稳定时它就是我们的“保险箱”。

事件分发器 ： 采用观察者模式，将接收到的消息精准地分发给对应的订阅者。它是SDK的“神经系统”。

HTTP客户端 ： 封装了POST请求，专门用于客户端向服务器发送指令或数据。

二、 前端SDK实现：构建坚韧的客户端

前端SDK是开发者直接接触的部分，它的API设计必须足够优雅。我们不想让使用者去关心底层的EventSource是如何工作的，他们只需要知道“连接”、“发送”、“监听”即可。

2.1 连接管理器：状态机的艺术

连接管理是整个系统Zui脆弱的环节。网络环境千变万化，用户可Neng会切换Wi-Fi到4G，也可Neng直接关闭浏览器标签页。我们的`SSEConnectionManager`必须Neng够应对这些情况。

我们定义了一个状态机：`DISCONNECTED` -> `CONNECTING` -> `CONNECTED` -> `RECONNECTING`。每一个状态的变gengdou需要通知上层应用。此外为了防止“僵尸连接”，我们引入了心跳机制。Ru果服务器长时间没有发送数据，或者客户端长时间没有收到心跳，就会触发重连逻辑。

export enum ConnectionState {
    DISCONNECTED = 'disconnected',
    CONNECTING = 'connecting',
    CONNECTED = 'connected',
    RECONNECTING = 'reconnecting',
    ERROR = 'error'
}
export class SSEConnectionManager {
    private eventSource: EventSource | null = null;
    private config: SSKSConfig;
    private state: ConnectionState = ConnectionState.DISCONNECTED;
    // ... 其他属性
    public connect: Promise {
        return new Promise => {
            if ) {
                resolve;
                return;
            }
            const prevState = this.state;
            this.setState;
            const url = this.buildConnectionUrl;
            try {
                this.eventSource = new EventSource;
                // 设置超时保护，防止一直处于Connecting状态
                const timeout = setTimeout => {
                    if  {
                        this.eventSource?.close;
                        reject);
                    }
                }, this.config.connectionTimeout!);
                this.eventSource.onopen =  => {
                    clearTimeout;
                    this.setState;
                    this.startHeartbeat;
                    resolve;
                };
                this.eventSource.onerror =  => {
                    clearTimeout;
                    // 错误处理逻辑，触发重连
                    this.handleDisconnect;
                };
            } catch  {
                reject;
            }
        });
    }
    // 指数退避算法计算重连延迟
    private calculateReconnectDelay: number {
        const delay = Math.min(
            this.config.reconnectBaseDelay! * Math.pow,
            this.config.reconnectMaxDelay!
        );
        // 添加随机抖动，避免惊群效应
        return delay + Math.random * delay * 0.2;
    }
}

2.2 消息队列：确保万无一失

有时候，消息发送出去了但连接突然断了。Ru果没有队列，这条消息就石沉大海了。我们的`MessageQueueManager`实现了优先级队列，确保高优先级的系统通知不会被低优先级的聊天消息堵在路上。

export class MessageQueueManager {
    private queue: QueueItem = ;
    private maxSize: number;
    public enqueue: boolean {
        if  {
            // 队列满了尝试驱逐低优先级消息
            this.evictLowPriority;
            if  {
                return false; // 实在没办法了只Neng丢弃
            }
        }
        const item: QueueItem = {
            message,
            enqueuedAt: Date.now,
            priority: options.priority ?? message.priority,
            retryCount: 0
        };
        this.queue.push;
        this.sortByPriority; // 按优先级排序
        return true;
    }
    private sortByPriority: void {
        this.queue.sort => {
            if  {
                return b.priority - a.priority; // 优先级高的在前
            }
            return a.enqueuedAt - b.enqueuedAt; // 同优先级按时间
        });
    }
}

2.3 事件分发器：解耦的关键

不要在连接管理器里直接写业务逻辑！这是开发的大忌。我们使用`EventDispatcher`来解耦消息的接收和处理。业务层只需要订阅自己感兴趣的消息类型，剩下的交给分发器。

export class EventDispatcher {
    private subscribers: Map = new Map;
    public subscribe(
        messageTypes: MessageType,
        handler: MessageHandler,
        options: { id?: string; priority?: number; once?: boolean; } = {}
    ): string {
        const id = options.id || this.generateId;
        const subscriber: Subscriber = {
            id,
            messageTypes,
            handler,
            priority: options.priority ?? 0,
            once: options.once
        };
        this.subscribers.set;
        return id;
    }
    public async dispatch: Promise {
        // 筛选出订阅了该消息类型的订阅者
        const sortedSubscribers = this.getSortedSubscribers;
        for  {
            try {
                await subscriber.handler;
                if  {
                    this.unsubscribe;
                }
            } catch  {
                console.error;
            }
        }
    }
}

2.4 SDK主类：统一入口

Zui后我们用一个主类`SSENotificationSDK`将上述组件串联起来。它对外暴露简洁的API，如`connect`, `sendText`, `sendCommand`, `subscribe`等。

export class SSENotificationSDK {
    private connectionManager: SSEConnectionManager;
    private messageQueue: MessageQueueManager;
    private eventDispatcher: EventDispatcher;
    private httpClient: HttpClient;
    constructor {
        this.config = { ...DEFAULT_CONFIG, ...config };
        this.connectionManager = new SSEConnectionManager;
        this.messageQueue = new MessageQueueManager;
        this.eventDispatcher = new EventDispatcher;
        this.httpClient = new HttpClient;
        this.setupInternalHandlers;
    }
    public async sendText: Promise {
        const message: Message = {
            id: this.generateMessageId,
            type: MessageType.TEXT,
            priority: MessagePriority.NORMAL,
            timestamp: Date.now,
            source: this.config.clientId,
            payload: { content }
        };
        await this.send;
    }
    // ... 其他方法
}

三、 后端服务实现：Node.js与Express的强强联合

前端有了SDK，后端也得跟上。我们选择Node.js和Express来实现服务端，因为它们的异步I/O模型非常适合处理高并发的长连接。

3.1 SSE服务端核心

服务端的核心在于维护一个`connections`映射表，Key是客户端ID，Value是对应的响应对象。当需要发送消息时服务器通过这个映射表找到对应的客户端，利用`res.write`将数据推出去。

import express, { Request, Response } from 'express';
class SSEMessageService {
    private connections: Map = new Map;
    private handleSSEConnection: void {
        const clientId = req.query.clientId as string;
        // 设置SSE专用的HTTP头
        res.writeHead(200, {
            'Content-Type': 'text/event-stream',
            'Cache-Control': 'no-cache',
            'Connection': 'keep-alive',
            'X-Accel-Buffering': 'no' // 禁用Nginx缓冲，确保实时性
        });
        const connection: SSEConnection = {
            id: this.generateConnectionId,
            clientId,
            response: res,
            createdAt: Date.now
        };
        this.connections.set;
        // 监听客户端断开事件
        req.on => {
            this.handleConnectionClose;
        });
        // 立即发送一个连接成功事件
        this.sendToClient(clientId, {
            type: 'system',
            payload: { event: 'connect' }
        });
    }
    private sendToClient: boolean {
        const connection = this.connections.get;
        if  return false;
        try {
            const data = JSON.stringify;
            // SSE格式要求：event: xxx
data: xxx

            connection.response.write;
            connection.response.write;
            return true;
        } catch  {
            this.handleConnectionClose;
            return false;
        }
    }
}

3.2 并发指令调度器

既然是双向通信，客户端发送的指令往往需要服务器进行耗时处理。我们不Neng让每一个HTTP请求dou阻塞线程，因此需要一个调度器来管理这些并发任务。

我们的`CommandScheduler`支持任务队列、并发限制、超时控制和重试机制。它就像一个交通指挥员，确保所有的任务douNeng有条不紊地执行。

class CommandScheduler {
    private runningTasks: Map = new Map;
    private pendingQueue: CommandTask = ;
    private config: SchedulerConfig;
    public async submitTask: Promise {
        const taskId = `task_${Date.now}`;
        const task: CommandTask = {
            taskId,
            commandId: command.payload.commandId,
            action: command.payload.action,
            status: 'pending',
            createdAt: Date.now
        };
        this.tasks.set;
        this.enqueueTask;
        return taskId;
    }
    private scheduleNext: void {
        if  return;
        const task = this.pendingQueue.shift;
        if  return;
        this.executeTask;
    }
    private async executeTask: Promise {
        task.status = 'running';
        this.runningTasks.set;
        try {
            // 模拟执行业务逻辑
            const handler = this.taskHandlers.get;
            if  {
                const result = await handler;
                task.result = result;
                task.status = 'completed';
                // 通知客户端任务完成
                this.notifyClient;
            }
        } catch  {
            task.status = 'failed';
            // 失败重试逻辑
            if ) {
                this.enqueueTask;
            }
        } finally {
            this.runningTasks.delete;
            this.scheduleNext; // 调度下一个
        }
    }
}

四、 与Zui佳实践

通过这套SSE通用双向消息通知SDK，我们成功地在轻量级和功Neng性之间找到了平衡点。相比于WebSocket的全双工复杂性，SSE+HTTP的组合在实现实时通知、股票行情、新闻geng新等以服务端数据为主导的业务场景时显得尤为简洁高效。

在实际落地过程中，我有几点切身体会想分享给大家：

不要忽视心跳： hen多中间件会默默切断长时间没有数据传输的连接。保持心跳是维持长连接的生命线。

重连要有策略： 盲目地立即重连可Neng会造成服务器雪崩。使用指数退避算法，并在重连之间加入随机抖动，是geng优雅的Zuo法。

消息ID至关重要： 每一条消息dou应该有唯一的ID。这不仅用于去重，geng是实现“消息确认”和“断点续传”的基础。

错误处理要分级： 并不是所有的错误dou需要重连。比如认证失败，就应该直接停止连接并提示用户，而不是傻傻地一直重试。

无论是即时消息、订单状态geng新，还是系统公告，Neng够及时将信息推送给用户，douNeng显著增强应用的互动性和实用性。希望这套方案Neng为你的下一个项目提供一些灵感。技术选型没有绝对的银弹，只有Zui适合业务场景的那一个。在SSE和WebSocket之间，愿你也Neng找到那个“对的人”。

---