pyzmq非阻塞套接字

你可以同时使用 zmq.Poller (你可以在zguide repo中找到很多例子,例如 rrbroker.py )或者 gevent-zeromq 实施( code sample )。

公认答案中提供的例子给出了它的要点,但您也可以通过使用 zmq.device 对于代理,同时在其他方面坚持来自的“Extended Request Reply”模式 the guide 因此,服务器的hello-worldy示例可能如下所示:

import time
import threading
import zmq

context = zmq.Context()

def worker():
    socket = context.socket(zmq.REP)
    socket.connect('inproc://workers')
    while True:
        msg = socket.recv_string()
        print(f'Received request: [{msg}]')
        time.sleep(1)
        socket.send_string(msg)

url_client = 'tcp://*:5556'
clients = context.socket(zmq.ROUTER)
clients.bind(url_client)
workers = context.socket(zmq.DEALER)
workers.bind('inproc://workers')

for _ in range(4):
    thread = threading.Thread(target=worker)
    thread.start()

zmq.device(zmq.QUEUE, clients, workers)

在这里,我们让四个工作人员并行处理传入请求。现在,您在客户端使用Node,但为了保持示例的完整性,可以使用下面的Python客户端来查看这一点。在这里,我们创建了10个请求,然后将分3批处理:

import zmq
import threading

context = zmq.Context()

def make_request(a):
    socket = context.socket(zmq.REQ)
    socket.connect('tcp://localhost:5556')
    print(f'Sending request {a} ...')
    socket.send_string(str(a))
    message = socket.recv_string()
    print(f'Received reply from request {a} [{message}]')

for a in range(10):
    thread = threading.Thread(target=make_request, args=(a,))
    thread.start()