React EventSource偶发无法接收Node.js后端SSE推送数据问题

核心根因 #

消息丢失不是浏览器、SQS或MongoDB的问题，是代码实现违反SSE协议规范+时序逻辑错误导致的，具体问题点如下：

• 单条SSE消息的帧字段被拆分多次异步写入，Node.js流缓冲会随机将相邻消息的帧拼接在同一个TCP块中，浏览器解析到不合法的帧会直接丢弃，既不触发onmessage回调也不抛出错误，和你观察到的“直接写data字段正常、加id/retry/event字段就丢消息”现象完全吻合。
• 每条消息都重复写入retry: 10000字段。retry是SSE连接级配置，仅需在连接建立时写入一次，重复写入会导致浏览器反复重置重连计时器，部分Chromium内核版本会在此场景下直接丢弃未解析完成的消息帧。
• SQS消息删除时序错误：你将消息写入逻辑放在setTimeout中异步执行，但删除SQS消息是同步await执行的，相当于消息还没推给客户端就已经从队列中删除，一旦流写入因为网络背压、缓冲调度失败，消息会永久丢失。
• 缺少连接断开监听：客户端关闭页面/断开连接后，服务端的轮询递归不会终止，持续向已销毁的响应流写入数据会触发Node.js内部缓冲错乱，导致后续新建立的SSE连接出现帧解析异常，和你观察到的“重启服务端临时恢复、刷新页面复现”现象一致。
• Keep-Alive头配置错误：max=10参数表示该TCP连接最多复用10个HTTP请求，对SSE这种单请求长连接场景无意义，反而会触发部分反向代理主动断开连接。

修复后代码 #

服务端代码 #

export const realTimeLogs = async (req: Request, res: Response) => {
  res.writeHead(200, {
    Connection: 'keep-alive',
    'Keep-Alive': 'timeout=3600',
    'Content-Type': 'text/event-stream',
    'Cache-Control': 'no-cache',
    'Transfer-Encoding': 'chunked',
    'Access-Control-Allow-Origin': 'http://localhost:3000',
    'Access-Control-Allow-Credentials': 'true',
    'X-Accel-Buffering': 'no',
  });

  // 连接级配置仅在握手阶段写入一次
  res.write('retry: 10000\n\n');
  res.write('');

  // 监听连接断开状态，终止无效轮询
  let isConnectionClosed = false;
  req.on('close', () => {
    isConnectionClosed = true;
  });

  const pollAndPushMessage = async () => {
    if (isConnectionClosed) return;
    try {
      const msg: ReceiveMessageCommandOutput = await createAndReceiveMessage('mySQSQueue');
      if (msg.Messages?.length) {
        const msgBody = msg.Messages[0].Body;
        const receiptHandle = msg.Messages[0].ReceiptHandle;
        const msgId = JSON.parse(msgBody!).payload._id;

        // 单条消息的所有帧字段一次性拼接完成，单次write调用写入，禁止拆分、禁止套setTimeout
        const sseFrame = [
          `id: ${msgId}`,
          'event: message',
          `data: ${msgBody}`,
          '',
          '' // 末尾两个\n标记单条消息结束，符合SSE协议规范
        ].join('\n');

        // 等待流写入完成后再删除SQS消息，确保消息至少成功写入TCP缓冲
        await new Promise((resolve) => {
          res.write(sseFrame, 'utf-8', resolve);
        });

        console.log(`消息推送成功: ${msgBody}`);
        if (receiptHandle) {
          await createAndDeleteMessage('mySQSQueue', receiptHandle);
        }
      }
    } catch (err) {
      console.error('SSE推送流程异常:', err);
    } finally {
      // 轮询间隔可根据业务调整，建议配合SQS长轮询降低空请求开销
      if (!isConnectionClosed) {
        setTimeout(pollAndPushMessage, 1000);
      }
    }
  };

  pollAndPushMessage();
}

客户端代码修复 #

原代码的清理逻辑存在闭包陷阱，useEffect卸载时读取的eventSource是初始null值，会导致旧连接无法正常关闭，修复如下：

useEffect(() => {
  const source = new EventSource(
    `${BASE_URL}/${ROOT_ENDPOINTS.MANAGER}/${MANAGER_ENDPOINTS.GETREALTIMELOGS}`,
    {
      withCredentials: true,
    }
  );
  source.onopen = (event: Event) => {
    console.log('SSE连接已建立: ', event);
  };
  source.onerror = (event: Event) => {
    console.error('SSE连接异常: ', event);
  };
  source.onmessage = (event: MessageEvent) => {
    try {
      const payload: MSGManager = JSON.parse(event.data);
      eventSourceHandler(payload);
    } catch (e) {
      console.error('消息解析失败: ', e);
    }
  };
  setEventSource(source);

  return () => {
    console.log('销毁SSE连接');
    // 直接使用当前作用域下的source实例关闭，不依赖外层state的闭包值
    source.close();
  };
  // eslint-disable-next-line react-hooks/exhaustive-deps
}, []);

优化建议 #

• 调用SQS ReceiveMessage接口时可配置WaitTimeSeconds: 20开启长轮询，无需固定间隔轮询，可大幅降低空请求开销。
• 若不需要自定义事件类型，可省略event: message行，默认消息就会触发onmessage回调，减少不必要的字段写入。

内容的提问来源于stack exchange，提问作者Stefano Leone