javascript - NodeJS 事件发射器阻塞问题

Question

我有一个 Node 应用程序处理来自另一个应用程序的一些 ZeroMQ 事件，这些事件利用此处找到的 Node-ZMQ 绑定(bind):https://github.com/JustinTulloss/zeromq.node

我遇到的问题是事件中的操作之一需要很长时间才能处理，并且这似乎会阻止在此期间处理任何其他事件。尽管应用程序当前没有集群，但这样做只会增加一些线程，并不能真正解决问题。我想知道是否有一种方法允许这些异步调用在处理时不阻止其他传入请求，以及我如何实现它们。

这是我当前正在做的高度浓缩/人为的代码示例:

var zmq = require('zmq');
var zmqResponder = zmq.socket('rep');
var Client = require('node-rest-client').Client;
var client = new Client();

zmqResponder.on('message', function (msg, data) {
  var parsed = JSON.parse(msg);
  logging.info('ZMQ Request received: ' + parsed.event);
  switch (parsed.event) {
    case 'create':
        //Typically short running process, not an issue
    case 'update':
        //Long running process this is the issue
        serverRequest().then(function(response){
            zmqResponder.send(JSON.stringify(response));
        });
     }

});

function serverRequest(){
    var deferred = Q.defer();
      client.get(function (data, response) {
      if (response.statusCode !== 200) {
        deferred.reject(data.data);
      } else {
        deferred.resolve(data.data);
      }
  });
    return deferred.promise;
}

编辑** 这是代码的要点:https://gist.github.com/battlecow/cd0c2233e9f197ec0049

Answer 1

我想，通过评论线程，我已经确定了您的问题。 REQ/REP 具有严格的同步消息顺序保证...您必须接收-发送-接收-发送等。 REQ 必须以发送开始，REP 必须以接收开始。因此，您一次只能处理一条消息，因为您选择的套接字类型强制执行这一点。

如果您使用不同的非事件驱动语言，当您尝试连续发送或接收两次时，您可能会收到一条错误消息，告诉您做错了什么，但 Node 允许您这样做并将后续消息排队，直到按消息顺序轮到它们为止。

您想要将 REQ/REP 更改为 DEALER/ROUTER，它将按您期望的方式工作。您必须稍微更改 ROUTER 套接字的逻辑才能使其正确发送，但其他一切都应该工作相同。

<小时/>

使用已发布要点的相关部分的粗略示例代码:

var zmqResponder = zmq.socket('router');

zmqResponder.on('message', function (msg, data) {
    var peer_id = msg[0];
    var parsed = JSON.parse(msg[1]);
    switch (parsed.event) {
        case 'create':
            // build parsedResponse, then...
            zmqResponder.send([peer_id, JSON.stringify(parsedResponse)]);
            break;
    }
});

zmqResponder.bind('tcp://*:5668', function (err) {
    if (err) {
        logging.error(err);
    } else {
        logging.info("ZMQ awaiting orders on port 5668");
    }
});

...你需要获取peer_id(或者任何你想调用它的东西，在ZMQ命名法中，它是你发送的套接字的套接字ID，将其视为“地址”之类的)，然后使用 send 将其作为发回消息的第一帧。

顺便说一句，我刚刚注意到在你的要点中，你在同一个套接字上都使用 connect()-ing 和 bind()-ing (zmq.js 行分别为 52 和 143)。不要那样做。从其他线索推断，您只想在进程的这一侧进行 bind() 。