javascript - 如何让 ZMQ 路由器在繁忙时引发错误？

Question

我有一个 REQ -> ROUTER -> [DEALER,DEALER...DEALER] 设置，其中 REQ 是客户端，ROUTER 是队列，DEALER 套接字是处理的工作人员数据并将其发送回 ROUTER，ROUTER 将其发送回 REQ。当有足够的经销商来处理工作时，工作就很好。但如果我放慢经销商的速度，路由器将永远不会告诉我它的工作量超出了它的处理能力。

文档说:

ROUTER sockets do have a somewhat brutal way of dealing with messages they can't send anywhere: they drop them silently. It's an attitude that makes sense in working code, but it makes debugging hard. The "send identity as first frame" approach is tricky enough that we often get this wrong when we're learning, and the ROUTER's stony silence when we mess up isn't very constructive.

Since ØMQ v3.2 there's a socket option you can set to catch this error: ZMQ_ROUTER_MANDATORY. Set that on the ROUTER socket and then when you provide an unroutable identity on a send call, the socket will signal an EHOSTUNREACH error.

老实说，我不确定这是否与我遇到的问题相同。冷漠的沉默确实符合我所看到的。

这是设置代码:

var argsToString, buildSocket, client, q;

buildSocket = function(desc, socketType, port) {
  var socket;
  log("creating socket: " + (argsToString(Array.apply(null, arguments))));
  socket = zmq.socket(socketType);
  socket.identity = "" + desc + "-" + socketType + "-" + process.pid + "-" + port;
  return socket;
};

argsToString = function(a) {
  return a.join(', ');
};

client = buildSocket("client", 'req', clientPort);

q = buildSocket("q", "router", qPort);

q.setsockopt(zmq.ZMQ_ROUTER_MANDATORY, 1);

q.on('error', function() {
  return log('router error ' + argsToString(Array.apply(null, arguments)));
});

如果需要，我可以发布更多代码。问题是，当 REQ 套接字每秒发送 10 条消息，但 DEALER 需要 2 秒才能完成工作时，ROUTER 只是忽略传入消息，而不管 ZMQ_ROUTER_MANDATORY 是多少。我已经发送了 1000 条消息，但从未见过任何套接字抛出错误 (.on 'error')。

有人谈论 ZMQ_HWM，但 Node 驱动程序似乎不支持 DEALER 或 ROUTER。

如何管理没有空间发送消息的路由器？

Answer 1

首先，如果您正在实现特定模式(据我所知，从您之前的问题来看，您正在实现 Paranoid Pirate)，那么这样说总是有帮助的，因为它将为您的代码提供上下文.

偏执海盗中没有专门解决您所讨论的问题。您可以通过跳到 the Titanic pattern 的指南来看到这一点。 ...当您处理偶发的连接问题，或者在您的情况下，由于您的工作人员在新消息到达时仍在工作而导致偶发的可用性时，您必须维护队列中工作人员的状态知道您需要如何处理该消息...要么将其发送给可用的工作人员，要么将其存储在某个地方，以便当工作人员可用时，您可以将其拉出并发送。

如果您尽可能严格地执行此操作，则会破坏 ZMQ 的“队列”性质，但您可以避免 HWM 中固有的不确定性，即会丢弃消息，而不是使系统崩溃.

您可以维护一个缓冲区，继续向队列添加消息，直到您意识到您已进入 HWM 的 40%(这取决于消息的大小)...这将为您提供在开始保存消息之前先缓冲，但最终过程是相同的。

这是 ZMQ 将责任转移给应用程序设计者的一个领域，因为没有单一的“正确”方法可以适用于所有场景。

<小时/>

编辑回应评论:

以下是我在 Node.js 中处理此问题的基本要点:

var worker_count = 0;
var job_count = 0;

// ...

q.on('message', function() {
    // ...

    if (msg.toString() == 'ready') worker_count++;
    else job_count--;

    // ...

    // this could use some TLC, but here's the basic gist of the logic...
    if (job_count >= worker_count) {
        // we'll assume the message was cached when it was received from
        // the req socket, if so nothing else to do here
    }
    else {
        // figure out if there is a cached message ready to go, if so, then...
        q.send(job);
        job_count++;
    }
});