asyncio.Queue无移除访问、清空、左端插入的实现方案对比

方案优劣对比 #

第一种（公开API实现）：不推荐 #

• 性能低下：clear、appendleft、get_all均为O(n)时间复杂度，队列元素数量较多时会产生不必要的性能开销，完全浪费了deque原生O(1)操作的优势。
• 逻辑存在隐性bug：所有操作中调用get_nowait()后都执行了self.task_done()，但这些元素并未被实际消费，只是临时取出后放回，会干扰依赖join()的逻辑，导致join()提前返回，产生难以排查的并发问题。
• 接口不友好：appendleft被设计为异步方法，需要额外await调用，不符合该操作的语义定位。

第二种（直接操作私有_queue属性）：推荐使用 #

优点

• 性能拉满：所有操作直接调用deque原生方法，clear、appendleft为O(1)时间复杂度，get_all仅需一次遍历复制，性能远优于第一种方案。
• 逻辑无副作用：仅操作底层存储队列，不会修改asyncio.Queue的未完成任务计数，也不会干扰put()/get()的阻塞逻辑，join()、阻塞获取等原生功能完全不受影响。
• 接口符合预期：三个方法均为同步方法，调用逻辑简单。

缺点

依赖asyncio.Queue的私有实现，官方未承诺_queue属性名永久不变，但从Python 3.4 asyncio正式加入到当前最新的3.12版本，该底层实现从未发生过变动，工业界大量开源项目都采用该实现方式，稳定性有足够保障。 #

优化建议 #

第二种方案已经是当前场景下的最优实现，可补充少量兼容性校验逻辑提升鲁棒性：

import asyncio
from collections import deque

class BlockingDeque(asyncio.Queue):
    def __init__(self, maxsize=0):
        super().__init__(maxsize)
        # 校验底层实现，避免不兼容的Python版本报错
        assert isinstance(self._queue, deque), "当前Python版本的asyncio.Queue底层实现非deque，不兼容该扩展"
    
    def clear(self):
        self._queue.clear()

    def get_all(self):
        return self._queue.copy()

    def appendleft(self, x):
        self._queue.appendleft(x)

注意事项：由于deque的原生方法均为CPU原子操作，在协程场景下不会中途让出执行权，不需要额外加锁即可保证并发安全。

内容的提问来源于stack exchange，提问作者upe