GNU Radio 3.10：Python实现运行中流图块间数据无中断动态获取

你遇到的核心问题是动态修改流图结构（添加/删除块）必须锁定流图，而锁定会暂停流图执行。下面提供三种无需修改流图、不中断执行的解决方案：

方法1：自定义带线程安全缓冲区的处理块 #

直接在目标块（块A）中集成线程安全队列，块的work函数处理数据时，同时将输出样本写入队列。外部Python线程可以非阻塞地从队列读取数据，完全不涉及流图结构修改。

import queue
import numpy as np
import threading
import time
from gnuradio import gr

class BlockA(gr.sync_block):
    def __init__(self, output_queue):
        gr.sync_block.__init__(
            self,
            name="BlockA",
            in_sig=None,  # 根据实际输入信号类型调整
            out_sig=[np.float32]  # 根据实际输出信号类型调整
        )
        self.output_queue = output_queue
        self.max_queue_size = 1000  # 限制队列大小防止内存溢出

    def work(self, input_items, output_items):
        # 替换为块A的实际处理逻辑
        output_items[0][:] = np.sin(np.linspace(0, 2*np.pi, len(output_items[0])))
        
        # 非阻塞写入线程安全队列
        for sample in output_items[0]:
            if not self.output_queue.full():
                self.output_queue.put(sample, block=False)
        
        return len(output_items[0])

def read_queue(queue):
    """非阻塞读取队列数据的线程函数"""
    while True:
        try:
            sample = queue.get(block=False)
            print(f"获取到样本: {sample:.4f}")
        except queue.Empty:
            time.sleep(0.01)  # 短暂休眠降低CPU占用

def main():
    # 创建线程安全队列
    data_queue = queue.Queue(maxsize=1000)
    
    # 构建流图
    tb = gr.top_block()
    block_a = BlockA(data_queue)
    # 按需连接其他块（比如源块、其他Sink）
    # tb.connect(...)
    
    # 启动数据读取线程
    reader_thread = threading.Thread(target=read_queue, args=(data_queue,), daemon=True)
    reader_thread.start()
    
    # 启动流图
    tb.start()
    
    # 让流图运行10秒
    time.sleep(10)
    
    # 停止并清理流图
    tb.stop()
    tb.wait()

if __name__ == "__main__":
    main()

方法2：使用GNU Radio异步消息端口 #

利用GNU Radio的消息传递机制，块A将处理后的数据打包成消息发送到消息端口，外部通过消息队列读取数据。消息操作是线程安全且非阻塞的，无需锁定流图。

import numpy as np
import threading
import time
from gnuradio import gr, blocks

class BlockAWithMsg(gr.sync_block):
    def __init__(self, msg_queue):
        gr.sync_block.__init__(
            self,
            name="BlockAWithMsg",
            in_sig=None,
            out_sig=[np.float32]
        )
        self.msg_queue = msg_queue

    def work(self, input_items, output_items):
        # 块A的实际处理逻辑
        output_items[0][:] = np.cos(np.linspace(0, 2*np.pi, len(output_items[0])))
        
        # 将批次数据打包成二进制消息发送（减少消息数量，提升效率）
        batch_data = output_items[0].tobytes()
        self.msg_queue.insert_tail(gr.message_from_string(batch_data))
        
        return len(output_items[0])

def read_messages(msg_queue):
    """读取消息队列的线程函数"""
    while True:
        msg = msg_queue.delete_head()
        if not msg:
            break
        # 将二进制消息解析为numpy数组
        data = np.frombuffer(msg.to_string(), dtype=np.float32)
        print(f"获取到批次数据（前5个样本）: {data[:5].round(4)}")

def main():
    # 创建GNU Radio消息队列
    msg_queue = gr.msg_queue()
    
    # 构建流图
    tb = gr.top_block()
    block_a = BlockAWithMsg(msg_queue)
    # 按需连接其他块
    # tb.connect(...)
    
    # 启动消息读取线程
    msg_reader = threading.Thread(target=read_messages, args=(msg_queue,), daemon=True)
    msg_reader.start()
    
    # 启动流图
    tb.start()
    
    # 运行10秒
    time.sleep(10)
    
    # 停止流图
    tb.stop()
    tb.wait()

if __name__ == "__main__":
    main()

方法3：零拷贝共享内存（适合大数据量场景） #

使用共享内存数组，块A直接将处理后的数据写入共享内存，外部线程/进程读取内存中的数据。这种方式完全不涉及流图交互，延迟最低，适合高吞吐量场景。

import multiprocessing
import numpy as np
import threading
import time
from gnuradio import gr

class BlockASharedMem(gr.sync_block):
    def __init__(self, shared_array):
        gr.sync_block.__init__(
            self,
            name="BlockASharedMem",
            in_sig=None,
            out_sig=[np.float32]
        )
        self.shared_array = shared_array
        # 将共享内存转换为numpy数组
        self.np_array = np.frombuffer(shared_array.get_obj(), dtype=np.float32)
        self.array_size = self.np_array.shape[0]
        self.write_index = 0

    def work(self, input_items, output_items):
        # 块A的实际处理逻辑
        output_items[0][:] = np.sin(np.linspace(0, 2*np.pi, len(output_items[0])))
        
        # 写入共享内存（环形缓冲区模式，避免溢出）
        num_samples = len(output_items[0])
        if self.write_index + num_samples <= self.array_size:
            self.np_array[self.write_index:self.write_index+num_samples] = output_items[0]
            self.write_index += num_samples
        else:
            # 覆盖开头的旧数据
            remaining = self.array_size - self.write_index
            self.np_array[self.write_index:] = output_items[0][:remaining]
            self.np_array[:num_samples - remaining] = output_items[0][remaining:]
            self.write_index = num_samples - remaining
        
        return len(output_items[0])

def read_shared_mem(shared_array):
    """读取共享内存的线程函数"""
    np_array = np.frombuffer(shared_array.get_obj(), dtype=np.float32)
    while True:
        # 读取最新的100个样本
        print(f"最新100个样本均值: {np_array[-100:].mean():.4f}")
        time.sleep(0.1)

def main():
    # 创建共享内存数组：10000个float32类型样本
    shared_array = multiprocessing.Array('f', 10000)
    
    # 构建流图
    tb = gr.top_block()
    block_a = BlockASharedMem(shared_array)
    # 按需连接其他块
    # tb.connect(...)
    
    # 启动共享内存读取线程
    mem_reader = threading.Thread(target=read_shared_mem, args=(shared_array,), daemon=True)
    mem_reader.start()
    
    # 启动流图
    tb.start()
    
    # 运行10秒
    time.sleep(10)
    
    # 停止流图
    tb.stop()
    tb.wait()

if __name__ == "__main__":
    main()

关键说明 #

• 三种方案均无需修改流图结构，因此不需要调用lock()/unlock()，不会中断流图执行。
• 线程安全队列适合小数据量、简单场景；消息端口适合按批次传递数据；共享内存适合低延迟、高吞吐量场景。

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