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

Question

我正在寻找一个pythonicasyncio“模式”，用于在我的程序中经常出现的构造

辅助任务执行一些通常由几个步骤组成的操作。这些操作的细节由从控制函数发送到辅助任务的命令控制。在各个步骤之间有休眠，工作人员只能在这些休眠期间接受新命令。新命令应立即将工作任务从睡眠中唤醒

这些命令表示所需的目标状态。我在用队列进行通讯。但是，只能有一个目标，这就是为什么命令不能构建真正的队列，但最后一个目标会替换以前的所有目标。队列最多有一个项目

目前我正在使用其他异步库。我想切换到标准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

但我想也许有一个更简单的解决办法。如果你有异步的经验，并认为队列超时是一种方法，这将帮助我太多

我试图搜索，但找不到合适的关键字为我的问题（同样的问题标题举行）

Answer 1

您的超时实现肯定没有问题。asynciosynchronization原语上的方法故意不支持显式超时参数，让调用者在需要时使用取消或wait_for超时

至于一个单元素队列，我会考虑用^{}替换它，它不仅被设计用来保存单个值，而且在asyncio中也是非常轻量级的，因为它是用来构建几乎所有其他东西的基本抽象

代替wait_for(cmd_queue.get(), ...)，你要写wait_for(cmd_future, ...)，代替cmd_queue.put(value)，你要写cmd_future.set_result(value)。唯一重要的区别是未来是一次性的，因此在获得一个项目之后，您需要为cmd_future分配一个新的未来