Node.js net.Socket close事件触发后仍传输数据异常排查

架构老林

2026-8-26

问题根因

首先是对socket的close事件触发时机存在认知偏差：官方文档描述的「套接字完全关闭」仅指本地TCP状态机进入CLOSED状态、对应文件描述符完成释放，不代表对端停止发送数据，也不代表用户态已经把所有接收缓冲区的数据处理完成。TCP是全双工协议，两个方向的传输通道独立关闭，触发close时两类残留数据会导致异常：
1. 本地关闭连接时，内核接收缓冲区可能还存着对端已经发出、尚未被用户态读取的TCP段，你在close回调里直接销毁流对象，后续缓冲区数据写入已销毁的Transform流时，就会抛出「已关闭流写入错误」。
2. IoT弱网场景下经常出现半开连接：比如设备突然掉电、NAT端口表过期，本地TCP栈因超时或收到RST包触发close事件时，设备端可能还没感知到连接断开，持续向旧连接四元组发送数据，这部分数据到达本地时流已经被销毁，同样会触发写入错误。
偶发数据丢失的核心原因是清理时机错误：你把流解绑、监听器移除的逻辑全放在close事件中，此时pipe链路的背压队列里可能还存着已经从内核读到用户态、但尚未推送给DeviceClass处理的数据，直接调用unpipe()会丢弃这部分存量数据，导致丢数。
原有代码存在隐性bug：close事件回调使用箭头函数，内部this继承自外层作用域，并不指向当前连接的socket实例，this.socket实际为undefined，会导致清理逻辑执行时报错，资源无法正常释放，进一步加剧异常概率。

修复方案

不要把所有清理逻辑绑定在close事件上，按TCP连接状态分层处理，核心原则是先排空存量数据，再释放资源：

调整事件监听逻辑，区分正常半关闭、错误、完全关闭三类场景：
- 监听end事件：该事件触发时代表对端已经发送FIN包，不会再产生新的业务数据，此时停止接收新数据，等待存量数据处理完成
- 监听error事件：捕获连接过程中的传输错误，标记异常状态触发清理
- close事件仅作为最终资源释放的兜底触发点，不做流解绑、数据处理相关操作
修正pipe配置，不要让socket结束时自动终止下游Transform流，避免数据未处理完流就被销毁
增加清理状态锁，避免重复执行释放逻辑；不要调用removeAllListeners()，该方法会移除流内部依赖的监听器导致状态异常，仅移除业务层手动绑定的监听器即可
不要永久关闭socket超时（即不要用socket.setTimeout(0)），结合IoT设备的心跳间隔设置合理的空闲超时，一般设为3倍心跳周期即可，及时清理僵死连接。

参考实现代码：

this.server = net.createServer(function onConnection(socket) {
  const device = new gateway.DeviceClass(gateway);
  // 按实际业务心跳配置超时，示例为3分钟
  socket.setTimeout(180000);
  let isCleaned = false;

  // 统一清理方法
  const cleanup = (hadError = false) => {
    if (isCleaned) return;
    isCleaned = true;

    // 解绑流链路
    socket.unpipe(device);
    // 仅移除业务绑定的自定义监听器
    socket.removeListener('end', handleEnd);
    socket.removeListener('error', handleError);
    socket.removeListener('close', handleClose);
    socket.removeListener('timeout', handleTimeout);

    // 等下游Transform流处理完所有存量数据后再销毁资源
    device.end(() => {
      if (!socket.destroyed) socket.destroy();
      socket.unref();
      // 这里执行设备离线、日志上报等业务逻辑
    });
  };

  const handleEnd = () => {
    // 对端发完数据，半关闭连接，停止接收新数据
    socket.end();
  };

  const handleError = (err) => {
    // 可按需打印错误日志
    cleanup(true);
  };

  const handleClose = (hadError) => {
    cleanup(hadError);
  };

  const handleTimeout = () => {
    // 空闲超时主动断开连接
    cleanup(false);
  };

  // 关键配置：end:false，避免socket结束时自动销毁device流
  socket.pipe(device, { end: false });

  socket.on('end', handleEnd);
  socket.on('error', handleError);
  socket.once('close', handleClose);
  socket.on('timeout', handleTimeout);

  // 捕获下游Transform流的解析错误，避免进程崩溃
  device.on('error', (err) => {
    // 可按需记录数据解析错误
    cleanup(true);
  });
});

额外优化点：如果对数据一致性要求极高，可以在handleEnd逻辑中轮询检查socket.readableLength属性，该值代表已经从内核读到用户态、尚未被流消费的数据长度，等该值归0后再触发清理，完全避免缓冲数据丢失。

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