如何检测ZMQ订阅者Socket是否存活并避免连接超时？

问题本质 #

在Windows 7上，长时间无数据传输的TCP连接很容易被系统防火墙或网络设备主动断开，但ZMQ的SUB套接字默认不会检测这种“死连接”——你的代码用了阻塞式recv_string()，一旦连接断开，这个调用会一直挂起，哪怕新发布者启动也收不到消息。而发布端因为ZMQ的消息队列机制，发送时不会立刻报错（消息会暂存在本地队列，直到超时才丢弃，不会给出提示）。

解决方案 #

结合TCP保活机制和带超时的接收逻辑，既能防止连接被无故断开，又能在连接真的失效时及时检测并重连，同时不影响正常消息接收。

1. 开启TCP保活 #

通过ZMQ套接字选项配置TCP保活参数，让系统定期发送探测包维持连接，避免空闲过久被断开：

• TCP_KEEPALIVE: 启用TCP保活（Windows下设为1即可）
• TCP_KEEPALIVE_IDLE: 空闲5分钟后开始发送探测包（300秒）
• TCP_KEEPALIVE_INTVL: 每隔1分钟发送一次探测包（60秒）
• TCP_KEEPALIVE_CNT: 连续5次探测失败则判定连接失效

2. 给接收设置超时，实现自动重连 #

给recv_string()设置超时（比如10秒），超时后触发重连逻辑，避免程序永久卡在阻塞接收上。

修改后的订阅者代码 #

import zmq
import time

def create_sub_socket(context, addr):
    socket = context.socket(zmq.SUB)
    # 配置TCP保活参数
    socket.setsockopt(zmq.TCP_KEEPALIVE, 1)
    socket.setsockopt(zmq.TCP_KEEPALIVE_IDLE, 300)
    socket.setsockopt(zmq.TCP_KEEPALIVE_INTVL, 60)
    socket.setsockopt(zmq.TCP_KEEPALIVE_CNT, 5)
    # 设置接收超时，单位为毫秒（这里设10秒）
    socket.setsockopt(zmq.RCVTIMEO, 10000)
    socket.connect(addr)
    socket.setsockopt_string(zmq.SUBSCRIBE, "")
    return socket

context = zmq.Context()
server_addr = "tcp://localhost:10001"
socket = create_sub_socket(context, server_addr)

while True:
    try:
        msg = socket.recv_string()
        do_something(msg)
    except zmq.Again:
        # 接收超时，尝试重连
        print("长时间未收到消息，尝试重新连接...")
        socket.close()
        socket = create_sub_socket(context, server_addr)
    except Exception as e:
        # 处理其他异常
        print(f"出现异常: {str(e)}")
        time.sleep(5)
        socket.close()
        socket = create_sub_socket(context, server_addr)

关键说明 #

• TCP保活：主动维持连接，减少被外部设备断开的概率，适配长时间可能无消息的场景。
• 接收超时：避免程序永久卡在recv_string()，超时后可及时检测连接状态并触发重连。
• 重连逻辑：关闭旧套接字后重建，确保使用全新的有效连接，避免残留无效状态。
• 无需额外线程发送空消息，通过ZMQ原生选项即可实现，简洁可靠。

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