使用asyncio.Queue和TaskGroup执行pytest时出现挂起/死锁问题

我是异步编程新手，正在学习如何将TaskGroup与asyncio.Queue结合使用。我编写了如下模块及测试用例，但执行pytest时，程序会打印队列中的所有项，之后就出现挂起/死锁的情况。请问我哪里出错了？

模块：AsynchronousQueueBeta.py

from asyncio import Queue, TaskGroup

class AsynchronousQueueBeta:
    """Asynchronous Queue Beta"""

    async def fetch_recursive(self, source_list: list[str], maximum_connection: int = 10):
        """Fetch Recursive"""
        print('Fetch Recursive')
        query_queue = Queue()

        for source in source_list:
            query_queue.put_nowait(source)

        async with TaskGroup() as group:
            task_list = [
                group.create_task(self.fetch_query(query_queue)) for _ in range(maximum_connection)
            ]

        await query_queue.join()
        result_list = [task.result() for task in task_list]
        print(f'Result List: {result_list}')


    async def fetch_query(self, queue: Queue):
        """Fetch Query"""
        while True:
            query = await queue.get()
            print(f'Query: {query}')
            queue.task_done()

测试：TestAsynchronousQueueBeta.py

import pytest
from AsynchronousQueueBeta import AsynchronousQueueBeta

class TestAsynchronousQueueBeta():
    """Test Asynchronous Queue Beta"""

    @pytest.mark.asyncio
    @pytest.mark.parametrize(
        'source_list', [
            [
                'https://httpbin.org/anything/1',
                'https://httpbin.org/anything/2',
                'https://httpbin.org/anything/3',
                'https://httpbin.org/anything/4',
                'https://httpbin.org/anything/5',
                'https://httpbin.org/anything/6',
                'https://httpbin.org/anything/7',
                'https://httpbin.org/anything/8',
                'https://httpbin.org/anything/9',
                'https://httpbin.org/anything/10',
                'https://httpbin.org/anything/11',
                'https://httpbin.org/anything/12',
            ],
        ]
    )
    async def test_fetch_recursive(self, source_list: list[str]):
        """Test Fetch Recursive"""
        beta = AsynchronousQueueBeta()

        await beta.fetch_recursive(
            source_list=source_list,
        )

执行结果

platform darwin -- Python 3.12.1, pytest-7.4.4, pluggy-1.3.0 -- /Users/abc/Desktop/Project/Workspace/Python/pv312/bin/python3.12
cachedir: .pytest_cache
rootdir: /Users/abc/Desktop/Project/Async
configfile: pytest.ini
plugins: asyncio-0.23.3, anyio-4.2.0
asyncio: mode=Mode.STRICT
collected 1 item

Test/TestAsynchronousQueueBeta.py::TestAsynchronousQueueBeta::test_fetch_recursive[source_list0] Fetch Recursive
Query: https://httpbin.org/anything/1
Query: https://httpbin.org/anything/2
Query: https://httpbin.org/anything/3
Query: https://httpbin.org/anything/4
Query: https://httpbin.org/anything/5
Query: https://httpbin.org/anything/6
Query: https://httpbin.org/anything/7
Query: https://httpbin.org/anything/8
Query: https://httpbin.org/anything/9
Query: https://httpbin.org/anything/10
Query: https://httpbin.org/anything/11
Query: https://httpbin.org/anything/12
^C

!!! KeyboardInterrupt !!!
/opt/python/3.12.1/lib/python3.12/selectors.py:566: KeyboardInterrupt
(to show a full traceback on KeyboardInterrupt use --full-trace)
...

核心问题 #

1. TaskGroup等待顺序错误：async with TaskGroup()块会等待所有子任务完成后才退出，但你的fetch_query是无限循环，永远不会结束，导致TaskGroup一直阻塞，程序卡死。
2. 工作任务无终止条件：队列中的任务处理完后，fetch_query还会卡在await queue.get()处等待新任务，永远无法退出。

修复方案 #

给工作任务添加终止信号，调整TaskGroup与队列操作的顺序，让任务能正常退出。

修改后的代码

from asyncio import Queue, TaskGroup

class AsynchronousQueueBeta:
    """Asynchronous Queue Beta"""

    async def fetch_recursive(self, source_list: list[str], maximum_connection: int = 10):
        """Fetch Recursive"""
        print('Fetch Recursive')
        query_queue = Queue()

        # 填充待处理任务
        for source in source_list:
            query_queue.put_nowait(source)
        
        # 添加与工作线程数一致的终止标记（用None作为信号）
        for _ in range(maximum_connection):
            query_queue.put_nowait(None)

        async with TaskGroup() as group:
            task_list = [
                group.create_task(self.fetch_query(query_queue)) for _ in range(maximum_connection)
            ]
        
        # TaskGroup已确保所有任务完成，直接收集结果
        result_list = [task.result() for task in task_list]
        print(f'Result List: {result_list}')


    async def fetch_query(self, queue: Queue):
        """Fetch Query"""
        task_results = []
        while True:
            query = await queue.get()
            if query is None:  # 收到终止信号，退出循环
                queue.task_done()
                break
            print(f'Query: {query}')
            task_results.append(query)  # 保存当前任务结果
            queue.task_done()
        return task_results  # 返回该工作线程的所有结果

关键修改点 #

• 给队列添加与工作线程数相同的None终止标记，确保每个工作任务都能收到退出信号
• fetch_query检测到终止信号时退出循环，返回收集的结果
• 移除await query_queue.join()，因为TaskGroup会自动等待所有任务完成

另一种保留queue.join()的写法

如果想继续使用queue.join()等待所有任务处理完成，可调整顺序：

async def fetch_recursive(self, source_list: list[str], maximum_connection: int = 10):
    print('Fetch Recursive')
    query_queue = Queue()

    for source in source_list:
        query_queue.put_nowait(source)

    async with TaskGroup() as group:
        task_list = [
            group.create_task(self.fetch_query(query_queue)) for _ in range(maximum_connection)
        ]
        
        await query_queue.join()  # 先等待所有队列任务处理完毕
        # 发送终止信号让工作任务退出
        for _ in range(maximum_connection):
            query_queue.put_nowait(None)
    
    result_list = [task.result() for task in task_list]
    print(f'Result List: {result_list}')

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