Python：如何实现顺序执行的异步调度程序？

Question

我想要一个用于“动作”执行的异步调度程序，它满足某些属性：

1. 操作是一次性的，并且按照精确的时间戳进行安排。

2. 操作应严格按顺序执行，即调度程序在前一个操作完成执行之前无法启动下一个操作。

3. 在执行动作之间，当调度程序等待下一个时间戳时，调度程序必须处于 状态，asyncio.sleep()以便让其他协程轮流执行。

4. 当调度新的动作时，调度器应立即重新调整其等待时间，以便调度器始终等待最快的动作。

5. 当没有调度任何动作时，调度程序应该处于永久状态asyncio.sleep()，直到添加新动作。

我的尝试：

import asyncio

import time

class Action:

    def __init__(self, timestamp):

        self.timestamp = timestamp

    async def do(self):

        print("Doing action...")

class Scheduler:

    def __init__(self):

        self._actions = []

        self._sleep_future = None

    def add(self, action):

        self._actions.append(action)

        self._actions.sort(key=lambda x: x.timestamp)

        if self._sleep_future:

            self._sleep_future.cancel()

    def pop(self):

        return self._actions.pop(0)

    async def start(self):

        asyncio.create_task(self.loop())

    async def loop(self):

        while True:

            now = time.time()            

            while self._actions:

                action = self._actions[0]

                if action.timestamp <= now:

                    action = self.pop()                

                    await action.do()                   

                else:

                    break

            self._sleep_future = asyncio.ensure_future(

                asyncio.sleep(self._actions[0].timestamp - now)

            )

            try:

                await self._sleep_future

            except asyncio.CancelledError:

                continue

            finally:

                self._sleep_future = None

这个实现不可靠，并且没有考虑我寻求的条件（5）！

你能给我推荐一些东西吗？

Answer 1

asyncio 事件循环已经包含您尝试实现的代码 - 排序超时并等待任务提交。您需要使接口适应Scheduler底层 asyncio 功能，例如如下所示：class Scheduler:&nbsp; &nbsp; def __init__(self):&nbsp; &nbsp; &nbsp; &nbsp; self._running = asyncio.Lock()&nbsp; &nbsp; async def start(self):&nbsp; &nbsp; &nbsp; &nbsp; pass&nbsp; # asyncio event loop will do the actual work&nbsp; &nbsp; def add(self, action):&nbsp; &nbsp; &nbsp; &nbsp; loop = asyncio.get_event_loop()&nbsp; &nbsp; &nbsp; &nbsp; # Can't use `call_at()` because event loop time uses a&nbsp; &nbsp; &nbsp; &nbsp; # different timer than time.time().&nbsp; &nbsp; &nbsp; &nbsp; loop.call_later(&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; action.timestamp - time.time(),&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; loop.create_task, self._execute(action)&nbsp; &nbsp; &nbsp; &nbsp; )&nbsp; &nbsp; async def _execute(self, action):&nbsp; &nbsp; &nbsp; &nbsp; # Use a lock to ensure that no two actions run at&nbsp; &nbsp; &nbsp; &nbsp; # the same time.&nbsp; &nbsp; &nbsp; &nbsp; async with self._running:&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; await action.do()