Python如何在不同线程中执行异步函数并按需获取返回值

• 原代码外层while循环每次迭代都会重新调用powers()，生成全新的异步生成器实例，重复初始化SDR硬件、新建采样流，之前未遍历完成的生成器没有被显式关闭，占用的硬件资源不会自动释放
• 内层async for循环的break仅能跳出当前轮次的生成器遍历，既没有终止正在运行的SDR采样流，也没有阻止外层循环再次启动新的powers()实例，逻辑上永远无法按预期终止
• 直接用threading模块包裹异步逻辑会触发asyncio事件循环的线程安全限制，跨线程直接调用协程方法会抛出运行时异常，无法正常工作

采用独立线程运行专属asyncio事件循环、线程安全队列做跨线程数据传递的架构，SDR采样流仅初始化一次，主线程无需感知异步逻辑，可随时获取最新功率采样值，停止时可保证硬件资源被正确释放。
可直接运行的修复代码如下：

import asyncio
import threading
import queue
from rtlsdr import RtlSdr
import numpy as np

# 跨线程通信队列，maxsize设为1仅保留最新采样值，避免数据积压
power_queue = queue.Queue(maxsize=1)
# 线程级停止信号
stop_event = threading.Event()

async def powers():
    sdr = RtlSdr()
    sdr.rs = 2.4e6
    sdr.fc = 801e6
    try:
        async for samples in sdr.stream(512):
            # 收到停止信号立即终止采样循环
            if stop_event.is_set():
                break
            samples = samples - np.mean(samples)
            power = np.mean(np.abs(samples ** 2))
            # 入队前先清空旧值，保证队列里永远是最新采样
            if not power_queue.empty():
                try:
                    power_queue.get_nowait()
                except queue.Empty:
                    pass
            power_queue.put(power)
    finally:
        # 异常/正常退出都保证释放SDR硬件资源
        await sdr.stop()
        sdr.close()

def sdr_sampling_worker():
    # 独立线程初始化专属事件循环，和主线程完全隔离
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    try:
        loop.run_until_complete(powers())
    finally:
        loop.close()

if __name__ == "__main__":
    nb_config = 1000
    fpower = -100
    i = 0
    # 启动采样守护线程，主程序退出时自动跟随终止
    sdr_thread = threading.Thread(target=sdr_sampling_worker, daemon=True)
    sdr_thread.start()

    try:
        while i < nb_config:
            # 主线程按需阻塞获取最新功率值，无需编写异步逻辑
            current_power = power_queue.get()
            print(f"采样序号: {i}")
            print(f"当前功率: {10 * np.log10(current_power):.2f} dB")
            if current_power > fpower:
                fpower = current_power
            i += 1
        print(f"采样完成，记录到的最大功率: {10 * np.log10(fpower):.2f} dB")
    finally:
        # 触发停止信号，等待采样线程干净退出
        stop_event.set()
        sdr_thread.join(timeout=2)

• 异步采样逻辑完全运行在独立线程的专属事件循环中，和主线程业务逻辑完全解耦，不会触发asyncio线程安全报错
• 固定大小为1的传输队列会自动丢弃过期采样值，避免采样速度快于主线程处理速度时出现数据积压，保证主线程每次拿到的都是最新的功率数据
• 基于线程Event实现停止信号，触发后会主动跳出SDR采样循环，finally块强制保证SDR硬件资源被正确释放，不会出现设备被占用无法重新连接的问题
• 主线程无需编写任何async/await异步代码，需要功率值时直接调用power_queue.get()即可，完全符合按需归集异步函数返回值的需求
• 移除了原代码多余的外层while循环，SDR设备和采样流全程仅初始化一次，从根源解决了循环无法按break条件终止的问题

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