ZMQ Pub/Sub模式下重启Publisher后Subscriber无法重连问题咨询

我在使用ZMQ的Pub/Sub模式时，部署了Publisher和Subscriber节点，分别用Python（pyzmq）和C++（cppzmq）做了测试。初始启动两者时一切正常，但重启Publisher节点并重新启动服务后，未重启的Subscriber始终无法重连、接收不到消息；而重启Subscriber后就能正常接收消息。抓包分析发现Subscriber从未尝试重连，推测Publisher在等待新的订阅请求。想问ZMQ不支持Publisher重启场景吗？还是我的配置有问题？

复现问题的代码 #

subscriber.py

import zmq
import sys

def subscriber():
    context = zmq.Context()
    socket = context.socket(zmq.SUB)
    socket.connect(sys.argv[1])

    # 订阅所有消息
    socket.setsockopt_string(zmq.SUBSCRIBE, '')

    while True:
        message = socket.recv_string()
        print(f"Received: {message}")

if __name__ == "__main__":
    subscriber()

运行命令：python subscriber.py tcp://localhost:5555

publisher.py

import zmq
import time

def publisher():
    context = zmq.Context()
    socket = context.socket(zmq.PUB)
    socket.bind("tcp://*:5555")

    while True:
        message = "Hello, World!"
        socket.send_string(message)
        print(f"Sent: {message}")
        time.sleep(0.5) 

if __name__ == "__main__":
    publisher()

运行命令：python publisher.py

注意事项 #

需隔离Publisher节点复现问题，可使用虚拟机或Docker。

复现步骤 #

• 启动Subscriber
• 启动Publisher
• 重启Publisher所在虚拟机
• 重新启动Publisher
• Subscriber无法接收新消息

问题原因 #

ZMQ的SUB套接字默认不会自动检测TCP连接断开并发起重连，且Pub/Sub模式中订阅请求是在连接建立时由Subscriber主动发送给Publisher的。当Publisher重启后，原有TCP连接已失效，但Subscriber未感知到连接断开，也不会主动重新发起连接并发送订阅请求，导致新启动的Publisher没有该Subscriber的订阅信息，无法推送消息。

ZMQ本身支持Publisher重启场景，问题出在默认配置缺少重连逻辑。

解决方法 #

方法1：配置ZMQ心跳机制

通过设置心跳相关套接字选项，让ZMQ底层自动检测连接失效并重连，同时重新发送订阅请求。修改subscriber.py，在connect前添加以下配置：

# 配置心跳参数，单位为毫秒
socket.setsockopt(zmq.HEARTBEAT_IVL, 1000)    # 每隔1秒发送一次心跳
socket.setsockopt(zmq.HEARTBEAT_TIMEOUT, 3000)# 3秒未收到心跳则判定连接断开
socket.setsockopt(zmq.HEARTBEAT_TTL, 2000)    # 心跳包存活时间

方法2：捕获异常手动重连

在Subscriber的消息接收循环中捕获连接异常，主动关闭套接字并重新创建连接。修改后的subscriber.py代码如下：

import zmq
import sys
import time

def subscriber():
    context = zmq.Context()
    while True:
        try:
            socket = context.socket(zmq.SUB)
            socket.connect(sys.argv[1])
            socket.setsockopt_string(zmq.SUBSCRIBE, '')
            print("已连接到Publisher，开始接收消息...")
            while True:
                message = socket.recv_string()
                print(f"Received: {message}")
        except zmq.ZMQError as e:
            print(f"连接异常: {e}，尝试重连...")
            socket.close()
            time.sleep(2)
        except KeyboardInterrupt:
            print("程序退出")
            socket.close()
            context.term()
            break

if __name__ == "__main__":
    subscriber()

方法3：使用CONNECT_RID选项（进阶）

若使用ZMQ 4.1+版本，可为每个Subscriber设置唯一连接ID（CONNECT_RID），让Publisher识别重连的Subscriber，该方案适用于更复杂的多Subscriber场景，一般前两种方法已能解决问题。

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