ZeroMQ 使用 XPUB/XSUB 的多个发布者和订阅者 - 这是一个正确的实现吗？

Question

我正在尝试使用 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()

Answer 1

我在 ZeroMQ github 页面上提出了一个问题并得到了回复。这是 ZeroMQ 中的一个已知错误，这是由于发布和订阅发生在不同的线程中，这些线程在订阅消息的接收者完全准备好之前提出订阅请求。更多细节可以在这里找到。

https://github.com/zeromq/libzmq/issues/897

我试图在这里模拟问题

https://gist.github.com/vivekfantain/9021979

为遇到同样问题的其他人分享所有这些。