Python asyncio.TaskGroup 实战：高并发采集API数据的正确姿势

年初接到一个需求，要从第三方数据平台拉取几千只股票的历史行情，每条数据都要单独调一次API。一开始想着简单，写了个循环一条一条拉，结果跑了将近十分钟。后来改成asyncio.gather加aiohttp，速度上来了但任务管理乱成一团——某个协程出了错其他协程还在继续跑，想中途取消整个采集过程还得手动遍历task列表。最头疼的是，在并发量大的时候偶尔会触发接口限流，抛出的异常把整个gather都带崩了。

切换到asyncio.TaskGroup后这个问题才算利索地解决了。TaskGroup是Python 3.11引入的特性，3.12和3.13进一步稳定，它借鉴了结构化并发的思想，让一组异步任务形成一个明确的作用域——组内任何一个任务异常，同组的所有其他任务都会被取消，不用再担心部分任务失控。配合Semaphore做并发控制，整个采集过程既快又稳。这篇文章就用这个实际案例，从零搭一个高并发采集程序，把TaskGroup的用法一步步拆解清楚。

先看看传统asyncio写法有什么不便

在TaskGroup出现之前，主流写法是用asyncio.gather或者手动创建asyncio.Task然后逐个await。拿采集行情数据来说，一般会这样写：

import asyncio
import aiohttp

async def fetch_price(session, symbol):
    url = f'https://api.example.com/price/{symbol}'
    async with session.get(url) as resp:
        return await resp.json()

async def main():
    symbols = ['AAPL', 'GOOGL', 'MSFT', ...]  # 几百个
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_price(session, s) for s in symbols]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        # 处理结果，区分正常数据和异常
        for symbol, result in zip(symbols, results):
            if isinstance(result, Exception):
                print(f'{symbol} 失败: {result}')
            else:
                print(f'{symbol}: {result}')

这里就有几个隐性问题。第一，gather默认是“一错全停”的，如果某个任务抛出异常且没有设置return_exceptions=True，整个gather会立刻取消所有其他还在运行的任务。可一旦设了return_exceptions=True，又不知道哪些任务被取消了哪些正常结束了，需要手动排查。第二，如果想在采集过程中根据已经获取到的数据决定是否提前停止（比如发现当日数据已经足够），gather没有提供优雅的中断方式，只能暴力地调cancel，然后逐个处理CancelledError。

TaskGroup提供了更明确的边界和更安全的结束机制。

用TaskGroup重构采集流程

TaskGroup通过async with语法创建一个任务组。组内的所有任务在退出async with块时会自动等待完成。如果组内任何一个任务以异常结束，组内的其他任务都会被取消，异常会传播到async with之外。这种“要么全部成功，要么全部取消”的语义非常适合一组彼此独立的网络请求——你不希望一部分数据已经拉回来了另一部分还在半路上跑着，最终得到的是不完整的结果集。

先把核心的获取函数写出来：

import asyncio
import aiohttp

async def fetch_price(session, symbol, semaphore):
    url = f'https://api.example.com/price/{symbol}'
    # 用信号量控制并发数，避免触发接口限流
    async with semaphore:
        try:
            async with session.get(url, timeout=aiohttp.ClientTimeout(total=5)) as resp:
                if resp.status == 200:
                    data = await resp.json()
                    return {'symbol': symbol, 'price': data['price']}
                else:
                    raise Exception(f'{symbol} 返回状态码 {resp.status}')
        except asyncio.TimeoutError:
            raise Exception(f'{symbol} 请求超时')

然后是主函数，用TaskGroup管理所有任务：

async def main():
    symbols = ['AAPL', 'GOOGL', 'MSFT', ...]  # 假设有几百个
    semaphore = asyncio.Semaphore(20)  # 最多同时20个请求

    async with aiohttp.ClientSession() as session:
        async with asyncio.TaskGroup() as tg:
            # 为每个symbol创建一个任务
            tasks = [
                tg.create_task(fetch_price(session, s, semaphore))
                for s in symbols
            ]
        # 离开TaskGroup时，所有任务都已经完成
        # 如果有任务抛异常，会在这里看到异常群

    # 收集结果：从已完成的任务里拿数据
    results = []
    for t in tasks:
        try:
            result = t.result()  # 获取返回值
            results.append(result)
        except Exception as e:
            print(f'任务失败: {e}')

    return results

当async with asyncio.TaskGroup() as tg块结束时，Python会确保组内所有任务都已结束。如果某个任务因为接口限流抛出了异常，TaskGroup会取消组内所有其他还未完成的任务，然后重新抛出异常。这一行为意味着如果整个批次中有一个失败，这一整批都可以被放弃或重试，而不会出现部分数据成功、部分数据还在跑的局面。

但在实际生产中，我们通常不希望一个失败的请求就断送掉整个批次，所以需要把异常捕获往任务内部收一收，让TaskGroup只关心真正的结构性错误（比如网络完全断开），而对单个接口的偶发失败做本地处理。这可以通过在fetch_price内部捕获异常并返回一个特殊结果（比如None）来实现，从而避免异常穿透到TaskGroup层。

处理部分失败与结果收集

改进后的fetch_price：

async def fetch_price(session, symbol, semaphore):
    url = f'https://api.example.com/price/{symbol}'
    async with semaphore:
        try:
            async with session.get(url, timeout=aiohttp.ClientTimeout(total=5)) as resp:
                if resp.status == 200:
                    data = await resp.json()
                    return {'symbol': symbol, 'price': data['price']}
                else:
                    print(f'{symbol} 状态异常: {resp.status}')
                    return None
        except (asyncio.TimeoutError, aiohttp.ClientError) as e:
            print(f'{symbol} 网络错误: {e}')
            return None

现在单个请求失败不会搞垮整个任务组。TaskGroup仍然会在所有任务都完成后退出块。结果收集也简单了：

results = [t.result() for t in tasks if t.result() is not None]
print(f'成功采集 {len(results)} 条数据')

加入限流和自动重试

接口限流是采集任务绕不开的问题。上面的Semaphore(20)控制了并发数上限，但有时候远程接口会返回429 Too Many Requests，要求稍后再试。可以在fetch_price里加一个简单的重试逻辑，遇到429就等几秒再试：

async def fetch_price(session, symbol, semaphore, retries=2):
    url = f'https://api.example.com/price/{symbol}'
    for attempt in range(retries + 1):
        async with semaphore:
            try:
                async with session.get(url, timeout=aiohttp.ClientTimeout(total=5)) as resp:
                    if resp.status == 200:
                        data = await resp.json()
                        return {'symbol': symbol, 'price': data['price']}
                    elif resp.status == 429:
                        wait = 2 ** attempt
                        print(f'{symbol} 被限流，等待 {wait} 秒重试...')
                        await asyncio.sleep(wait)
                        continue
                    else:
                        return None
            except (asyncio.TimeoutError, aiohttp.ClientError):
                if attempt < retries:
                    await asyncio.sleep(1)
                else:
                    return None
    return None

把这些都串起来，完整的主流程就很清晰了：

async def main():
    symbols = load_symbols_from_file('stocks.txt')
    semaphore = asyncio.Semaphore(20)

    async with aiohttp.ClientSession() as session:
        async with asyncio.TaskGroup() as tg:
            tasks = [
                tg.create_task(fetch_price(session, sym, semaphore))
                for sym in symbols
            ]

    prices = [t.result() for t in tasks if t.result() is not None]
    save_to_database(prices)
    print(f'采集完成，成功 {len(prices)} / {len(symbols)} 条')

任务中途取消与超时控制

另一个常见需求是给整个采集过程设一个总超时时间。如果十分钟还没拉完所有数据，就放弃剩余任务，把已经拉到的数据保存下来。这可以通过在外层加一个asyncio.wait_for实现：

async def main_with_timeout():
    semaphore = asyncio.Semaphore(20)
    async with aiohttp.ClientSession() as session:
        try:
            async with asyncio.TaskGroup() as tg:
                tasks = [tg.create_task(fetch_price(session, sym, semaphore))
                         for sym in symbols]
                # 给整个TaskGroup加一个10分钟超时
                await asyncio.wait_for(asyncio.gather(*tasks, return_exceptions=True), timeout=600)
        except asyncio.TimeoutError:
            print('采集超时，取消剩余任务')

    # 收集已完成的结果
    results = [t.result() for t in tasks if t.done() and not t.cancelled() and t.result()]
    save_to_database(results)

这里有一点绕：TaskGroup本身没有内建超时机制，所以我们在外层用asyncio.wait_for包裹了整个任务组。当超时发生时，asyncio.wait_for会取消当前协程，导致async with asyncio.TaskGroup()块退出，进而取消组内所有尚未完成的任务。效果和预期一样——超时后剩余任务全部停掉，已经完成的保留数据。这种组合拳在离线数据采集场景里很实用。

与传统线程池的性能简单对比

为了直观感受差别，我用同一个任务（采集500个股票的行情）分别用concurrent.futures.ThreadPoolExecutor（20个线程）和上面的TaskGroup方案（20个并发）跑了一遍，结果如下：

线程池方式:  耗时 187秒，峰值内存 230MB
TaskGroup方式: 耗时 23秒，峰值内存 48MB

差距主要来自两个层面：线程的上下文切换和内存开销远高于协程，而且线程池在等待网络响应时线程是阻塞住的，并不能像协程那样释放执行权。20个线程真正并发工作的时刻并不多，大量时间都在傻等。协程模式下，20个并发用的是一个线程，底层事件循环不断切换，CPU几乎一直处于有效利用状态。

TaskGroup的几个限制

TaskGroup虽然好用，但有一些边界值得注意。第一，它要求Python 3.11及以上，如果你的生产环境还在3.9或者3.10，需要先升级解释器。第二，一个TaskGroup内部只能创建任务，不能直接await任务——你必须把任务放进组里，然后让组在退出时自动等待。第三，TaskGroup不能跨异步上下文传递，它绑定在当前的async with块生命周期内，一旦离开这个块，任务就结束了。如果你需要更灵活的任务生命周期管理（比如动态添加和取消），可能还是得回退到手动管理asyncio.Task对象的方式。

另外，TaskGroup的异常传播比较激进。组内任何一个任务以未捕获的异常结束，整个组都会被取消。这是结构化并发的设计哲学，但对于一些“允许部分失败”的场景，就需要在任务内部把异常吃掉。上面已经展示了这种做法。

总结

从asyncio.gather到asyncio.TaskGroup，本质上是一个从“自由并发”到“结构化并发”的转变。TaskGroup强制你划定任务的生命周期边界，让一组协程要么全部完成，要么全部取消，这正好对应了采集任务这类场景的核心需求——你需要的不是一部分数据先跑回来，而是一个确定的整体结果。

股票行情采集这个例子几乎覆盖了TaskGroup的完整用法：创建任务、控制并发、处理部分失败、超时取消、结果收集。把这套模板稍微调整一下，就可以复用到任何一个批量API调用的场景里。如果你的Python版本已经跟上，不妨把gather换下来体验一下，这种不用再为任务取消和异常处理头疼的感觉，写过的都懂。