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

【问题标题】：Nodejs TCP server process packets in orderNodejs TCP服务器按顺序处理数据包
【发布时间】：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'。是的：接收者必须缓冲传入的数据）`

标签： node.js sockets asynchronous tcp async-await

【解决方案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);
  });
}

【讨论】：