在异步睡眠期间检查新的控制命令

我正在寻找一个pythonic asyncio“模式”,用于在我的程序中经常出现的构造。

工作任务执行一些操作,通常由几个步骤组成。这些操作的详细信息由从控制函数发送到辅助任务的命令控制。在各个步骤之间存在睡眠,并且工作进程只能在这些睡眠期间接受新命令。新命令应立即将工作任务从睡眠中唤醒。

这些命令表示所需的目标状态。我在用队列来交流。但是,只能有一个目标,这就是为什么命令不构建真正的队列,而最后一个目标将替换所有先前的队列。队列最多有一个项目。

目前我正在使用其他异步库。我想换成标准的 asyncio . 例如:

# warning: not asyncio code; not real code
cmd_queue = Queue()

async def worker():
    cmd = 'INIT'
    while cmd != 'STOP':
        ... do_something1 sync or async ...
        newcmd = await cmd_queue.get(timeout=SLEEPTIME1, timeout_value=None)
        if newcmd is not None:
            cmd = newcmd
            continue
        ... do_something2 sync or async ...
        newcmd = await cmd_queue.get(timeout=SLEEPTIME2, timeout_value=None)
        if newcmd is not None:
            cmd = newcmd
            continue

def controlloler():
    ...
    if newcmd:
        cmd_queue.clear() # replaces a waiting command
        cmd_queue.put(newcmd) # put_nowait() in asyncio
    ...

我可以重写 queue.get 异步代码:

try:
    cmd=wait_for(cmd_queue.get(), timeout=SLEEPTIME)
    continue # or process otherwise
except asyncio.TimeoutError:
    pass

但我想也许有一个更简单的解决办法。如果你有异步的经验,并且认为有超时的队列是可行的,那对我也有帮助。

我试图搜索,但找不到适合我的问题的关键字(同样的问题标题)。