python - 如何使用pyzmq正确发布和订阅最新消息？

Question

我有一个进程A不断发布消息，进程B和C订阅主题，获取进程A中发布者发布的最新消息。

因此，我将 zmq.CONFLATE 设置为发布者和订阅者。但是，我发现有一位订阅者收不到消息。

def publisher(sleep_time=1.0, port="5556"):

    context = zmq.Context()
    socket = context.socket(zmq.PUB)
    socket.setsockopt(zmq.CONFLATE, 1)
    socket.bind("tcp://*:%s" % port)
    print ("Running publisher on port: ", port)

    while True:
        localtime = time.asctime( time.localtime(time.time()))
        string = "Message published time: {}".format(localtime)
        socket.send_string("{}".format(string))
        time.sleep(sleep_time)

def subscriber(name="sub", sleep_time=1, ports="5556"):

    print ("Subscriber Name: {}, Sleep Time: {}, Port: {}".format(name, sleep_time, ports))

    context = zmq.Context()
    print ("Connecting to publisher with ports %s" % ports)
    socket = context.socket(zmq.SUB)
    socket.setsockopt(zmq.CONFLATE, 1)
    socket.setsockopt_string(zmq.SUBSCRIBE, "")
    socket.connect ("tcp://localhost:%s" % ports)

    while True:

        message = socket.recv()
        localtime = time.asctime( time.localtime(time.time()))
        print ("\nSubscriber [{}]\n[RECV]: {} at [TIME]: {}".format(name, message, localtime))
        time.sleep(sleep_time)


if __name__ == "__main__":
    Process(target=publisher).start()
    Process(target=subscriber, args=("SUB1", 1.2, )).start()
    Process(target=subscriber, args=("SUB2", 1.1, )).start()

我试图取消设置发布者中的 socket.setsockopt(zmq.CONFLATE, 1)，这似乎解决了问题。进程B和C中的订阅者都能收到消息，而且消息好像是最新的。

我试图找出为什么使用 CONFLATE 设置发布者会导致我遇到的问题。我找不到有关它的信息。有谁知道是什么导致了这种行为？

另外，我想知道，在一个发布者对多个订阅者的情况下，正确的代码设置是怎样的，才能让订阅者始终获得最新消息？

Answer 1

这很可能是时间问题，ZMQ_CONFLATE 套接字选项将入站 和出站队列限制为 1 条消息。

PUB/SUB 的工作方式是订阅者在设置 ZMQ_SUBSCRIBE 选项时向发布者发送订阅消息。如果您同时启动两个订阅者，则到达发布者队列的订阅消息之一可能会被丢弃。

尝试在开始每个订阅者之间添加一个 sleep 。

来自 zeromq 文档

If set, a socket shall keep only one message in its inbound/outbound queue, this message being the last message received/the last message to be sent. Ignores ZMQ_RCVHWM and ZMQ_SNDHWM options. Does not support multi-part messages, in particular, only one part of it is kept in the socket internal queue.

我并不是说这是您问题的解决方案，但如果是这种情况，我们可能需要发布对 libzmq 的更改以使合并选项更细化，以便您可以选择是否应将合并应用于入站或出站队列。