python - 在 asyncio sleep 期间检查新的控制命令

Question

我正在为我的程序中经常出现的构造寻找Pythonic异步“模式”。

辅助任务执行一些通常由多个步骤组成的操作。这些操作的细节由从控制功能发送到工作任务的命令控制。各个步骤之间存在 sleep ，并且工作人员仅在这些 sleep 期间才能接受新命令。新命令应立即将工作任务从 sleep 状态唤醒。

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

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

但我认为也许有一个更简单的解决方案。 OTOH 如果您有 asyncio 经验并且认为带有超时的队列是可行的方法，那也会对我有帮助。

我尝试搜索，但找不到适合我的问题的关键字(问题标题也是如此)。

Answer 1

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

对于单元素队列，我会考虑用Future替换它。 ，它不仅设计用于保存单个值，而且在 asyncio 中也非常轻量级，因为它是用于构建几乎所有其他内容的基本抽象。

您可以编写wait_for(cmd_future, ...)，而不是wait_for(cmd_queue.get(), ...)，并且代替cmd_queue.put(value)，您可以编写cmd_future.set_result(value)。唯一重要的区别是 future 是一次性的，因此在获得一个项目后，您需要为 cmd_future 分配一个新的 future。