GNU Radio ZeroMQ SUB块消息间隔超约50ms时无法接收消息

背景 #

我正尝试通过UHF天线向远程系统发送命令并接收数据，计划用一台计算机的Python程序和另一台的GNU Radio程序通过ZeroMQ的PUB/SUB块收发数据。远程UHF和GNU Radio的链路正常，但Python到GNU Radio的传输存在问题：GNU Radio用ZMQ PUB、Python用SUB的组合完全正常，但反向的Python PUB到GNU Radio SUB的链路有问题。

环境设置 #

Python PUB测试代码 #

import zmq
import time
import pmt

context = zmq.Context()
socket = context.socket(zmq.PUB)
socket.connect("tcp://127.0.0.1:5557")
i=1
while True:
    message = str("(Message #"+str(i)+") Hello!\n")
    res = bytes(message, 'utf-8')
    socket.send(res)
    print("Sending "+str(i))
    time.sleep(0.01)
    i+=1

GNU Radio流程图 #

使用ZMQ SUB块连接到文件输出块，配置为接收本地主机（127.0.0.1:5557）的消息，将接收内容写入文本文件。

Python SUB测试代码 #

import zmq
import time

context = zmq.Context()
socket = context.socket(zmq.SUB)
socket.bind("tcp://127.0.0.1:5556")
socket.subscribe("")
while True:
    try:
        message = socket.recv(flags=zmq.NOBLOCK)
        print("Message rx")
        print(str(message))
    except zmq.Again as e:
        e=1
    time.sleep(0.01)

问题描述 #

先启动GNU Radio流程图，再启动Python PUB脚本，短间隔（如0.01秒）发送数百条消息后关闭程序，文本文件能得到预期输出（仅首尾少量消息丢失）；但增大time.sleep()间隔后（超过0.1秒时几乎完全失败，低于0.02秒基本正常），文本文件为空。而用Python SUB脚本接收时，无论间隔长短都能正常工作（开头几条丢失属预期）。我需要实现从几分之一秒到数小时的可变间隔收发。

已尝试方案 #

• 交换两端的connect/bind设置
• 调整GNU Radio SUB块的超时值
• 更改程序启动顺序
• 调整流程图采样率（本不相关仍尝试）
• 使用pmt.serialize_str(pmt.to_pmt())及多种数据类型格式化消息
• 流程图启动后等待几秒再发送消息
• 切换SUB块的“Pass Tags”选项
• 突发式发送：先每秒发10条间隔0.01秒的“牺牲消息”，再发1条内容消息，延迟1秒后重复

以上操作均未解决问题或完全无法接收。

其他信息 #

• 操作系统：Windows 10（测试），正式使用Ubuntu
• Python版本：3.10.10
• GNU Radio：Windows下用Radioconda安装

解决方案 #

1. 正确使用PMT格式传递消息 #

GNU Radio的ZMQ块默认使用**PMT（Polymorphic Types）**格式传输数据，直接发送原始字节流时，长间隔消息可能无法被GNU Radio的流处理逻辑识别为有效消息。修改Python代码，将消息序列化为PMT格式：

import zmq
import time
import pmt

context = zmq.Context()
socket = context.socket(zmq.PUB)
socket.connect("tcp://127.0.0.1:5557")
i=1
while True:
    message = f"(Message #{i}) Hello!\n"
    # 将字符串转为PMT字符串并序列化
    pmt_msg = pmt.serialize_str(pmt.intern(message))
    socket.send(pmt_msg)
    print(f"Sending {i}")
    time.sleep(0.5)  # 测试长间隔
    i+=1

同时确保GNU Radio的ZMQ SUB块**“Raw”选项未勾选**（默认状态）。

2. 改用ZMQ PUSH/PULL块 #

如果不需要PUB/SUB的多订阅广播特性，改用PUSH/PULL组合更适合点对点间歇性传输：

• Python端使用zmq.PUSH socket，GNU Radio端使用ZMQ PULL块
• 勾选GNU Radio块的“Raw”选项，可直接传输字节流，无需PMT格式，且不受消息间隔影响

3. 调整GNU Radio ZMQ SUB块参数 #

• 设置**“Timeout”为0**（无限等待），避免因超时而丢弃长间隔消息
• 调大**“High Water Mark”**值，防止消息因缓冲区满被丢弃

4. 切换为消息模式处理 #

如果传输的是离散消息而非流数据，将ZMQ SUB块的**“Output Type”设置为“Message”**，连接到消息类处理块（如Message Debug或消息模式的File Sink），让GNU Radio以离散消息方式处理，更适配间歇性传输场景

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