如何管理基于消息队列侦听器 (NodeJS) 的并行 HTTP 请求答案

【问题标题】：How to manage parallel HTTP request that are based on Message Queuing Listneners (NodeJS)如何管理基于消息队列侦听器 (NodeJS) 的并行 HTTP 请求
【发布时间】：2019-07-17 07:10:44
【问题描述】：

我确信这种问题已经在这里解决了很多次，但我找不到这些问题是如何提出的。

我有一个微服务来处理我的基础架构和 MQTT 代理之间的通信。每次收到 HTTP 请求时，我都会发送“谁在房间 XXX 中还活着？” MQTT Broker 上的消息，并且在“XXX/alive”主题上注册的每个客户端都必须回答，我等待 Y 毫秒，然后通过将收到的响应发送回客户端来关闭请求。

当我处理一个请求时它运行良好。但是当一次提出多个请求时，它就搞砸了。

这是处理 HTTP 请求的 Express 路由：

app.get('/espPassports', (req, res) => {
  mqttHelper.getESPPassports(req.query.model_Name).then((passports) => {
    res.send(passports).end();
  }).catch(err => {
    res.send(err).end();
  })
})

getESPPassports 的工作原理如下：

getESPPassports: async (model_Name) => {
    return new Promise((resolve, reject) => {
      // Say there is a request performed
      ongoing_request.isOpen = true;
      ongoing_request.model_Name = model_Name;
      //  Ask who is alive
      con.publish(topic, "ASK");
      setTimeout(() => {
      // If no answer after given timeout
        if (ongoing_request.passports.length == 0) {
          reject({ error: "No MQTT passports found" });
      // Else send a deep clone of the answers (else it's empty)
        } else {
          resolve(JSON.parse(JSON.stringify(ongoing_request.passports)));
        }
      // Delete the current request object and 'close it'
        ongoing_request.passports.length = 0;
        ongoing_request.isOpen = false;
        ongoing_request.model_Name = ""
      }, process.env.mqtt_timeout || 2000)
    })
  }
};

这里是 MQTT 监听器：

con.on("message", (topic, message) => {
      // If a passport is received check the topic and if there is a request opened
      if (_checkTopic(topic) && ongoing_request.isOpen) {
        try {
          ongoing_request.passports.push(JSON.parse(message));
        } catch (error) {
      // do stuff if error
        }
      }
  }
})

我知道问题来自我用来指定是否有正在进行的请求的布尔值，我正在考虑为每个新请求创建一个对象并通过唯一 ID（如时间戳）识别它们，但我有没有办法让 MQTT 侦听器知道这个唯一的 id。

我有一些其他的解决方案，但我不确定它们是否会起作用，我觉得有一种我不知道的方法可以很好地处理这个问题。

祝你有美好的一天。

【问题讨论】：

标签： node.js http asynchronous microservices mqtt

【解决方案1】：

您需要为每个请求生成一个唯一的 id 并将其包含在 MQTT 消息中，然后您可以缓存由唯一 id 键入的 Express 响应对象。

设备需要在其响应中包含唯一 ID，以便它们可以与正确的响应配对。

另一种方法是缓存来自设备的响应并为缓存分配一个生存时间，这样您就不需要每次都询问设备。

【讨论】：