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 的接口适应底层的异步功能，例如：

class Scheduler:
    def __init__(self):
        self._running = asyncio.Lock()

    async def start(self):
        pass  # asyncio event loop will do the actual work

    def add(self, action):
        loop = asyncio.get_event_loop()
        # Can't use `call_at()` because event loop time uses a
        # different timer than time.time().
        loop.call_later(
            action.timestamp - time.time(),
            loop.create_task, self._execute(action)
        )

    async def _execute(self, action):
        # Use a lock to ensure that no two actions run at
        # the same time.
        async with self._running:
            await action.do()