ZeroMQ EPGM模式PUB/SUB通信接收端无法接收消息的代码修复请求

看起来你在Ubuntu无线环境下用ZeroMQ的EPGM组播+PUB/SUB模式遇到了收不到消息的问题，我来帮你一步步排查和修复。

首先，先从系统层面排查可能的限制，再调整代码：

一、系统层面的配置调整（无线环境组播常见问题） #

无线网卡对组播的支持有时候会有限制，先做这几个配置：

1. 调整内核组播转发参数
   编辑/etc/sysctl.conf文件，添加以下内容：

   net.ipv4.conf.wlp2s0.mc_forwarding = 1
   net.ipv4.conf.all.mc_forwarding = 1
   net.ipv4.ip_forward = 1
   net.ipv4.icmp_echo_ignore_broadcasts = 0
   

   执行sudo sysctl -p让配置生效，这能确保系统允许组播数据包转发和处理。

2. 调整无线网卡组播速率
   有些无线网卡默认组播速率很低，容易丢包，执行以下命令调整：

   sudo iwconfig wlp2s0 rate 54M multicast
   

   如果不行，可以试试sudo iwconfig wlp2s0 rate auto multicast。

3. 检查防火墙设置
   Ubuntu的ufw防火墙可能会拦截组播UDP流量，先临时关闭测试：

   sudo ufw disable
   

   如果关闭后能正常接收，再添加允许规则：

   sudo ufw allow 5555/udp
   

二、代码调整（修复ZeroMQ配置问题） #

你的代码核心逻辑没问题，但缺少几个关键的组播相关配置，修改后的代码如下：

发布端代码 #

import zmq
import struct
import time

context = zmq.Context()
context.set(zmq.IO_THREADS, 1)

socket = context.socket(zmq.PUB)
# 提高组播TTL，确保数据包能在无线环境中正常传播
socket.setsockopt(zmq.MULTICAST_TTL, 10)
# 允许本地回环（同一机器测试时有用）
socket.setsockopt(zmq.MULTICAST_LOOP, 1)
socket.bind("epgm://192.168.1.122;224.0.0.251:5555")

# 给订阅端足够的时间完成连接和订阅（PUB/SUB模式下，过早发送的消息会丢失）
time.sleep(2)

while True:
    data = struct.pack("q", time.perf_counter_ns())
    socket.send(data)
    print(f"Sent message: {data}")
    # 添加小延迟，避免发送过快导致网卡或ZeroMQ队列积压
    time.sleep(0.1)

订阅端代码 #

import zmq
import struct
import time

numbers = []

context = zmq.Context()
context.set(zmq.IO_THREADS, 1)

socket = context.socket(zmq.SUB)
# 取消接收高水位限制，防止高负载下丢包
socket.setsockopt(zmq.RCVHWM, 0)
# 允许接收本地回环的组播包
socket.setsockopt(zmq.MULTICAST_LOOP, 1)
socket.connect("epgm://192.168.1.122;224.0.0.251:5555")
# 订阅所有消息（空字节订阅）
socket.setsockopt(zmq.SUBSCRIBE, b"")

print("Waiting for messages...")
while True:
    data = socket.recv()
    current_time = time.perf_counter_ns()
    numbers.append((current_time, data))
    print(f"Received message: {data} | Local timestamp: {current_time}")

三、关键修改点说明 #

• 发布端的延迟等待：time.sleep(2)是为了给订阅端足够时间完成连接和订阅流程，ZeroMQ的PUB/SUB模式中，订阅端刚连接时无法接收之前发送的消息，所以需要等一下再开始发。
• 组播TTL设置：默认的TTL是1，可能在无线环境中不足以让数据包到达订阅端，提高到10能覆盖大多数局域网场景。
• 接收高水位调整：订阅端设置RCVHWM=0取消了接收队列的长度限制，避免消息过多时被ZeroMQ丢弃。
• 发送延迟：time.sleep(0.1)避免发布端发送过快，导致网卡或ZeroMQ内部队列处理不过来。

按照这个步骤调整后，应该就能正常接收消息了。

备注：内容来源于stack exchange，提问作者Sergey Ko