REQ/REP &用于双向异步工作人员处理的经销商/路由器
REQ/REP &用于双向异步工作人员处理的经销商/路由器
提问于 2017-06-14 03:57:43
我正在学习ZeroMQ，刚刚看了一下教程和几个例子。我使用Node.js作为我的主要环境(最终用Python替换我的工作人员)。
试图解决如何创建一个完全异步的消息传递系统，该系统将允许我的API将任务(通过REQ套接字)推送到路由器，让经销商将消息传递给工作人员，处理消息并将其结果发送回我的客户端(这是一个Express路由)。
我相信这种模式会像这样工作(尚未测试或正确实现代码，因此请将其作为概念大纲)：
router.js
const zmq = require('zmq');;
const frontend = zmq.socket('router');
const backend = zmq.socket('dealer');
frontend.on('message', function() {
var args = Array.apply(null, arguments);
backend.send(args);
});
backend.on('message', function() {
var args = Array.apply(null, arguments);
frontend.send(args);
});
frontend.bindSync('tcp://*:5559');
backend.bindSync('tcp://*:5560');client.js
var zmq = require('zmq'),
var express = require('express');
var app = express();
app.post('send', function(req, res) {
var client = zmq.socket('req');
// listen for responses from the server
client.on('message', function(data) {
console.log(data);
client.close();
});
// connect to the server port
client.connect('tcp://0.0.0.0:5454');
client.send('Request from ' + process.id);
});
app.listen('80');worker.js
var zmq = require('zmq');
var server = zmq.socket('rep');
server.on('message', function(d){
server.send('Response from ' + process.id);
});
// bind to port 5454
server.bind('tcp://0.0.0.0:5454', function(err){
if (err){
console.error("something bad happened");
console.error( err.msg );
console.error( err.stack );
process.exit(0);
}
});我不完全理解的是，ROUTER/DEALER是否将处理将响应工作人员发送到正确的客户端。同样，在这种情况下，经销商处理公平排队，因为我希望我的工作均匀地分配给工人。
我的客户端可以分布在许多不同的盒子中(负载平衡器API服务器)，我的路由器将在它自己的服务器上，工作人员也将分布在多个盒子中。
回答 2
Stack Overflow用户
发布于 2017-06-14 22:49:33
对于将来阅读这篇文章的人来说，在我的进一步研究中，我无意中发现了Majordomo协议/模式。这正是我想要实现的。有关实现、优缺点的文档可以在这里阅读：https://rfc.zeromq.org/spec:18/MDP/。以下是代理实现：https://github.com/zeromq/majordomo
Stack Overflow用户
发布于 2017-06-14 14:31:59
好像我在使用DEALER/ROUTER，而我应该使用XREQ和XREP.
broker.js
var zmq = require('zmq');
var frontPort = 'tcp://127.0.0.1:5559';
var backPort = 'tcp://127.0.0.1:5560';
var frontSocket = zmq.socket('xrep');
var backSocket = zmq.socket('xreq');
frontSocket.identity = 'xrep_' + process.pid;
backSocket.identity = 'xreq_' + process.pid;
frontSocket.bind(frontPort, function (err) {
console.log('bound', frontPort);
});
frontSocket.on('message', function() {
//pass to back
console.log('router: sending to server', arguments[0].toString(), arguments[2].toString());
backSocket.send(Array.prototype.slice.call(arguments));
});
backSocket.bind(backPort, function (err) {
console.log('bound', backPort);
});
backSocket.on('message', function() {
//pass to front
console.log('dealer: sending to client', arguments[0].toString(), arguments[2].toString());
frontSocket.send(Array.prototype.slice.call(arguments));
});
console.log('Broker started...');worker.js
var zmq = require('zmq');
var socket = zmq.socket('rep');
socket.identity = 'worker_' + process.pid;
socket.on('message', function(data) {
console.log(socket.identity + ': received ' + data.toString());
socket.send(data * 2);
});
socket.connect('tcp://127.0.0.1:5560', function(err) {
if (err) throw err;
console.log('server connected!');
});
console.log('Worker started...');client.js
var zmq = require('zmq');
var socket = zmq.socket('req');
socket.identity = 'client_' + process.pid;
socket.on('message', function(data) {
console.log(socket.identity + ': answer data ' + data);
});
socket.connect('tcp://127.0.0.1:5559');
setInterval(function() {
var value = Math.floor(Math.random()*100);
console.log(socket.identity + ': asking ' + value);
socket.send(value);
}, 100);
console.log('Client started...');我仍然不确定在每个API入站请求上打开连接是否安全。
页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接：
https://stackoverflow.com/questions/44535162
复制相关文章
相似问题
腾讯云开发者
