我想要一個用于“動作”執(zhí)行的異步調(diào)度程序，它滿足某些屬性：操作是一次性的，并且按照精確的時間戳進(jìn)行安排。操作應(yīng)嚴(yán)格按順序執(zhí)行，即調(diào)度程序在前一個操作完成執(zhí)行之前無法啟動下一個操作。在執(zhí)行動作之間，當(dāng)調(diào)度程序等待下一個時間戳?xí)r，調(diào)度程序必須處于 狀態(tài)，asyncio.sleep()以便讓其他協(xié)程輪流執(zhí)行。當(dāng)調(diào)度新的動作時，調(diào)度器應(yīng)立即重新調(diào)整其等待時間，以便調(diào)度器始終等待最快的動作。當(dāng)沒有調(diào)度任何動作時，調(diào)度程序應(yīng)該處于永久狀態(tài)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這個實現(xiàn)不可靠，并且沒有考慮我尋求的條件（5）！你能給我推薦一些東西嗎？
1 回答
動漫人物
TA貢獻(xiàn)1815條經(jīng)驗 獲得超10個贊
asyncio 事件循環(huán)已經(jīng)包含您嘗試實現(xiàn)的代碼 - 排序超時并等待任務(wù)提交。您需要使接口適應(yīng)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
提交
取消