控制asyncio协程的调度优先级可能吗？

Question

有没有办法控制准备运行的所有协同程序中的调度优先级？

具体来说,我有几个协同程序处理从网络到多个队列的流I/O,第二组协同程序将数据从队列中摄取到数据结构中.这些摄取协同程序发出第三组协同程序的信号,这些协同程序在每次摄取新数据时都会分析该数据结构.

来自网络的数据到达是具有非确定性消息速率的无限流.我希望分析步骤在新数据到达时立即运行,但不要在处理所有待处理数据之前运行.我看到的问题是,根据调度的顺序,分析协程可以在读者协程之前运行,也可以准备好数据,因此分析协程甚至无法检查摄取队列中的待处理数据,因为它可能尚未被读取即使这些读者协同程序已准备好运行,但网络还没有.

一种解决方案可能是将协同程序组织成优先级组,以便在分析协同程序之前始终安排读取程序协同程序,如果它们都能够运行,但我没有看到这样做的方法.

是否有asyncio的功能可以实现此优先级？或者我可能会问错误的问题,我可以重新设计协程,这样就不会发生这种情况(但我没有看到).

- 编辑 -

基本上我有一个N协程,看起来像这样:

while True:
  data = await socket.get()
  ingestData(data)
  self.event.notify()

所以我遇到的问题是,我无法知道任何其他N-1套接字在执行此协程时已准备好数据,因此我无法知道是否应通知该事件.如果我可以将这些协同程序优先于分析协程(正在等待self.event.wait()),那么我可以确定在安排分析协程时它们都不可运行.

Answer 1

asyncio不支持显式指定协程优先级，但使用库提供的工具可以直接实现相同的效果。鉴于您问题中的示例：

async def process_pending():
    while True:
        data = await socket.get()
        ingestData(data)
        self.event.notify()

您可以直接使用 等待套接字asyncio.wait，然后您就会知道哪些套接字是可操作的，并且只有在所有处理完成后才通知分析器。例如：

async def process_pending():
    while True:
        data = await socket.get()
        ingestData(data)
        self.event.notify()