python - 理解ZMQ的HWM

Question

我在理解 ZeroMQ 高水位线 (HWM) 队列的工作原理时遇到一些问题。

我在下面附上了两个脚本，它们重现了以下内容。

• 建立 PUSH/PULL 连接，将所有 HWM 队列设置为 1。
• 让牵引车睡一会儿。
• 从推送器发送 2200 条消息。
• 当puller醒来后，接收2200条消息并打印出来。

我得到的结果是拉取器能够成功接收(打印)所有消息。此外，推送器似乎几乎立即完成执行。根据ZMQ official documentation我期望的是推送器不会在拉动器醒来之前完成执行，因为由于达到 HWM 而在第二次 send(...) 调用时被阻止。我还尝试在每次 send(...) 调用之间添加 0.001 秒的 sleep ，结果相同。

所以，我的问题是:

• 为什么在达到 HWM(大小 1)后，推送器在第二次调用 send(...) 时没有阻塞？
• 推送器和拉取器中存储的消息在哪里？
• HWM 大小与存储的消息数量之间是否存在直接关系？

脚本:

推杆.py

import zmq

context = zmq.Context()
push_socket = context.socket(zmq.PUSH)

push_socket.setsockopt(zmq.SNDHWM, 1)
push_socket.setsockopt(zmq.RCVHWM, 1)

push_socket.bind("tcp://127.0.0.1:5557")
print(push_socket.get_hwm()) # Prints 1
print('Sending all messages')

for i in range(2200):
    push_socket.send(str(i).encode('ascii'))

print('Finished execution...')

拉取器.py

import zmq
import time

context = zmq.Context()

pull_socket = context.socket(zmq.PULL)

pull_socket.setsockopt(zmq.RCVHWM, 1)
pull_socket.setsockopt(zmq.SNDHWM, 1)

pull_socket.connect("tcp://127.0.0.1:5557")
print(pull_socket.get_hwm()) # Prints 1

print('Connected, but not receiving yet... (Sleep 4s)')
time.sleep(4)
print('Receiving everything now!')

rec = ''

for i in range(2200):
    rec += '{} '.format(pull_socket.recv().decode('ascii'))

print(rec) # Prints `0 1 2 ... 2198 2199 `

为了重现我的测试用例，打开两个终端并在一个终端中启动第一个 puller.py，然后快速(4 秒窗口)在另一个终端中启动 pusher.py。

Answer 1

这里至少涉及到4个buffer:zmq send buffer，OS write tcp buffer，OS read tcp buffer，zmq recv buffer。

zmq io 线程在消息成功写入操作系统 tcp 写入缓冲区后将其标记为“已发送”。这些消息现在被视为“正在传输”。

然后网络堆栈负责将尽可能多的数据传输到其他进程的匹配操作系统接收缓冲区中，最后，接收 zmq io 线程一次最多从该缓冲区读取 HWM 消息到 ZMQ 接收队列。

默认情况下，操作系统缓冲区通常在 10-100kb 左右，并且在 ZMQ 甚至注意到另一方没有使用任何消息之前，这两个缓冲区都可能被“传输中”的消息完全填满。出于性能原因，这些缓冲区是必需的 - 您不能只是摆脱它们。

您的问题的解决方案可能涉及请求/代表套接字和显式应用程序级确认，即指南中的懒惰盗版模式。