基于Promise的net.createServer版本及TCP服务器data事件异步实现问询

阿华AIGC实验室

2026-5-19

适配async/await的TCP服务器data事件处理方案

嘿，这个问题问得很实用！Node.js原生的net模块确实是基于事件驱动的设计，没有直接提供Promise风格的API来处理data事件，但我们完全可以自己动手封装，或者用一些小技巧来适配async/await，不一定非得依赖第三方库。下面给你几种可行的方案：

1. 封装单次data事件的Promise版本

如果你只需要等待客户端发送的一次数据，可以把conn.on('data')封装成一个返回Promise的函数，同时处理错误和连接关闭的边界情况，避免内存泄漏：

const net = require('net');

// 封装等待data事件的Promise函数
function waitForData(conn) {
  return new Promise((resolve, reject) => {
    // 定义事件处理函数
    const handleData = (data) => {
      cleanup();
      resolve(data);
    };
    const handleError = (err) => {
      cleanup();
      reject(err);
    };
    const handleClose = () => {
      cleanup();
      reject(new Error('连接已关闭，未收到数据'));
    };

    // 绑定一次性事件监听
    conn.once('data', handleData);
    conn.once('error', handleError);
    conn.once('close', handleClose);

    // 清理函数：移除所有事件监听
    function cleanup() {
      conn.off('data', handleData);
      conn.off('error', handleError);
      conn.off('close', handleClose);
    }
  });
}

// 使用async/await创建服务器
const server = net.createServer(async (conn) => {
  console.log('客户端已连接');
  try {
    // 等待并接收数据
    const data = await waitForData(conn);
    console.log('收到数据:', data.toString());
    
    // 可以继续等待下一次数据（如果需要）
    // const moreData = await waitForData(conn);
  } catch (err) {
    console.error('处理出错:', err.message);
  } finally {
    conn.end();
    console.log('连接已关闭');
  }
});

server.listen(3000, () => {
  console.log('TCP服务器监听在端口3000');
});

2. 封装流式数据的异步迭代器版本

TCP是流式传输，客户端可能会分多次发送数据。这种情况下，把连接对象转换成异步迭代器，就能用for await...of循环来优雅处理所有数据块：

const net = require('net');

// 将流转换为异步迭代器
async function* streamToAsyncIterator(stream) {
  const dataQueue = [];
  let resolveNext;
  let rejectNext;

  // 监听流事件，把数据推入队列
  stream.on('data', (chunk) => {
    dataQueue.push(chunk);
    resolveNext?.(); // 唤醒等待的迭代器
  });
  stream.on('error', (err) => {
    rejectNext?.(err);
  });
  stream.on('end', () => {
    resolveNext?.(); // 流结束时唤醒迭代器
  });

  // 迭代队列中的数据
  while (!stream.destroyed) {
    if (dataQueue.length > 0) {
      yield dataQueue.shift();
    } else {
      // 等待新数据或流结束
      await new Promise((resolve, reject) => {
        resolveNext = resolve;
        rejectNext = reject;
      });
    }
  }
}

// 使用异步迭代器处理流式数据
const server = net.createServer(async (conn) => {
  console.log('客户端已连接');
  try {
    // 遍历所有收到的数据块
    for await (const chunk of streamToAsyncIterator(conn)) {
      console.log('收到数据块:', chunk.toString());
      // 在这里处理每个数据块
    }
  } catch (err) {
    console.error('流处理出错:', err.message);
  } finally {
    console.log('客户端已断开连接');
  }
});

server.listen(3000, () => {
  console.log('TCP服务器监听在端口3000');
});