消息队列体系结构(客户端到Web服务器到工作者和后面)

Question

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

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

我怎么能这样做:

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

第四步是我被困住的地方.我想我在某个地方读过客户端应该连续轮询(HTTP POSTs？)Web服务器,直到它的数据准备就绪.我想我需要在第1步之后回复客户端,因此请求不会超时.任何想法/建议表示赞赏!

框图:

Answer 1

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

当HTTP请求进入时,Web服务器通过RabbitMQ发送消息.后端在未来的某个时刻将其收录起来.同时,Web服务器通过HTTP请求发回响应,说明正在发生的事情,稍后将通知用户.

如果你使用快递,它看起来像这样:

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发送消息的方法.我相信你可以根据你已经说过的内容填写发送邮件的详细信息.

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

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

在后端,当其他服务完成它的工作时,它需要做以下两件事之一:

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

要么

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

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

你需要第二个队列 - 一个是Web服务器正在监听的队列.当Web服务器从此队列接收消息时,它将使用接收的状态更新其自己的数据库.

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

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

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

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请求的一部分.

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

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

一般来说,每隔几秒就可以很容易地在Web服务器上对HTTP API进行AJAX调用,以查找有效的响应.

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

var timer = setInterval(function(){

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

});

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

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
    });

  });

}

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

...

PS:在我的RabbitMQ 4 Devs培训课程中,除了浏览器检查计时器状态更新之外,我将覆盖所有这些内容.这是一个完整的包,可以使用RabbitMQ和Node.js启动和运行