ZeroMQ HEARTBEAT套接字选项工作原理及pyzmq配置异常问题

ZeroMQ 心跳机制不触发异常问题说明 #

心跳机制的预期运行逻辑 #

• 核心参数作用：
  • ZMQ_HEARTBEAT_IVL：两次发送心跳包的间隔，单位为毫秒
  • ZMQ_HEARTBEAT_TIMEOUT：未收到对端任何数据（包含心跳包、业务数据）的最长容忍时间，超过该值即判定对端离线，单位为毫秒
  • ZMQ_HEARTBEAT_TTL：心跳包的网络存活时长，限制中转节点的最长保留时间，通常和超时时间设为一致即可
• 服务端（ROUTER 端）检测逻辑：判定客户端离线后，会直接丢弃该客户端的连接上下文，不会主动向上层抛出异常，仅当后续程序尝试给该离线客户端发消息时，才会返回 EAGAIN 错误
• 客户端（DEALER 端）检测逻辑：判定服务端离线后，底层会标记连接失效，同样默认不会主动抛出异常，仅当程序调用 recv()/send() 执行 IO 操作时才会触发错误反馈

原有代码的核心问题 #

1. 客户端仅配置了 ZMQ_HEARTBEAT_IVL 和 ZMQ_HEARTBEAT_TIMEOUT，缺少 ZMQ_HEARTBEAT_TTL 参数配置，心跳机制需要两端三个参数配套才能正常生效
2. 使用了无超时限制的阻塞式 recv() 调用，哪怕底层连接已经判定失效，阻塞调用会无限等待，不会返回异常信息

可复现心跳异常的修复代码 #

服务端代码

# Server Code:
import zmq
c = zmq.Context()
s = c.socket(zmq.ROUTER)
s.setsockopt(zmq.HEARTBEAT_IVL, 1000)
s.setsockopt(zmq.HEARTBEAT_TIMEOUT, 5000)
s.setsockopt(zmq.HEARTBEAT_TTL, 5000)
s.bind('tcp://127.0.0.1:5555')

while True:
    try:
        # 用带超时的poll检测事件，替换无限阻塞recv
        if s.poll(timeout=1000):
            id, data = s.recv_multipart()
            s.send_multipart([id, data], zmq.NOBLOCK)
    except zmq.ZMQError as e:
        print(f"服务端异常：{e}")

客户端代码

# Client Code
import zmq
import time
c = zmq.Context()
s = c.socket(zmq.DEALER)
# 补全三个心跳参数配置
s.setsockopt(zmq.HEARTBEAT_IVL, 1000)
s.setsockopt(zmq.HEARTBEAT_TIMEOUT, 5000)
s.setsockopt(zmq.HEARTBEAT_TTL, 5000)
s.connect('tcp://127.0.0.1:5555')
i = 0
while True:
    try:
        s.send(str(i).encode(), flags=zmq.NOBLOCK)
        # 带超时的recv检测
        if s.poll(timeout=1000):
            print(s.recv().decode())
        i += 1
        time.sleep(1)
    except zmq.ZMQError as e:
        print(f"客户端检测到服务端离线，异常信息：{e}")
        break

验证方式 #

同时启动服务端和客户端，正常运行后直接终止服务端进程，等待5秒左右即可看到客户端抛出对应超时异常。

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