1 项异步队列 - 这是标准的东西吗？

Question

在我的一个 asyncio 项目中，我经常使用一种同步方法，并且想知道它是否是某种标准工具，其名称我可以给谷歌提供以了解更多信息。我使用“1 项队列”一词只是因为我没有更好的名称。这是一个降级队列，与

Queue(maxsize=1)

无关。

#  [controller] ---- commands ---> [worker]

控制器向工作人员发送命令（

queue.put

，实际上是

put_nowait

），工作人员等待它们（

queue.get

）并执行它们，但特殊规则是，只有最后一个命令很重要，并立即替换所有命令之前未完成的命令。因此，队列中等待执行的命令永远不会超过 1 个。

为了实现这一点，控制器在放置之前清除队列。没有

queue.clear

，因此它必须丢弃（使用

get_nowait

）等待项目（如果有）。 （缺乏

queue.clear

让我产生了怀疑，导致了这个问题。）

在工作端，如果命令执行需要睡眠，则会被带有超时的

newcmd=queue.get

取代。当发生超时时，就是睡眠；当

get

成功时，当前工作将中止并开始执行

newcmd

。

Answer 1

您使用的队列类型不是标准的 - 存在一次性队列之类的东西，但它完全是另一回事。

队列并不真正适合您的用例，尽管您付出了一些努力使其工作。您实际上并不需要任何类型的排队，您需要一个容纳单个对象（可以替换）的槽和一个唤醒机制。

asyncio.Event

可用于唤醒，您可以将有效负载对象（命令）附加到事件的属性。例如：

async def worker(evt):
    while True:
        await evt.wait()
        evt.clear()
        if evt.last_command is None:
            continue
        last_command = evt.last_command
        evt.last_command = None
        # execute last_command, possibly with timeout
        print(last_command)

async def main():
    evt = asyncio.Event()
    workers = [asyncio.create_task(worker(evt)) for _ in range(5)]
    for i in itertools.count():
        await asyncio.sleep(1)
        evt.last_command = f"foo {i}"
        evt.set()

asyncio.run(main())

此方法与基于队列的方法之间的一个区别是，设置事件将唤醒所有工作人员（如果有多个工作人员），即使第一个工作人员立即调用

evt.clear()

。另一方面，队列项目将保证被移交给单个等待者

queue.get()

。

Answer 2

我还在使用 1 项异步队列并寻找更好的解决方案。 答案@user2357112中的“evt.last_command”是用户添加到标准类对象的属性。我不确定这是否是一个好的做法。或者我可以子类化 asyncio.Event 来添加我需要的属性吗？