跨不同类的Python的asyncio.Event()

Question

我正在编写一个Python程序来与基于CAN总线的设备进行交互。为此，我成功地使用了python模块。我还使用异步对异步事件作出反应。我编写了一个"CanBusManager“类，这个类由"CanBusSequencer”类使用。"CanBusManager“类负责生成/发送/接收消息，CanBusSequencer驱动要发送的消息序列。

在序列中的某个时刻，我希望等到接收到特定消息时“解锁”要在序列中发送的其余消息。代码概述：

main.py

async def main():
   
   event = asyncio.Event()
   sequencer = CanBusSequencer(event)
   task = asyncio.create_task(sequencer.doSequence())
   await task
 
asyncio.run(main(), debug=True)

canBusSequencer.py

from canBusManager import CanBusManager

class CanBusSequencer:
 
   def __init__(self, event)
 
      self.event = event
      self.canManager = CanBusManager(event)

   async def doSequence(self):
 
      for index, row in self.df_sequence.iterrows():
         if:...
            self.canManager.sendMsg(...)
         else:
            self.canManager.sendMsg(...)
            await self.event.wait()
            self.event.clear()

canBusManager.py

import can

class CanBusManager():
 
   def __init__(self, event):
 
      self.event = event
      self.startListening()
 
 **EDIT**
    def startListening(self):
    
       self.msgNotifier = can.Notifier(self.canBus, self.receivedMsgCallback)
 **EDIT**
 
   def receivedMsgCallback(self, msg):
 
      if(msg == ...):
         self.event.set()

现在，即使接收到相关消息并执行self.event.set()，我的程序仍然停留在等待self.event.set()的位置。使用debug = True运行程序会发现

RuntimeError: Non-thread-safe operation invoked on an event loop other than the current one

我不太明白。它与异步事件循环有关，在某种程度上没有正确定义/管理。我来自C++世界，目前正在用Python编写我的第一个大型程序。如能提供任何指导，将不胜感激：)

Answer 1

您的问题无法解释如何安排调用receivedMsgCallback。

如果使用后台线程的经典“异步”API调用它，那么它将从运行事件循环的线程外部调用。根据文献资料的说法，异步原语是而不是线程安全的，因此从另一个线程调用event.set()不能与正在运行的事件循环正确同步，这就是为什么您的程序应该在什么时候才会醒来的原因。

如果您希望从事件循环线程之外执行任何与异步相关的操作，例如调用Event.set，则需要使用call_soon_threadsafe或等效的。例如：

    def receivedMsgCallback(self, msg):
        if msg == ...:
            self.loop.call_soon_threadsafe(self.event.set)

事件循环对象应该提供给CanBusManager对象，也许可以将它传递给它的构造函数并将它分配给self.loop。

另外，如果您创建一个任务只是为了立即等待它，那么首先您不需要任务。换句话说，您可以用更简单的task = asyncio.create_task(sequencer.doSequence()); await task代替await sequencer.doSequence()。