Node.js TCP Socket通信中意外销毁问题排查及复用需求

问题描述 #

使用Node.js net模块实现单客户端TCP服务端，客户端连接后通信正常，但发送若干次数据包后socket意外销毁，仅能通过重启服务或客户端临时恢复，目标是复用同一socket连接。

相关代码如下：

服务端核心代码 #

import net from "net";
private static server: net.Server;
private static socket: net.Socket | null = null;

private static createTCPServer(): Promise<{ error: boolean; message?: string; result?: any }> {
    return new Promise((resolve, reject) => {
      const port = 8192;
      this.server = net.createServer((socket) => {
        // Connection Keep alive forever
        socket.setKeepAlive(true);
        console.log(false, "TCP IP client connected." , { ClientIp: socket.remoteAddress , Port: socket.remotePort });
        
        this.socket = socket;
        this.promiseResolve = resolve;
        this.promiseReject = reject;

        this.handleIncomingData(this.promiseResolve, false);

        socket.on('error', (err) => {
          if (err) {
            console.log(true, "TCP IP Client connection reset, Try again later...", err);
          } else {
            console.log(true, "TCP IP Client Socket error: Try again later...", err);
          }
          reject({error: true, message: "TCP Client connection problem, Try again later...", result : err });
        });

        // Handle Client Disconnection
        socket.on('end', () => {
          console.log(false, "TCP IP Client Disconnected ..", { ClientIp: socket.remoteAddress , Port: socket.remotePort });
          reject({ error: true, message: "TCP IP Client Disconnected." });
          }
        );

        socket.on('close', hadError => {
          if (hadError) {
            console.log(true, "TCP IP Client Disconnected with error.", hadError);
          } else {
            console.log(false, "TCP IP Client Disconnected gracefully.", hadError);
          }
          reject({ error: true, message: "TCP IP Client Disconnected.", result: hadError});
        });
      });

      this.server.listen(8192, () => {
        console.log(false, `TCP Server Listen at port 8192`);
        resolve({ error: false, message: `TCP Server Listen at port ${port}` }); // Resolve the promise when the server is successfully listening
      });

      this.server.on("error", (err) => {
        console.log(true, `TCP Server error: ${err}`);
        reject({ error: true, message: "TCP Server Error.", result: err }); // Reject the promise if there's an error starting the server
      });
    });
  }

Socket状态判断代码 #

if (!socket.destroyed) {
      socket.write(packet);
   } else {
      reject({ error: true, message: "Socket Destroyed Try again later." });
   }

客户端响应处理代码 #

let onResponseReceived;
onResponseReceived = (response) => {
        if (response) {
          console.log(false, `Received data from NMDL Client, Data ${this.socket.listenerCount("data")}.`, response.toString().trim());
          resultResponse = Client_Response(response.toString().trim(), resolve);
            this.socket.removeListener("data", onResponseReceived);
}
}

应用流程：

1. Node.js服务启动，TCP客户端连接
2. 支持两种响应方式：持续接收客户端数据包；发送数据包后接收响应（后者出现socket销毁问题）

问题根源分析 #

1. 一次性移除data事件监听器导致后续数据无法处理
   响应处理代码中，每次收到客户端响应后调用this.socket.removeListener("data", onResponseReceived)移除了data事件监听器。若客户端后续持续发送数据，服务端无法处理会导致TCP缓冲区溢出，最终触发socket错误或关闭逻辑，导致socket被销毁。

2. Promise作用域冲突引发异常
   createTCPServer返回的Promise仅用于通知服务器启动状态，但客户端连接后的error、end、close事件直接调用了reject，导致Promise被多次拒绝（服务器启动时已resolve），引发未捕获的Promise异常，间接导致socket资源被错误回收。

3. 未处理socket的'drain'事件导致写入失败
   当socket.write()返回false时，说明TCP发送缓冲区已满，若未监听drain事件等待缓冲区可用，后续写入操作可能失败，触发socket错误导致连接关闭。

解决方案 #

1. 修正data事件监听器逻辑 #

区分两种接收模式，避免误移除必要的监听器：

• 持续接收模式：全局绑定一次监听器，不主动移除
• 请求-响应模式：为每次请求绑定临时监听器，处理完响应后仅移除该临时监听器

示例代码：

// 持续接收的全局监听器（客户端连接时绑定一次）
const onContinuousData = (response) => {
  console.log(false, `Received continuous data from client:`, response.toString().trim());
  // 处理持续接收的业务逻辑
};
this.socket.on('data', onContinuousData);

// 请求-响应模式的临时监听器
const sendRequestAndWaitForResponse = (packet, resolve) => {
  const onResponseReceived = (response) => {
    const data = response.toString().trim();
    console.log(false, `Received response data from client:`, data);
    const resultResponse = Client_Response(data, resolve);
    // 仅移除当前请求的临时监听器
    this.socket.removeListener("data", onResponseReceived);
  };
  this.socket.on('data', onResponseReceived);
  
  // 发送数据包并处理缓冲区状态
  if (!this.socket.destroyed) {
    const canWrite = this.socket.write(packet);
    if (!canWrite) {
      this.socket.once('drain', () => {
        console.log(false, "Socket buffer drained, ready for next write.");
      });
    }
  } else {
    reject({ error: true, message: "Socket Destroyed Try again later." });
  }
};

2. 分离服务器启动与客户端事件的Promise逻辑 #

createTCPServer的Promise仅用于通知服务器启动状态，客户端连接、断开、错误等事件使用独立的回调或自定义事件处理：

修改后的核心代码片段：

private static createTCPServer(): Promise<{ error: boolean; message?: string; result?: any }> {
    return new Promise((resolve, reject) => {
      const port = 8192;
      this.server = net.createServer((socket) => {
        socket.setKeepAlive(true);
        console.log(false, "TCP IP client connected." , { ClientIp: socket.remoteAddress , Port: socket.remotePort });
        
        this.socket = socket;
        this.handleIncomingData(socket);

        socket.on('error', (err) => {
          console.log(true, "TCP IP Client connection error:", err);
          this.emitClientError(err);
        });

        socket.on('end', () => {
          console.log(false, "TCP IP Client Disconnected ..", { ClientIp: socket.remoteAddress , Port: socket.remotePort });
          this.emitClientDisconnected("Client initiated disconnect");
        });

        socket.on('close', hadError => {
          const message = hadError ? "TCP IP Client Disconnected with error." : "TCP IP Client Disconnected gracefully.";
          console.log(hadError, message, hadError);
          this.emitClientDisconnected(message, hadError);
        });
      });

      this.server.listen(8192, () => {
        console.log(false, `TCP Server Listen at port 8192`);
        resolve({ error: false, message: `TCP Server Listen at port ${port}` });
      });

      this.server.on("error", (err) => {
        console.log(true, `TCP Server error: ${err}`);
        reject({ error: true, message: "TCP Server Error.", result: err });
      });
    });
  }

// 新增独立事件通知方法
private static emitClientError(err: Error) {
  console.error("Client error occurred:", err);
  // 可在此触发自定义事件，让业务逻辑处理错误
}

private static emitClientDisconnected(message: string, hadError?: boolean) {
  console.log("Client disconnected:", message);
  this.socket = null;
  // 可在此触发自定义事件，处理断开后的重连等逻辑
}

3. 处理socket的'drain'事件，保证写入可靠性 #

封装写入方法，处理缓冲区满的情况：

const writeToSocket = (packet: Buffer | string) => {
  if (!this.socket || this.socket.destroyed) {
    return Promise.reject({ error: true, message: "Socket is not available." });
  }

  return new Promise((resolve) => {
    const canWrite = this.socket.write(packet);
    if (canWrite) {
      resolve();
    } else {
      this.socket.once('drain', resolve);
    }
  });
};

// 使用示例
writeToSocket(packet)
  .then(() => console.log("Packet sent successfully"))
  .catch(err => console.error("Failed to send packet:", err));

4. 增加socket状态监控日志 #

在关键节点打印socket状态，辅助定位问题：

console.log("Socket current status:", {
  destroyed: this.socket?.destroyed,
  writable: this.socket?.writable,
  readable: this.socket?.readable,
  bufferSize: this.socket?.bufferSize
});

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