asia 2011-03-24
之前已经讲过,zeroMQ是可以多对多的,但需要成对匹配才行,即多个发布端都是同一种模式,而这里要涉及到的是,多个发布端模式不统一的情况。
文中先给出了一个比较"脏"的处理方式:
import zmq import time context = zmq.Context() receiver = context.socket(zmq.PULL) receiver.connect("tcp://localhost:5557") subscriber = context.socket(zmq.SUB) subscriber.connect("tcp://localhost:5556") subscriber.setsockopt(zmq.SUBSCRIBE, "10001") while True: while True: try: rc = receiver.recv(zmq.NOBLOCK)#这是非阻塞模式 except zmq.ZMQError: break while True: try: rc = subscriber.recv(zmq.NOBLOCK) except zmq.ZMQError: break
显然,如此做既不优雅,还有出现单来源循环不止,另一来源又得不到响应的状况。
自然,官方也做了相应的封装,给了一个相对优雅的实现:
import zmq context = zmq.Context() receiver = context.socket(zmq.PULL) receiver.connect("tcp://localhost:5557") subscriber = context.socket(zmq.SUB) subscriber.connect("tcp://localhost:5556") subscriber.setsockopt(zmq.SUBSCRIBE, "10001") poller = zmq.Poller() poller.register(receiver, zmq.POLLIN) poller.register(subscriber, zmq.POLLIN) while True: socks = dict(poller.poll()) if receiver in socks and socks[receiver] == zmq.POLLIN: message = receiver.recv() if subscriber in socks and socks[subscriber] == zmq.POLLIN: message = subscriber.recv()
这种方式采用了平衡兼顾的原则,实现了类似于同一模式多发布端推送的"平衡队列"功能。
(未完待续)