ZMQ在PUB-SUB中的延迟（慢订阅者）

Question

我在网上找到很多类似的问题，但都没能帮我解决我的困扰。

我使用的是：

Linux Ubuntu 14.04
python 3.4
zmq : 4.0.4 // pyZMQ 14.3.1

简而言之

在ZMQ的订阅（SUB）套接字中，接收队列不断增长，即使我设置了高水位标记（HWM）。这种情况发生在订阅者的速度比发布者慢的时候。我该怎么做才能防止这种情况呢？

背景

我在人与计算机交互的领域工作。我们有一个庞大的代码库来控制鼠标光标之类的东西。我想把它“拆分”成几个模块，通过ZMQ进行通信。我希望延迟尽可能小，但丢失消息并不是那么重要。

另一个有趣的方面是可以在节点之间添加“监视者”。因此，发布/订阅（PUB/SUB）套接字似乎是最合适的选择。

大致是这样的：

+----------+                +-----------+                 +------------+
|          | PUB            |           |  PUB            |            |
|  Input   | +----+------>  |  Filter   |  +----+------>  |   Output   |
|          |      |     SUB |           |       |     SUB |            |
+----------+      v         +-----------+       v         +------------+
               +-----+                       +-----+                   
               |Spy 1|                       |Spy 2|                   
               +-----+                       +-----+

问题

一切正常，除了当我们添加监视者的时候。如果我们添加一个进行“重负载”操作的监视者，比如用matplotlib进行实时可视化，我们会发现图表的延迟在增加。也就是说：在上面的图中，过滤和输出都很快，没有延迟，但在监视者2上，延迟可能在运行20分钟后达到10分钟（！！）

看起来接收方的队列在不断增长。我们研究了ZMQ的高水位标记（HWM）功能，试图把它设置得低一些以丢弃旧消息，但没有任何变化。

最小代码

架构：

+------------+                +-------------+
|            |  PUB           |             |
|   sender   | -------------> |  receiver   |
|            |             SUB|             |
+------------+                +-------------+

接收方是一个慢接收器（在第一个图中充当监视者）

代码：

Sender.py

import time
import zmq

ctx = zmq.Context()

sender = ctx.socket(zmq.PUB)
sender.setsockopt(zmq.SNDBUF, 256)
sender.set_hwm(10)
sender.bind('tcp://127.0.0.1:1500')

print(zmq.zmq_version()) ## 4.0.4
print(zmq.__version__) ## 14.3.1
print(sender.get_hwm()) ## 10

i = 0
while True:
    mess = "{} {}".format(i, time.time())
    sender.send_string(mess)
    print("Send : {}".format(mess))
    i+= 1

receiver.py:

import time
import zmq

ctx = zmq.Context()
front_end = ctx.socket(zmq.SUB)

front_end.set_hwm(1)
front_end.setsockopt(zmq.RCVBUF, 8)

front_end.setsockopt_string(zmq.SUBSCRIBE, '')
front_end.connect('tcp://127.0.0.1:1500')

print(zmq.zmq_version()) ## 4.0.4
print(zmq.__version__) ## 14.3.1
print(front_end.get_hwm()) ## 1

while True:
    mess = front_end.recv_string()
    i, t = mess.split(" ")
    mess = "{} {}".format(i, time.time() - float(t))
    print("received : {}".format(mess))
    time.sleep(1)  # slow

我觉得这不是ZMQ发布/订阅的正常行为。我尝试在接收器、订阅者和两者中设置HWM，但没有任何变化。

我遗漏了什么呢？

补充：

我觉得我在解释问题时没有说清楚。我实现了一个移动鼠标光标的功能。输入是以200Hz的频率通过ZMQ发送的鼠标光标位置（用.sleep( 1.0 / 200 )），经过一些处理后更新鼠标光标的位置（在我的最小示例中没有这个sleep）。

一切都很顺利，即使我启动了监视者。尽管如此，监视者的延迟还是在增加（因为处理速度慢）。这种延迟在光标的“管道”末尾并没有出现。

我认为问题出在慢速订阅者排队消息上。

在我的示例中，如果我们关闭发送者而让接收者继续运行，消息会继续显示，直到所有（？）提交的消息都被显示。

监视者正在绘制光标的位置以提供反馈，但有这么大的延迟还是很不方便……我只想获取最后发送的消息，这就是我尝试降低HWM的原因。

消息队列订阅者延迟 zmq pub-sub 高水位标记监视者实时可视化

ZMQ在PUB-SUB中的延迟（慢订阅者）

简而言之

背景

问题

最小代码

架构：

代码：

我遗漏了什么呢？

2 个回答

缺少更好的实时设计/验证

事实很重要。

撰写回答