如何在 python 中限制 ZMQ (ZeroMQ - PyZMQ) 队列缓冲区大小？

Question

我正在使用带有 pub/sub 模式的 pyzmq 库。我有 一些使用 .connect() 方法的快速 ZMQ 发布者 和使用 .bind() 方法的较慢的 ZMQ 订阅者。 几分钟后，我的订阅者收到了发布者发布的旧数据——由于 ZMQ 缓冲区。

---

我的问题：

有没有办法管理ZMQ队列缓冲区大小？（设置有限缓冲区）

[注意]：

• 我不想使用 ZMQ PUSH/PULL。
• 我读过这篇文章，但这种方法只清除缓冲区：clear ZMQ buffer
• 我也尝试了high watermark 选项，但没有成功：

socket.setsockopt(zmq.RCVHWM, 10)  # not working
socket.setsockopt(zmq.SNDHWM, 10)  # not working

---

出版商：

import zmq
import time

port = "5556"
context = zmq.Context()
socket = context.socket(zmq.PUB)
socket.bind("tcp://*:%s" % port)
socket.setsockopt(zmq.SNDHWM, 10)  # not working

while True:
    data = time.time()
    print("%d" % data)
    socket.send("%d" % data)
    time.sleep(1)

订阅者：

import zmq
import time

port = "5556"
context = zmq.Context()
socket = context.socket(zmq.SUB)
socket.connect("tcp://localhost:%s" % port)
socket.setsockopt(zmq.SUBSCRIBE, '')
socket.setsockopt(zmq.RCVHWM, 10)  # not working

while 1:
    time.sleep(2)  # A speed reducer like.
    data = socket.recv()
    print(data)

尽管有这些选项（通过配置的send/receive high watermark），队列大小仍然大于 10。

Answer 1

我找到了一种方法，可以使用CONFLATE 选项在ZMQ 订阅者中获取仅最后一条消息。

请注意，您应该在连接之前设置CONFLATE 选项：

import zmq
import time

port = "5556"
context = zmq.Context()
socket = context.socket(zmq.SUB)

socket.setsockopt(zmq.SUBSCRIBE, '')
socket.setsockopt(zmq.CONFLATE, 1)  # last msg only.
socket.connect("tcp://localhost:%s" % port)  # must be placed after above options.

while 1:
    time.sleep(2)  # Dummy delay
    data = socket.recv()
    print(data)

换句话说，它会删除订阅者端的所有缓冲队列。

---

[注意]：

此外，通过使用zmq.SNDBUF 和zmq.RCVBUF 选项，我们可以设置ZMQ 缓冲区大小的限制。 (More information)

---

Answer 2

要设置队列/缓冲区大小，您需要通过套接字选项设置高水位标记

setsockopt(zmq.SNDHWM, 10)
setsockopt(zmq.RCVHWM, 10)