可以控制异步协程的调度优先级吗？

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 等待套接字，然后您就会知道哪些套接字是可操作的，并且只有在所有套接字都处理完毕后才通知分析器。例如：

def _read_task(self, socket):
    loop = asyncio.get_event_loop()
    task = loop.create_task(socket.get())
    task.__process_socket = socket
    return task

async def process_pending_all(self):
    tasks = {self._read_task(socket) for socket in self.sockets}
    while True:
        done, not_done = await asyncio.wait(
            tasks, return_when=asyncio.FIRST_COMPLETED)
        for task in done:
            ingestData(task.result())
            not_done.add(self._read_task(task.__process_socket))
        tasks = not_done
        self.event.notify()