Nodejs TCP服务器按顺序处理数据包

Posted 2023-03-23

【中文标题】Nodejs TCP服务器按顺序处理数据包

【英文标题】：Nodejs TCP server process packets in order

【发布时间】：2021-11-29 01:34:14

【问题描述】：

我正在使用 tcp 服务器在 Node.js 中接收和处理数据包。它应该收到 2 个数据包：

“create”用于在数据库中创建对象。它首先检查对象是否已经存在，然后创建它。 （-> 需要一些时间） “update”用于更新数据库中新创建的对象

为简单起见，我们将假设第一步总是比第二步花费更长的时间。 （在我的原始代码中总是如此）

这是一个 MWE：

const net = require("net");

const server = net.createServer((conn) => 
  conn.on('data', async (data) => 
    console.log(`Instruction $data recieved`);
    await sleep(1000);
    console.log(`Instruction $data done`);
  );
);
server.listen(1234);
const client = net.createConnection(1234, 'localhost', async () => 
  client.write("create");
  await sleep(10); // just a cheap workaround to "force" sending 2 packets instead of one
  client.write("update");
);

// Just to make it easier to read
function sleep(ms) 
  return new Promise((resolve) => 
    setTimeout(resolve, ms);
  );

如果我运行这段代码，我会得到：

Instruction create recieved
Instruction update recieved
Instruction create done
Instruction update done

但我希望“创建”指令阻止 conn.on('data', func)，直到最后一个回调异步返回。当前代码尝试在数据库中创建条目之前对其进行更新，这并不理想。

有没有（优雅的）方法来实现这一点？我怀疑某种存储数据的缓冲区和某种处理数据的工作循环？但是我如何避免运行阻塞事件循环的无限循环？ （事件循环是正确的术语，是吗？）

注意：我有更多的逻辑来处理碎片等。但这解释了我遇到的问题。

【问题讨论】：

... just a cheap workaround to "force" sending 2 packets instead of one ... TCP 是一种流协议。没有数据包。但是有order。

@wildplasser TCP 中没有数据包，正确。但是我指的是IP。我希望 single TCP stream 在 two IP packets 中发送。

你可能会混淆 Nagle，但不能混淆 TCP 窗口协商。您的 solution 根本上是错误的。真正的解决方案是：设计一个协议。 （最简单的协议：在每个数据包之后添加一个'\n'。是的：接收者必须缓冲传入的数据）`

【参考方案1】：

我设法让它与包async-fifo-queue 一起工作。 这不是最干净的解决方案，但它应该做我想做的并且尽可能高效（使用 async/await 而不是无限循环）。

代码：

const net = require("net");
const afq = require("async-fifo-queue");

const q = new afq.Queue();

const server = net.createServer((conn) => 
  conn.on('data', q.put.bind(q));
);
server.listen(1234);
const client = net.createConnection(1234, 'localhost', async () => 
  client.write("create");
  await sleep(10);
  client.write("update");
);

(async () => 
  while(server.listening) 
    const data = await q.get();
    console.log(`Instruction $data recieved`);
    await sleep(1000);
    console.log(`Instruction $data done`);
  
)();



function sleep(ms) 
  return new Promise((resolve) => 
    setTimeout(resolve, ms);
  );

【参考方案2】：

您可以在收到“create”事件时暂停套接字。完成后，您可以恢复套接字。示例：

const server = net.createServer((conn) => 
  conn.on('data', async (data) => 
    if (data === 'create') 
        conn.pause()
    
    console.log(`Instruction $data recieved`);
    await sleep(1000);
    console.log(`Instruction $data done`);
    if (data === 'create') 
        conn.resume()
    
  );
);
server.listen(1234);
const client = net.createConnection(1234, 'localhost', async () => 
  client.write("create");
  await sleep(10); // just a cheap workaround to "force" sending 2 packets instead of one
  client.write("update");
);

// Just to make it easier to read
function sleep(ms) 
  return new Promise((resolve) => 
    setTimeout(resolve, ms);
  );