我正试图使用ZMQ构建一个多发布者/多订阅者拓扑。我创建了一个使用espresso.py示例的例子，对它做了一些轻微的修改。我想确保我所做的是正确的，因为我对zeromq还很陌生。请随意评论。
我基本上已经记下了一些教训。
-
zmq套接字只能跨多个进程绑定到一个端口，绑定到单个网卡(也称为常规套接字)
-
绑定并不意味着监听，即您可以在绑定后发出connect()(对于套接字开发人员来说非常困惑，但嘿，这不是套接字)
-
Proxy和XPUB/XSUB旨在使用s模式，此时订阅者不必计算并连接到所有发布者。
我真正不喜欢下面代码的是每个订阅者绑定到一个单独的套接字。虽然这是一种必要的邪恶，但不知何故，我一直认为这看起来不对劲。
这是我的示例代码。
# Espresso Pattern
# This shows how to capture data using a pub-sub proxy
#
import time
from random import randint
from string import uppercase
from threading import Thread
import zmq
from zmq.devices import monitored_queue
from zhelpers import zpipe
# The subscriber thread requests messages starting with
# A and B, then reads and counts incoming messages.
def subscriber_thread():
ctx = zmq.Context.instance()
# Subscribe to "A" and "B"
subscriber = ctx.socket(zmq.SUB)
subscriber.connect("tcp://localhost:6001")
subscriber.setsockopt(zmq.SUBSCRIBE, b"A")
subscriber.setsockopt(zmq.SUBSCRIBE, b"B")
count = 0
while True:
try:
msg = subscriber.recv_multipart()
except zmq.ZMQError as e:
if e.errno == zmq.ETERM:
break # Interrupted
else:
raise
count += 1
print ("Subscriber received %d messages" % count)
# .split publisher thread
# The publisher sends random messages starting with A-J:
def publisher_thread(port, char):
ctx = zmq.Context.instance()
publisher = ctx.socket(zmq.PUB)
publisher.bind("tcp://*:"+str(port))
while True:
string = "%s-%05d" % (char, randint(port, port+500))
try:
publisher.send(string)
except zmq.ZMQError as e:
if e.errno == zmq.ETERM:
break # Interrupted
else:
raise
time.sleep(0.1) # Wait for 1/10th second
# .split listener thread
# The listener receives all messages flowing through the proxy, on its
# pipe. Here, the pipe is a pair of ZMQ_PAIR sockets that connects
# attached child threads via inproc. In other languages your mileage may vary:
def listener_thread(pipe):
# Print everything that arrives on pipe
while True:
try:
print (pipe.recv_multipart())
except zmq.ZMQError as e:
if e.errno == zmq.ETERM:
break # Interrupted
# .split main thread
# The main task starts the subscriber and publisher, and then sets
# itself up as a listening proxy. The listener runs as a child thread:
def main():
# Start child threads
ctx = zmq.Context.instance()
p_thread1 = Thread(target=publisher_thread, args=(6000,'A'))
p_thread2 = Thread(target=publisher_thread, args=(7000,'B'))
s_thread = Thread(target=subscriber_thread)
p_thread1.start()
p_thread2.start()
s_thread.start()
pipe = zpipe(ctx)
subscriber = ctx.socket(zmq.XSUB)
subscriber.connect("tcp://localhost:6000")
subscriber.connect("tcp://localhost:7000")
publisher = ctx.socket(zmq.XPUB)
publisher.bind("tcp://*:6001")
l_thread = Thread(target=listener_thread, args=(pipe[1],))
l_thread.start()
try:
monitored_queue(subscriber, publisher, pipe[0], 'pub', 'sub')
except KeyboardInterrupt:
print ("Interrupted")
del subscriber, publisher, pipe
ctx.term()
if __name__ == '__main__':
main()
我在ZeroMQ github页面上提出了一个问题，并得到了响应。这是ZeroMQ中的一个已知错误，是由于发布和订阅发生在不同的线程中，这些线程在订阅消息的接收方完全准备好之前就提出了订阅请求。更多详细信息可以在这里找到。
https://github.com/zeromq/libzmq/issues/897
我试着在这里模拟的问题
https://gist.github.com/vivekfantain/9021979
为任何在同一问题上遇到麻烦的人分享这一切。