ZMQ中ROUTER/DEALER模式下客户端重启后如何重连？

问题根源 #

ROUTER套接字会维护一张身份（IDENTITY）与连接的映射表，但它不会主动检测客户端连接是否已断开。当客户端重启后，即便使用了相同的IDENTITY，ROUTER的映射表中该身份对应的仍是旧的失效连接，后续发送给该身份的数据会被静默丢弃，导致新客户端收不到消息。

解决方法 #

1. 启用ROUTER强制发送选项

给ROUTER设置ZMQ_ROUTER_MANDATORY选项，这样当ROUTER往不存在的身份发送数据时会抛出异常，而非默默丢弃。你可以捕获异常并移除无效身份条目，避免后续白发送数据。

修改后的ROUTER代码：

# ROUTER
import time
import random
import zmq

context = zmq.Context.instance()
client = context.socket(zmq.ROUTER)
client.setsockopt(zmq.TCP_KEEPALIVE, 1)
# 启用强制发送，发送无效身份时抛出异常
client.setsockopt(zmq.ROUTER_MANDATORY, 1)
client.bind("tcp://127.0.0.1:99999")

# 维护活跃身份列表
active_ids = set()

for _ in range(100):
    ident = random.choice([b'A', b'A', b'B'])
    work = b"This is the workload"
    try:
        client.send_multipart([ident, work])
        active_ids.add(ident)
    except zmq.ZMQError as e:
        if e.errno == zmq.EHOSTUNREACH:
            active_ids.discard(ident)
            print(f"身份 {ident} 连接失效，已移除")
        else:
            raise
    time.sleep(0.5)

2. 客户端重启后主动注册身份

让客户端连接ROUTER后先发送一条注册消息，ROUTER收到后更新对应身份的连接映射，确保后续消息能发到新连接上。

修改后的CLIENT代码：

# CLIENT
import zmq

context = zmq.Context.instance()
worker = context.socket(zmq.DEALER)
worker.setsockopt(zmq.IDENTITY, b'A')
worker.connect("tcp://127.0.0.1:99999")

# 发送注册消息告知ROUTER自己的身份
worker.send(b"REGISTER")

while True:
    request = worker.recv()
    # 跳过注册确认，只处理业务消息
    if request != b"REGISTER_ACK":
        print(request)

ROUTER端需添加注册消息处理逻辑：

# 在ROUTER循环中加入消息监听
poller = zmq.Poller()
poller.register(client, zmq.POLLIN)

for _ in range(100):
    # 先检查是否有注册消息（超时100ms，不影响定时发送）
    socks = dict(poller.poll(100))
    if client in socks and socks[client] == zmq.POLLIN:
        ident, msg = client.recv_multipart()
        if msg == b"REGISTER":
            active_ids.add(ident)
            # 可选：回复注册确认
            client.send_multipart([ident, b"REGISTER_ACK"])
    
    # 发送业务数据
    ident = random.choice([b'A', b'A', b'B'])
    work = b"This is the workload"
    try:
        if ident in active_ids:
            client.send_multipart([ident, work])
    except zmq.ZMQError as e:
        if e.errno == zmq.EHOSTUNREACH:
            active_ids.discard(ident)
            print(f"身份 {ident} 连接失效，已移除")
        else:
            raise
    time.sleep(0.5)

3. 强化TCP保活检测

你已经开启了TCP_KEEPALIVE，可以进一步配置参数让ROUTER更快识别断开的连接：

# ROUTER端添加TCP保活参数
client.setsockopt(zmq.TCP_KEEPALIVE, 1)
client.setsockopt(zmq.TCP_KEEPALIVE_IDLE, 30)  # 30秒无数据发送保活包
client.setsockopt(zmq.TCP_KEEPALIVE_INTVL, 5)   # 保活包间隔5秒
client.setsockopt(zmq.TCP_KEEPALIVE_CNT, 3)     # 3次无响应则判定连接断开

ROUTER-DEALER模式本身支持客户端重启重连，只是需要额外机制维护身份与连接的映射，确保ROUTER能将消息发送到新的有效连接上。

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