如何让 ROUTER -> DEALER 回显？

Question

我正在尝试构建Paranoid Pirate Pattern 的后半部分，这是一个将工作发送到一组 DEALER 节点的 ROUTER（我可能误解了图表）。现在我只想让经销商回应工作，或者只是发回一条“完成”的消息。问题是工作节点（DEALER）从不接收任何消息。

var buildSocket, connectionTemplate, delay, frontPort, log, q, qPort, worker, workerPort, zmq;

zmq = require("zmq");    
frontPort = 5000;    
qPort = 5100;    
workerPort = 5200;    
connectionTemplate = "tcp://127.0.0.1:";    
log = console.log;    
debugger;    
delay = process.argv[2] || 1000;

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

q = buildSocket('q_output', 'router', qPort);

worker = buildSocket('worker', 'dealer', workerPort);

worker.bind("" + connectionTemplate + workerPort);

q.connect("" + connectionTemplate + workerPort);

q.on('message', function() {
  var args;
  args = Array.apply(null, arguments);
  log('queue received ' + JSON.stringify(arguments));
  return worker.send('work done');
});

worker.on('message', function() {
  var args;
  log('back received ' + JSON.stringify(arguments));
  args = Array.apply(null, arguments);
  return q.send(args);
});

setInterval((function() {
  var value;
  value = Math.floor(Math.random() * 100);
  console.log(q.identity + ": sending " + value);
  q.send(value);
}), delay);

队列和工作线程on 'message' 事件永远不会触发。我理解的方式是您设置 ROUTER 节点，将其绑定到端口（用于返回消息），设置 DEALER 节点并将它们绑定到端口，然后将 ROUTER 连接到 DEALER 端口并开始发送消息。在实践中，消息被发送但从未被接收：

creating router socket
creating dealer socket
q_output-router-60326-5100: sending 30
q_output-router-60326-5100: sending 25
q_output-router-60326-5100: sending 65
q_output-router-60326-5100: sending 68
q_output-router-60326-5100: sending 50
q_output-router-60326-5100: sending 88

Answer 1

你的事情有点倒退，在这里。将DEALER 套接字视为修改后的REQ 套接字......它应该向路由器发起消息。 ROUTER 套接字更像是修改后的REP 套接字......它应该响应您的经销商发送的初始请求。

您严格不需要使用ROUTER/DEALER 配对来遵循该模式...但它确实让事情更加更容易，所以你应该坚持下去你正在学习。

你的代码让我印象深刻的第二件事是你的消息处理程序，你有错误的套接字发送消息。

以这段代码为例（直接复制，不做修改）：

q.on('message', function() {
  var args;
  args = Array.apply(null, arguments);
  log('queue received ' + JSON.stringify(arguments));
  return worker.send('work done');
});

...上面写着（伪代码）：

when `q` receives a message from `worker`
  print out the message we received
  now have `worker` send *another* message that says "work done"

你想要的是更像下面的东西（简化）：

var zmq = require("zmq");
var q = zmq.socket('router');
var worker = zmq.socket('dealer');

// I've switched it so the router is binding and the worker is connecting
// this is somewhat arbitrary, but generally I'd consider workers to be less
// reliable, more transient, and also more numerous. I'd think of the queue
// as the "server"

// I've used bindSync, which is synchronous, but that's typically OK in the
// startup phase of a script, and it simplifies things.  If you're spinning
// up new sockets in the middle of your script, using the async bind()
// is more appropriate
q.bindSync('tcp://127.0.0.1:5200');

worker.connect('tcp://127.0.0.1:5200');

q.on('message', function() {
  var args;
  args = Array.apply(null, arguments);
  log('queue received ' + JSON.stringify(arguments));

  // if we've received a message at our queue, we know the worker is ready for
  // more work, so we ready some new data, regardless of whether we
  // received work back
  var value = Math.floor(Math.random() * 100);

  // note that the socket that received the message is responding back
  if (args[1].toString() == 'ready') console.log('worker now online');
  else console.log('work received: '+args[1].toString());

  // we need to specify the "ID" of our worker as the first frame of
  // the message
  q.send([args[0], value]);

  // we don't need to return anything, the return value of a
  // callback is ignored
});

worker.on('message', function() {
  var args;
  log('back received ' + JSON.stringify(arguments));
  args = Array.apply(null, arguments);

  // we're just echoing back the "work" we received from the queue
  // for additional "workiness", we wait somewhere between 10-1000
  // milliseconds to respond
  setTimeout(function() {
    worker.send(args[0].toString());
  }, parseInt(args[0].toString())*10);
});

setTimeout((function() {
  var value;
  console.log('WORKER STARTING UP');

  // the worker starts the communication, indicating it's ready
  // rather than the queue just blindly sending work
  worker.send('ready'); // sending the first message, which we catch above
}), 500); // In my experience, half a second is more than enough, YMMV

...如您所见，模式是：

1. 工作人员表示准备就绪
2. 队列发送可用工作
3. 工人完成工作并返回
4. 队列接收完成的工作并发回更多工作
5. 转到 3