如何从asyncio事件循环中获取异步任务的实时返回值

需要实时获取RTL-SDR dongle采集的信号功率电平，返回主函数完成多组测量对比，最终得到最优功率值。原代码异步逻辑写法错误，无法正常获取实时计算的功率结果。

• 异步接口调用错误：asyncio.run()已经启动了运行中的事件循环，在协程内部再调用loop.run_until_complete()、手动执行loop.close()会直接触发运行时异常
• 设备重复初始化：每次测量循环都重新创建SDR实例、启动采集流，会反复占用/释放硬件，既无法拿到连续实时数据，还容易触发设备占用错误
• 传值方式不可靠：用全局变量跨协程传值没有做同步，容易读到未初始化、过期的功率值
• 协程调用语法错误：asyncio.gather(*task)写法不成立，streaming()返回的是单个协程对象，不是可迭代的任务集合，解引用会直接抛运行错误

用异步队列做协程间的安全传值，SDR采集作为后台常驻任务运行，主逻辑按需读取最新功率值做对比即可，代码如下：

import asyncio
from rtlsdr import RtlSdr
import numpy as np

async def streaming(power_queue: asyncio.Queue):
    # 仅初始化一次SDR设备
    sdr = RtlSdr()
    sdr.rs = 2.4e6
    sdr.fc = 801e6
    calibrate = 3.2

    try:
        async for samples in sdr.stream(num_samples_or_bytes=512):
            # 实时计算校准后的功率值
            samples = samples - np.mean(samples)
            raw_power = np.mean(np.abs(samples ** 2))
            cal_power = calibrate * 10 * np.log10(raw_power)
            
            # 队列只保留最新功率值，满了就丢弃旧数据
            if power_queue.full():
                _ = power_queue.get_nowait()
            await power_queue.put(cal_power)
    finally:
        # 确保程序退出时正确释放设备
        await sdr.stop()
        sdr.close()

async def main():
    measure_times = 100
    max_power = -100 # 初始功率阈值
    # 长度为1的异步队列，仅存储最新的功率计算结果
    power_queue = asyncio.Queue(maxsize=1)
    # 启动后台采集任务
    sdr_task = asyncio.create_task(streaming(power_queue))

    try:
        for _ in range(measure_times):
            # 等待获取最新功率值
            current_power = await power_queue.get()
            # 更新最优功率
            if current_power > max_power:
                max_power = current_power
                print(f"当前最优功率: {max_power:.2f} dB")
            # 此处可插入调整SDR参数（增益、频点等）的逻辑
            await asyncio.sleep(0.01)
    finally:
        # 结束后取消采集任务，释放资源
        sdr_task.cancel()
        try:
            await sdr_task
        except asyncio.CancelledError:
            pass

if __name__ == "__main__":
    asyncio.run(main(), debug=True)

• 弃用全局变量，用asyncio.Queue做协程间的异步安全通信，队列长度设为1，保证主函数每次拿到的都是最新计算的功率值，不会堆积过期数据
• SDR设备初始化、采集流启动仅执行一次，作为后台任务持续运行，避免反复开关硬件带来的异常和延迟
• 移除错误的事件循环手动操作逻辑，完全依托asyncio.run()管理事件循环生命周期，符合Python异步编程的标准写法
• 增加完整的资源回收逻辑，无论程序正常结束还是抛出异常，都能正确关闭SDR流、释放硬件占用

内容的提问来源于stack exchange，提问作者Mohammed Ayoub Toubal