如何将asyncio.Event变量传递给非异步回调函数？

你遇到的LookupError是因为**ContextVar的上下文绑定特性**：ContextVar的值仅在设置它的执行上下文（比如当前asyncio任务、线程）中有效，而LoRa的on_rx_done回调是由硬件中断触发，运行在独立的系统线程中，这个线程没有继承你在superloop任务里设置的ContextVar上下文，因此无法获取到绑定的asyncio.Event实例。

另外需要注意：asyncio.Event的set()方法不是线程安全的，即使拿到实例，也不能在非asyncio线程直接调用，必须通过事件循环的线程安全接口操作。

方案1：直接将Event绑定到LoRa类实例（推荐） #

通过修改自定义LoRa类的初始化逻辑，直接把asyncio.Event实例传递给类并保存为属性，同时用事件循环的线程安全方法触发事件：

修改后的核心代码 #

import asyncio
from SX127x.LoRa import *
from SX127x.LoRaArgumentParser import LoRaArgumentParser
from SX127x.board_config import BOARD

class LoRaRcvCont(LoRa):
    server_status = False

    def __init__(self, verbose=False, thing_event=None):
        super(LoRaRcvCont, self).__init__(verbose)
        self.set_mode(MODE.SLEEP)
        self.set_dio_mapping([0] * 6)
        self.thing_event = thing_event
        self.loop = asyncio.get_event_loop()  # 保存事件循环引用

    def on_rx_done(self):
        print("LoRa RxDone")
        if self.thing_event:
            # 用线程安全方式调用Event.set()
            self.loop.call_soon_threadsafe(self.thing_event.set)
        self.clear_irq_flags(RxDone=1)

def lora_config_setup(thing_event):
    parser = LoRaArgumentParser("Continous LoRa receiver.")
    lora = LoRaRcvCont(verbose=False, thing_event=thing_event)
    args = parser.parse_args(lora)
    assert (lora.get_agc_auto_on() == 1)
    lora.reset_ptr_rx()
    lora.set_mode(MODE.RXCONT)
    return lora

async def superloop():
    thing = asyncio.Event()
    global lora
    lora = lora_config_setup(thing)  # 传入Event实例
    print("asyncio loop starts ...")
    await asyncio.gather(asyncio.create_task(task_1(thing)))

async def task_1(thing):
    count = 0
    while True:
        print("waiting for thing")
        await thing.wait()
        print("task1: got event thing set")
        count += 1
        await asyncio.sleep(1)
        print("task 1: reset event thing set")
        thing.clear()

if __name__ == "__main__":
    print('code execution started')
    BOARD.setup()
    print("\nasync superloop code running")
    asyncio.run(superloop())
    print("code finished")

关键说明 #

• 直接通过类实例传递Event，避免了ContextVar的上下文绑定限制
• 使用loop.call_soon_threadsafe()确保跨线程操作asyncio.Event的安全性

方案2：用线程安全队列传递信号 #

如果不想修改LoRa类的初始化逻辑，可以通过threading.Queue作为中间层传递信号，在asyncio任务中监听队列并触发事件：

修改后的核心代码 #

import asyncio
import threading
from SX127x.LoRa import *
from SX127x.LoRaArgumentParser import LoRaArgumentParser
from SX127x.board_config import BOARD

# 全局线程安全队列
signal_queue = threading.Queue()

class LoRaRcvCont(LoRa):
    server_status = False

    def __init__(self, verbose=False):
        super(LoRaRcvCont, self).__init__(verbose)
        self.set_mode(MODE.SLEEP)
        self.set_dio_mapping([0] * 6)

    def on_rx_done(self):
        print("LoRa RxDone")
        signal_queue.put("rx_trigger")  # 发送信号到队列
        self.clear_irq_flags(RxDone=1)

async def queue_listener(thing):
    while True:
        await asyncio.to_thread(signal_queue.get)  # 在线程中等待队列消息
        thing.set()

async def superloop():
    thing = asyncio.Event()
    print("asyncio loop starts ...")
    await asyncio.gather(
        asyncio.create_task(task_1(thing)),
        asyncio.create_task(queue_listener(thing))  # 启动队列监听任务
    )

# 其余task_1、lora_config_setup、main函数保持不变

关键说明 #

• 利用threading.Queue实现跨线程信号传递
• 通过asyncio.to_thread()在异步任务中安全监听队列

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