ZeroMQ Pub/Sub action 队列中的最后一个元素和其他元素

Question

我开始使用带有Publisher/Subscriber 引用的python 使用zeromq。但是，我没有找到任何有关如何处理队列中消息的文档。我想将最后收到的消息与队列的其余元素不同。

示例

publisher.py

import zmq
import random
import time

port = "5556"
topic = "1"

context = zmq.Context()
socket = context.socket(zmq.PUB)
socket.bind("tcp://*:%s" % port)

while True:
    messagedata = random.randrange(1,215)
    print "%s %d" % (topic, messagedata)
    socket.send("%s %d" % (topic, messagedata))
    time.sleep(.2)

订阅者.py

import zmq

port = "5556"
topic = "1"

context = zmq.Context()
socket = context.socket(zmq.SUB)

print "Connecting..."
socket.connect ("tcp://localhost:%s" % port)
socket.setsockopt(zmq.SUBSCRIBE,topic)

while True:
    if isLastMessage(): # probably based on socket.recv()
         analysis_function() # time consuming function
    else:
         simple_function()  # something simple like print and save in memory

我只想知道如何创建subscriber.py 文件中描述的isLastMessage() 函数。如果直接在 zeromq 中有内容或解决方法。

Answer 1

欢迎来到非阻塞消息/信号的世界

这是任何严肃的分布式系统设计的基本特征。

如果您通过管道中没有另一个消息来假设“最后一个”消息，那么 Poller() 实例可能有助于您的主要事件循环，您可以在其中控制“等待”的时间量-a-在考虑管道“空”之前，不要用零等待旋转循环破坏您的 IO 资源。

显式信号总是更好^{（如果你可以设计远程端行为）}

接收方有零知识，接收到的“最后”消息的上下文是什么（建议从消息发送方广播明确的信令），但是有一个相反的功能为此——指示 ZeroMQ 原型“在内部”丢弃所有此类消息，这些消息不是“最后一个”消息，从而减少接收方处理以正确处理“最后一个”消息。

aQuoteStreamMESSAGE.setsockopt( zmq.CONFLATE, 1 )

如果您想阅读有关 ZeroMQ 模式和反模式的更多信息，请不要错过 Pieter HINTJENS 的精彩书籍“Code Connected, Volume 1”（也有 pdf 格式），并且可能希望使用 @ 更广泛地了解 distributed-computing 987654322@

Answer 2

如果isLastMessage() 用于识别publisher.py 生成的消息流中的最后一条消息，那么这是不可能的，因为没有最后一条消息。 publisher.py 产生无限量的消息！

但是，如果publisher.py 知道它最后的“真实”消息，即没有while True:，它可以在之后发送“我完成了”消息。在subscriber.py 中识别它是微不足道的。

Answer 3

对不起，我会保留这个问题以供参考。我刚刚找到了答案，在文档中有一个 NOBLOCK 标志，您可以将其添加到接收器中。有了这个recv 命令不会阻塞。从answer 的一部分中提取的一个简单的解决方法如下：

while True:
    try:
        #check for a message, this will not block
        message = socket.recv(flags=zmq.NOBLOCK)

        #a message has been received
        print "Message received:", message

    except zmq.Again as e:
        print "No message received yet"

至于真正的实现，不确定它是否是您使用标志NOBLOCK 的最后一次调用，并且一旦您进入exception 块。 Wich 翻译成如下内容：

msg = subscribe(in_socket)
is_last = False
while True:
    if is_last:
        msg = subscribe(in_socket)
        is_last = False
    else:
        try:
            old_msg = msg
            msg = subscribe(in_socket,flags=zmq.NOBLOCK)
            # if new message was received, then process the old message
            process_not_last(old_msg)
        except zmq.Again as e:
            process_last(msg)
            is_last = True  # it is probably the last message