如何设置ZMQ Pub/Sub订阅者完全不缓存消息？

你遇到的是ZMQ高水位线（HWM）设置不符合预期的典型场景，核心原因是HWM的生效逻辑是双向协同的，且默认针对单连接缓冲区限制，仅在订阅端设置根本压不住发布端的消息积压。下面是具体的原因分析和解决办法：

为什么set_hwm(1)没生效？ #

1. HWM是两端协同限制：ZMQ的HWM不是单方面生效的，只在订阅端设置HWM，发布端仍会把所有消息缓存到自身发送缓冲区，等订阅端休眠结束恢复连接后，会一次性推送全部积压消息。
2. 默认HWM值过大：ZMQ默认HWM为1000（不同版本可能有差异），发布端未设置时，会无限制缓存消息直到达到系统级缓冲区上限。
3. 订阅端休眠期间发布端持续发送：如果测试代码中发布端在订阅端休眠时持续发消息，这些消息都会存在发布端队列中，订阅端醒来后会全部接收。

解决办法 #

1. 两端同时设置HWM #

必须在发布端和订阅端同步设置相同的HWM值，这样发布端的发送缓冲区会被限制，当订阅端无法接收时，发布端会被阻塞，不会继续发送超出HWM的消息。

示例代码：

发布端

import zmq
import time

ctx = zmq.Context()
pub_sock = ctx.socket(zmq.PUB)
pub_sock.set_hwm(1)  # 发布端同步设置HWM
pub_sock.bind("tcp://*:5555")

# 等待订阅端建立连接（测试场景用，生产环境可按需调整）
time.sleep(1)

for i in range(6):
    pub_sock.send_string(f"Msg {i}")
    print(f"Sent: Msg {i}")
    time.sleep(0.2)

订阅端

import zmq
import time

ctx = zmq.Context()
sub_sock = ctx.socket(zmq.SUB)
sub_sock.set_hwm(1)
sub_sock.connect("tcp://localhost:5555")
sub_sock.setsockopt_string(zmq.SUBSCRIBE, "")

print("Sleeping 3s...")
time.sleep(3)

# 尝试接收消息
while True:
    try:
        msg = sub_sock.recv_string(flags=zmq.NOBLOCK)
        print(f"Received: {msg}")
    except zmq.Again:
        break

2. 使用ZMQ_CONFLATE选项（适合仅需最新消息的场景） #

如果你的需求是订阅端只获取最新的一条消息、完全忽略积压旧消息，直接设置CONFLATE选项比HWM更直接。该选项会让订阅端缓冲区仅保留最新一条消息，所有旧消息自动丢弃。

修改订阅端代码：

sub_sock = ctx.socket(zmq.SUB)
sub_sock.setsockopt(zmq.CONFLATE, 1)  # 开启只保留最新消息
sub_sock.connect("tcp://localhost:5555")
sub_sock.setsockopt_string(zmq.SUBSCRIBE, "")

3. 生产环境额外注意事项 #

• 不同ZMQ版本的HWM行为可能略有差异，建议查阅对应版本官方文档确认参数细节。
• 若使用TCP传输，需考虑TCP本身的缓冲区，必要时可结合ZMQ_SNDBUF和ZMQ_RCVBUF参数进一步限制。
• 避免发布端无限制发送消息，最好根据订阅端处理能力做流量控制。

内容的提问来源于stack exchange，提问作者YNX