我想要一個用于“動作”執行的異步調度程序，它滿足某些屬性：操作是一次性的，并且按照精確的時間戳進行安排。操作應嚴格按順序執行，即調度程序在前一個操作完成執行之前無法啟動下一個操作。在執行動作之間，當調度程序等待下一個時間戳時，調度程序必須處于 狀態，asyncio.sleep()以便讓其他協程輪流執行。當調度新的動作時，調度器應立即重新調整其等待時間，以便調度器始終等待最快的動作。當沒有調度任何動作時，調度程序應該處于永久狀態asyncio.sleep()，直到添加新動作。我的嘗試：import asyncioimport timeclass 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）！你能給我推薦一些東西嗎？
1 回答
動漫人物
TA貢獻1815條經驗 獲得超10個贊
asyncio 事件循環已經包含您嘗試實現的代碼 - 排序超時并等待任務提交。您需要使接口適應Scheduler底層 asyncio 功能，例如如下所示：
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()
添加回答
舉報
0/150
提交
取消