消息队列体系结构(客户端到 Web 服务器到工作线程并返回) [英] Message queue architecture (client to web server to worker and back)

问题描述

我有一个用 Node JS 编写的 Web 服务器，在 Heroku 上运行.服务器有一个 web 服务器进程和一个工作进程.Web 服务器通过 RabbitMQ 队列成功地向 worker 发送消息；worker 成功地将处理后的数据返回给 web 服务器.我使用随机生成的 Uuid 来跟踪消息并确保正确的消息与正确的原始消息配对.

I have a web server written in Node JS running on Heroku. The server has a web server process and a worker process. The web server successfully sends messages to the worker via a RabbitMQ queue; the worker successfully returns the processed data to the web server. I use a randomly generated Uuid to track the messages and ensure the right message is paired up with the right original message.

在一个单独的项目中，我让客户端(网站)与 Web 服务器成功通信.现在，我需要将两者放在一起.

In a separate project, I have the client (website) successfully communicating with the web server. Now, I need to put the two together.

我怎样才能做到这一点:

How can I make it so:

1. 客户端发送 HTTP POST.
2. Web 服务器接收请求并将请求传递到消息队列中.
3. Worker 处理请求并返回到 Web 服务器.
4. Web 服务器将正确的数据返回给客户端，以便客户端可以将响应与原始请求相关联.

第 4 步是我卡住的地方.我想我在某处读到客户端应该不断轮询(HTTP POST?)Web 服务器，直到它的数据准备就绪.我想我需要在第 1 步之后回复客户端，这样请求就不会超时.任何想法/建议表示赞赏！

Step 4 is where I'm stuck. I think I read somewhere that the client should continuously poll (HTTP POSTs?) the web server until it's data is ready. I think I need to respond to the client after step 1 so the request doesn't time out. Any ideas/advice appreciated!

框图:

推荐答案

您需要做的简短版本是双向消息传递.您的 Web 应用程序需要是消息生产者和消息消费者.您的后端服务也是如此.

The short version of what you need to do is two-way messaging. Your web app needs to be a message producer and a message consumer. The same goes for your back-end service.

当 HTTP 请求进来时，Web 服务器通过 RabbitMQ 发送消息.后端会在未来的某个时间点将其取出.同时，Web 服务器通过 HTTP 请求发回响应，说明发生了一些事情，稍后会通知用户.

When the HTTP request comes in, the web server sends a message through RabbitMQ. The back-end picks it up at some point in the future. Meanwhile, the web server sends a response back through the HTTP request saying something is happening and the user will be notified later.

如果您使用 express，它看起来像这样:

If you're using express, it would look something like this:

var router = express.Router();
router.post("/", postJob);

function postJob(req, res, next){
  req.session.inProgress = true;

  var msg = {
    job: "do some work"
  };

  jobSender.sendJobRequest(msg, function(err){
    if (err) { return next(err); }
    res.render("some-response");
  });
}

这段代码做了很多假设，比如 jobSender 是某种封装的对象，具有通过 RabbitMQ 发送消息的方法.我相信你可以根据你已经说过的内容填写发送消息的详细信息.

This code makes a lot of assumptions, like jobSender being some kind of encapsulated object with a method to send a message across RabbitMQ. I'm sure you can fill in the details of sending a message, based on what you've said already.

重要的是，HTTP 请求处理程序通过 RabbitMQ 发送消息，然后将 HTTP 响应发送回 Web 浏览器.

The important thing, is that the HTTP request handler sends the message across RabbitMQ and then sends an HTTP response back to the web browser.

此时，浏览器可以做任何它需要做的事情.

At this point, the browser can do whatever it needs to do.

在后端，当另一个服务完成它的工作时，它需要做两件事之一:

On the back-end, when the other service has completed it's work, it will need to do one of two things:

1) 在某处更新共享数据库，以便您的 Web 服务器知道已经完成了哪些工作(并且可以读取状态)

1) update a shared database somewhere, so that your web server knows what work has been done (and can read a status)

或

2) 通过rabbitmq 将消息发送回Web 服务器

2) send a message back to the web server, via rabbitmq

由于各种原因，选项 #1 可能并不总是一个好的选择.从你的问题来看，无论如何你都想要选项#2.

option #1 might not always be a good option, for various reasons. and from your question, you want option #2 anyways.

您将需要第二个队列 - Web 服务器正在侦听的队列.当 Web 服务器收到来自该队列的消息时，它将使用接收到的状态更新它自己的数据库.

you'll need a second queue - one that the web server is listening to. when the web server receives a message from this queue, it will update it's own database with the status that is receives.

此状态可能是完成"或进行中"或错误"或您认为合适的其他状态.

this status may be "complete" or "in progress" or "error" or something else that you see fit.

例如，如果您有作业状态"消息，则可能有一个名为JobStatusReceiver"的抽象来接收状态消息.

For example, if you have a "job status" message, you may have an abstraction called "JobStatusReceiver" to receive the status message.

像这样的简单模块可以从您的作业状态队列接收消息，并使用状态更新本地数据库

A simple module like this could receive messages from your job status queue, and update a local database with the status

var JobStatusReceiver = require("./jobStatusReceiver");
var someDataObject = require("someDataObject");

var jobStatus = {

  listen: function(){
    var receiver = new JobStatusReceiver();
    receiver.receive(function(statusMessage){

      someDataObject.status = statusMessage.status;
      someDataObject.data = statusMessage.data;


      someDataObject.save();
    });
  }

};

module.exports = jobStatus;

请注意，这可能发生在 Web 服务器中，但它不是 HTTP 请求的一部分.消息通过 RabbitMQ 和 JobStatusReceiver 传入，而不是 HTTP 请求的一部分.

note that this may be happening in the web server, but it's not part of an HTTP request. the message came in through RabbitMQ with teh JobStatusReceiver, and not part of an HTTP request.

someDataObject 对象很可能是来自 yoru 数据库的对象，因此可以将其保存回数据库.

the someDataObject object is most likely going to be an object from yoru database, so it can be saved back to the database.

最后，您需要通过数据通知用户已完成的操作的部分可以通过多种方式进行.

Lastly, the part where you need to notify the user of the completed action, with the data, can happen in a number of ways.

一般来说，每隔几秒就对 Web 服务器上的 HTTP API 进行一次 AJAX 调用以查找有效响应相当容易.

Generally speaking, it's fairly easy to make an AJAX call to an HTTP API on your web server every few seconds, to look for a valid response.

在浏览器方面，这可能很简单:

On the browser side, this could be as simple as:

var timer = setInterval(function(){

  $.ajax({
    url: "/api/check-status",
    success: function(data){
      if (data.complete){
        clearInterval(timer);
        doSomeOtherWork(data);
      }
    })
  });

});

再次在 Express 应用程序中，处理/api/check-status"，您将使用相同的someDataObject"模型来检查状态:

and again in the Express app, handling the "/api/check-status", you would use the same "someDataObject" model to check the status:

var someDataObject = require("someDataObject");

var router = new express.Router();
router.get("/", checkStatus);

function checkStatus(req, res, next){

  someDataObject.load(function(err, someData){
    if (err) { return next(err); }

    res.json({
      complete: (someData.status === "complete"),
      data: someData.data
    });

  });

}

这应该会让你走上正确的道路.当然，我遗漏了很多细节，但希望您能够填补缺失的部分.

This should hopefully get you down the right path. There are a lot of details I've left out, of course, but hopefully you'll be able to fill in the missing parts.

...

PS:我在我的RabbitMQ 4 Devs 培训课程.这是一个完整的包，用于启动和运行 RabbitMQ 和 Node.js

P.S.: I cover all of this, except for the browser checking for status updates on a timer, in my RabbitMQ 4 Devs training course. It's a complete package to get up and running with RabbitMQ and Node.js

这篇关于消息队列体系结构(客户端到 Web 服务器到工作线程并返回)的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！