多任务向同一asyncio Queue推送数据是否需加asyncio.Lock？

场景与示例代码 #

3个异步Worker任务将结果推送至同一队列，另有一个任务处理队列中的数据，示例代码如下：

import asyncio

async def do_some_work(param):
    # 模拟耗时工作
    await asyncio.sleep(0.1)
    return f"result from worker {param}"

async def handle_result(res):
    # 模拟处理结果
    print(f"Handled: {res}")
    await asyncio.sleep(0.05)

async def worker1(queue: asyncio.Queue):
    while True:
        res = await do_some_work(param=1)
        await queue.put(res)

async def worker2(queue: asyncio.Queue):
    while True:
        res = await do_some_work(param=2)
        await queue.put(res)

async def worker3(queue: asyncio.Queue):
    while True:
        res = await do_some_work(param=3)
        await queue.put(res)

async def handle_results(queue: asyncio.Queue):
    while True:
        res = await queue.get()
        await handle_result(res)
        queue.task_done()

async def main():
    queue = asyncio.Queue()
    t1 = asyncio.create_task(worker1(queue))
    t2 = asyncio.create_task(worker2(queue))
    t3 = asyncio.create_task(worker3(queue))
    handler = asyncio.create_task(handle_results(queue))
    
    while True:
        # 执行其他逻辑
        await asyncio.sleep(1)

asyncio.run(main())

疑问点 #

文档指出asyncio.Queue并非线程安全，但本场景中所有任务均运行在同一线程的异步事件循环中。那么当3个任务向该队列推送数据时，是否需要使用asyncio.Lock来保护队列？查看Python 3.12的实现（推入队列前会创建putter future并等待），推测无需加锁，但对此不确定且文档未提及该场景，故有此疑问。

不需要额外使用asyncio.Lock来保护asyncio.Queue，原因如下：

• asyncio.Queue是专门为单线程异步环境设计的同步组件，内部已经通过异步安全的机制（比如putter/getter future的等待逻辑）处理了多个异步任务并发调用方法的场景，确保队列操作的原子性。
• 异步任务的并发是协作式调度，asyncio.Queue的put()、get()等方法在执行关键操作时，会通过内部状态管理和future等待，避免多个任务同时修改队列内部数据，不会出现数据竞争问题。
• 文档中提到的“非线程安全”，特指跨线程调用的场景（比如在非asyncio管理的线程中直接调用队列方法），而同一事件循环下的异步任务之间使用asyncio.Queue是完全安全的，不需要额外加锁。

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