如何在Python asyncio并行场景下向列表追加异步返回的元素

核心思路 #

原代码在for循环内每次await异步操作，所以会等前一个任务完成才启动下一个，属于串行执行。要改成并行，只需要把每个符号对应的异步操作封装为独立协程，先批量生成所有任务再用asyncio.gather统一执行即可，同时保留原有异常自动跳过的逻辑。

最终实现代码 #

import asyncio

async def __create_sockets(self):
    # 单个符号的处理逻辑，单独封装做异常捕获
    async def _fetch_single_socket(symbol):
        try:
            print(f'Collecting data of {symbol}')
            socket = DepthCacheManager(self.client, symbol, refresh_interval=None)
            return await socket.__aenter__()
        except:
            # 异常场景返回None，后续过滤即可
            return None
    
    # 先生成所有协程任务，此时还未执行
    all_tasks = [_fetch_single_socket(s) for s in self.symbols.index]
    # 并行执行所有任务，等待全部完成后返回结果列表
    task_results = await asyncio.gather(*all_tasks)
    # 过滤掉异常返回的None，和原逻辑输出完全一致
    return [res for res in task_results if res is not None]

可选优化：限制最大并发数 #

如果你的symbols数量过多，直接全量并行可能触发服务端连接限制、本地端口占满等问题，可以加asyncio.Semaphore控制最大同时执行的任务数，示例如下（限制最多20个并发）：

import asyncio

async def __create_sockets(self):
    # 最多同时执行20个任务，可根据实际情况调整数值
    max_concurrency = asyncio.Semaphore(20)
    
    async def _fetch_single_socket(symbol):
        # 拿到信号量才会执行后续逻辑，超过并发数的任务会排队等待
        async with max_concurrency:
            try:
                print(f'Collecting data of {symbol}')
                socket = DepthCacheManager(self.client, symbol, refresh_interval=None)
                return await socket.__aenter__()
            except:
                return None
    
    all_tasks = [_fetch_single_socket(s) for s in self.symbols.index]
    task_results = await asyncio.gather(*all_tasks)
    return [res for res in task_results if res is not None]

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