【发布时间】：2021-11-27 12:41:01
【问题描述】：
我使用 ZeroMQ 编写了一个玩具 Master/Worker”或“task farm”。
这是我到目前为止所得到的 - 但我想添加 PUB/SUB，以便工作人员收听并响应主题（特定主题或通配符匹配）。
主人
#!/usr/bin/env python
from __future__ import print_function
import random
import time
from multiprocessing import Pool, Process
import zmq
from zmq.devices.basedevice import ProcessDevice
REQ_ADDRESS = 'tcp://127.0.0.1:6240'
REP_ADDRESS = 'tcp://127.0.0.1:6241'
if __name__ == '__main__':
# Start queue
context = zmq.Context()
sock_in = context.socket(zmq.ROUTER)
sock_in.bind(REQ_ADDRESS)
sock_out = context.socket(zmq.DEALER)
sock_out.bind(REP_ADDRESS)
zmq.device(zmq.QUEUE, sock_in, sock_out)
工人
#!/usr/bin/env python
from __future__ import print_function
import random
import time
import zmq
REP_ADDRESS = 'tcp://127.0.0.1:6241'
def receive_tasks():
"""
Client action: request tasks
"""
# ID: just to show that we're getting the right replies
my_id = random.randint(1, 1000000)
context = zmq.Context()
socket = context.socket(zmq.REP)
socket.connect(REP_ADDRESS)
while True:
# Data is received here. Note that this blocks until
# we get a job.
job = socket.recv_json()
# Do work here
time.sleep(0.5)
# Send the result back. Pass any JSON-serializable object.
socket.send_json([my_id, job['id'], job['task_id']])
if __name__ == '__main__':
receive_tasks()
客户
#!/usr/bin/env python
from __future__ import print_function
import random
import zmq
from zmq.core.poll import select
REQ_ADDRESS = 'tcp://127.0.0.1:6240'
def request_tasks():
"""
Client action: request tasks
"""
# ID: just to show that we're getting the right replies
my_id = random.randint(1, 1000000)
context = zmq.Context()
socket = context.socket(zmq.REQ)
socket.connect(REQ_ADDRESS)
for i in xrange(100):
job = {'id': my_id, 'task_id': random.randint(1, 100)}
socket.send_json(job)
# Selects the sockets that have READ, WRITE, and ERROR
# events respectively within the lists, with timeout 5.
# Same API as: http://docs.python.org/library/select.html
(rlist, wlist, xlist) = select([socket], [], [], 5)
if len(rlist) > 0:
# This receives the reply and deserializes it from JSON.
msg = socket.recv_json()
print('Client {0}, task #{1}: received work from {2} (for: {3})'.format(
my_id, i+1, msg[0], msg[1]))
else:
print('Client {0}, task#{1}: error, timeout reached.'.format(my_id,
i+1))
socket.close()
socket = context.socket(zmq.REQ)
socket.connect(REQ_ADDRESS)
if __name__ == '__main__':
request_tasks()
我的问题是：如何修改 master 和 workers 以“了解 TOPIC” - 使用 PUB/SUB？
注意：虽然我的示例代码是用 Python 编写的，并且图像插图指的是 Java - 我实际上是用 C++ 编写我的真实代码，所以请（如果可能）不要在你的回答中使用任何语言细节。
【问题讨论】：
-
您对此模式有任何经过身份验证的参考吗？如果是这样，请将其添加到标签的指导中。
-
@GertArnold 它被称为 Master/Worker（以前的 Master/Slave），我创建了标签
master-worker，因为它以前不存在 -
我知道你创造了它，这就是我问的原因。创建标签是有责任的。