带有异步队列的 Python CancelledError

如何解决带有异步队列的 Python CancelledError

我使用 this 答案中的代码，但在队列为空时得到 asyncio.exceptions.CancelledError。在实际项目中，我将任务从消费者添加到队列中，这就是为什么我使用 while True 语句

我压缩该代码以使调试更容易：

import asyncio
import traceback


async def consumer(queue: asyncio.Queue):
    try:
        while True:
            number = await queue.get()  # here is exception
            queue.task_done()
            print(f'consumed {number}')
    except BaseException:
        traceback.print_exc()


async def main():
    queue = asyncio.Queue()
    for i in range(3):
        await queue.put(i)
    consumers = [asyncio.create_task(consumer(queue)) for _ in range(1)]
    await queue.join()
    for c in consumers:
        c.cancel()


asyncio.run(main())

和错误：

consumed 0
consumed 1
consumed 2
Traceback (most recent call last):
  File "/Users/abionics/Downloads/BaseAsyncScraper/ttt.py",line 8,in consumer
    number = await queue.get()
  File "/usr/local/Cellar/python@3.9/3.9.4/Frameworks/Python.framework/Versions/3.9/lib/python3.9/asyncio/queues.py",line 166,in get
    await getter
asyncio.exceptions.CancelledError

顺便说一下，queue.get() 的文档说 If queue is empty,wait until an item is available。这个错误的真正原因是什么？也许有更好的解决方案？

解决方法

原因是因为你取消了任务：

Task.cancel：

请求取消任务。

这安排了一个 CancelledError 异常被抛出到 在事件循环的下一个循环中包装协程。

您有几个选择来处理这个问题：

1.使用 asyncio.gather

如果 return_exceptions 为 True，则异常的处理方式与 成功的结果，并汇总在结果列表中。

await queue.join()

for c in consumers:
    c.cancel()

await asyncio.gather(*consumers,return_exceptions=True)

2.在消费者中捕获异常

async def consumer(q):
    while True:
        try:
            num = await q.get()
            print(f"Working on: {num}")
        except asyncio.CancelledError:
            print(f"Exiting...")
            break
        else:
            q.task_done()

3.抑制异常

form contextlib import suppress

async def consumer(q):
    with suppress(asyncio.CancelledError):
        while True:
            num = await q.get()
            print(f"Working on: {num}")
            q.task_done()