一、引言：为什么TaskGroup是异步编程的新标配
在构建现代Web服务时，高并发处理能力是衡量系统性能的关键指标。Python的asyncio生态在过去几年中蓬勃发展，但管理大量并发任务时的异常处理和资源清理一直是个棘手问题。传统上我们使用asyncio.gather()来并发执行多个协程，但当一个任务抛出异常时，其他任务可能继续运行而未被取消，造成资源浪费和状态不一致。
Python 3.11正式引入并在3.12中持续完善的asyncio.TaskGroup从根本上改变了这一局面。它借鉴了Trio库的”结构化并发”理念，确保任务组内任何一个任务失败时，所有其他任务都会被自动取消并等待完成，从而保证资源安全释放。本文将以构建一个高性能异步Web服务为完整案例，展示如何在实际项目中充分运用TaskGroup的能力。
二、环境准备与技术栈
本项目基于Python 3.12构建，使用FastAPI作为Web框架，httpx进行异步HTTP调用，asyncpg驱动异步PostgreSQL操作。完整依赖如下：
pip install fastapi uvicorn httpx asyncpg项目采用分层架构：
- 路由层（Router）：处理HTTP请求，调用业务逻辑。
- 服务层（Service）：使用TaskGroup编排多个异步操作。
- 仓库层（Repository）：封装数据库查询，返回异步结果。
- 客户端层（Client）：封装对外部API的异步请求。
三、TaskGroup基础：从简单示例理解核心机制
在深入Web服务之前，我们先通过一个简单示例理解TaskGroup的基本行为。TaskGroup是一个异步上下文管理器，在async with块内创建的所有任务都属于同一个组，具有自动取消和异常传播的特性。
import asyncio
async def fetch_user(user_id, delay):
"""模拟获取用户信息的异步操作"""
await asyncio.sleep(delay)
if user_id == 0:
raise ValueError(f"无效的用户ID: {user_id}")
return {"id": user_id, "name": f"用户{user_id}"}
async def main():
results = {}
try:
async with asyncio.TaskGroup() as tg:
# 创建多个并发任务
task1 = tg.create_task(fetch_user(1, 0.5))
task2 = tg.create_task(fetch_user(2, 0.8))
task3 = tg.create_task(fetch_user(0, 0.3)) # 这个会失败
# 逐个收集结果（需在TaskGroup内部完成）
results[1] = await task1
results[2] = await task2
results[0] = await task3 # 这里会抛出异常
except Exception as e:
print(f"任务组异常: {e}")
print(f"成功收集的结果: {results}")
# 输出: 成功收集的结果: {1: {...}, 2: {...}}
# 注意：task1和task2的结果仍然可用，task3被自动取消
asyncio.run(main())关键观察：当task3抛出异常时，TaskGroup会立即取消所有其他正在运行的任务（task1和task2如果在等待中也会被取消），然后重新抛出该异常。但那些已经完成的任务（task1和task2因为延迟较短，在task3失败前已完成）其返回值仍然可以通过await获取。这个特性在实际应用中非常重要——我们可以安全地收集已完成的结果，同时确保未完成的任务被清理。
四、构建异步数据访问层
首先搭建数据库操作的基础。使用asyncpg实现连接池管理和并发查询，为后续TaskGroup的并发编排提供底层支持。
import asyncpg
from typing import Optional, List
class DatabasePool:
"""异步数据库连接池管理器"""
def __init__(self, dsn: str):
self.dsn = dsn
self.pool: Optional[asyncpg.Pool] = None
async def connect(self):
self.pool = await asyncpg.create_pool(
self.dsn,
min_size=5,
max_size=20,
command_timeout=30
)
async def close(self):
if self.pool:
await self.pool.close()
async def get_user_orders(self, user_id: int) -> List[dict]:
"""获取用户订单列表"""
async with self.pool.acquire() as conn:
rows = await conn.fetch(
"SELECT id, product, amount, created_at FROM orders WHERE user_id = $1",
user_id
)
return [dict(row) for row in rows]
async def get_user_profile(self, user_id: int) -> Optional[dict]:
"""获取用户个人资料"""
async with self.pool.acquire() as conn:
row = await conn.fetchrow(
"SELECT id, username, email, avatar FROM users WHERE id = $1",
user_id
)
return dict(row) if row else None
async def get_user_reviews(self, user_id: int) -> List[dict]:
"""获取用户评价列表"""
async with self.pool.acquire() as conn:
rows = await conn.fetch(
"SELECT id, product_id, rating, content FROM reviews WHERE user_id = $1",
user_id
)
return [dict(row) for row in rows]五、服务层：TaskGroup编排并发业务逻辑
现在构建核心服务层，使用TaskGroup同时从数据库获取用户资料、订单和评价，并在聚合过程中处理部分失败的情况。
from asyncio import TaskGroup
from typing import Dict, Any
class UserService:
def __init__(self, db: DatabasePool, http_client):
self.db = db
self.http = http_client
async def get_user_dashboard(self, user_id: int) -> Dict[str, Any]:
"""使用TaskGroup并发获取用户仪表盘所需的所有数据"""
result = {
"profile": None,
"orders": [],
"reviews": [],
"recommendations": [],
"errors": []
}
# 使用信号量限制数据库并发连接
db_semaphore = asyncio.Semaphore(10)
async with TaskGroup() as tg:
# 定义任务工厂，每个任务独立处理异常
async def safe_fetch(name, coro):
try:
async with db_semaphore:
result[name] = await coro
except asyncpg.PostgresError as e:
result["errors"].append({"source": name, "error": str(e)})
except Exception as e:
result["errors"].append({"source": name, "error": str(e)})
# 创建三个数据库查询任务
tg.create_task(
safe_fetch("profile", self.db.get_user_profile(user_id))
)
tg.create_task(
safe_fetch("orders", self.db.get_user_orders(user_id))
)
tg.create_task(
safe_fetch("reviews", self.db.get_user_reviews(user_id))
)
# 同时调用外部推荐服务（不占用数据库信号量）
tg.create_task(
self._fetch_recommendations(user_id, result)
)
# TaskGroup退出后，所有任务已完成或取消
return result
async def _fetch_recommendations(self, user_id: int, result: dict):
"""获取外部推荐数据，内部处理异常"""
try:
response = await self.http.get(
f"https://api.recommend.com/users/{user_id}/recommendations",
timeout=5.0
)
if response.status_code == 200:
result["recommendations"] = response.json()["items"]
except Exception as e:
result["errors"].append({"source": "recommendations", "error": str(e)})这个服务层展示了TaskGroup在实际业务中的最佳实践：每个任务内部使用try/except捕获预期异常，通过共享的result字典安全传递数据。即使某个数据库查询失败，其他查询仍能正常完成并被收集。信号量的使用确保不会对数据库连接池造成过大压力。
六、路由层与FastAPI集成
将服务层与FastAPI路由绑定，处理HTTP请求并返回聚合结果。
from fastapi import FastAPI, HTTPException
from contextlib import asynccontextmanager
@asynccontextmanager
async def lifespan(app: FastAPI):
# 应用启动时初始化数据库连接池
app.state.db = DatabasePool("postgresql://user:pass@localhost/mydb")
await app.state.db.connect()
app.state.http_client = httpx.AsyncClient()
yield
# 应用关闭时清理资源
await app.state.db.close()
await app.state.http_client.aclose()
app = FastAPI(lifespan=lifespan)
@app.get("/api/users/{user_id}/dashboard")
async def user_dashboard(user_id: int):
service = UserService(
db=app.state.db,
http_client=app.state.http_client
)
try:
data = await service.get_user_dashboard(user_id)
return {"status": "success", "data": data}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))启动服务：
uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4访问http://localhost:8000/api/users/123/dashboard即可获得用户仪表盘的完整数据。由于使用了TaskGroup并发调用，整个请求的响应时间约等于最慢的那个子任务的时间，而非所有任务时间之和。
七、高级场景：批量处理与限流控制
TaskGroup在处理批量操作时同样表现出色。以下示例展示如何并发处理多个用户的数据汇总，同时实施严格的并发限流。
class BatchProcessor:
def __init__(self, db: DatabasePool, max_concurrency: int = 20):
self.db = db
self.semaphore = asyncio.Semaphore(max_concurrency)
async def process_user_batch(self, user_ids: list[int]) -> list[dict]:
"""批量处理用户数据，每批最多max_concurrency个并发"""
results = []
errors = []
async def process_one(uid: int):
async with self.semaphore:
try:
# 模拟单个用户的数据处理
profile = await self.db.get_user_profile(uid)
orders = await self.db.get_user_orders(uid)
return {
"user_id": uid,
"profile": profile,
"order_count": len(orders),
"total_amount": sum(o["amount"] for o in orders)
}
except Exception as e:
errors.append({"user_id": uid, "error": str(e)})
return None
async with TaskGroup() as tg:
tasks = {}
for uid in user_ids:
tasks[uid] = tg.create_task(process_one(uid))
for uid, task in tasks.items():
try:
data = await task
if data:
results.append(data)
except Exception as e:
errors.append({"user_id": uid, "error": str(e)})
print(f"批量处理完成: {len(results)} 成功, {len(errors)} 失败")
return results这里的Semaphore确保同时最多只有max_concurrency个数据库操作在执行，避免连接池耗尽。TaskGroup则保证在发生致命错误时所有未完成的任务被取消，防止资源泄漏。
八、错误处理策略对比
了解TaskGroup与asyncio.gather的错误处理差异对于正确选用工具至关重要：
async def compare_error_handling():
async def risky_task(name, fail=False):
await asyncio.sleep(0.1)
if fail:
raise RuntimeError(f"{name} 失败")
return f"{name} 成功"
# 方式一：使用gather（return_exceptions=True）
results = await asyncio.gather(
risky_task("A"),
risky_task("B", fail=True),
risky_task("C"),
return_exceptions=True
)
for i, r in enumerate(results):
print(f"gather结果 {i}: {type(r).__name__} - {r}")
# 输出: gather结果 0: str - A 成功
# gather结果 1: RuntimeError - B 失败
# gather结果 2: str - C 成功
# 方式二：使用TaskGroup（内部捕获异常）
collected = []
async with asyncio.TaskGroup() as tg:
async def safe_task(name, fail):
try:
collected.append(await risky_task(name, fail))
except Exception as e:
collected.append(f"{name} 错误: {e}")
tg.create_task(safe_task("A", False))
tg.create_task(safe_task("B", True))
tg.create_task(safe_task("C", False))
print(f"TaskGroup结果: {collected}")
# 输出: TaskGroup结果: ['A 成功', 'B 错误: ...', 'C 成功']两者的主要区别在于：gather需要设置return_exceptions=True才能收集异常，且所有任务都会运行到完成；TaskGroup则会在未捕获异常出现时取消其他任务，但通过在任务内部捕获异常，可以实现细粒度的错误处理，同时保留已完成的结果。
九、性能基准测试
通过简单的基准测试对比串行执行、asyncio.gather和TaskGroup在并发数据库查询中的表现：
import time
import asyncio
async def benchmark():
# 模拟10个各需100ms的数据库查询
async def mock_query(i):
await asyncio.sleep(0.1)
return i * 2
# 串行执行
start = time.perf_counter()
serial_results = []
for i in range(10):
serial_results.append(await mock_query(i))
serial_time = time.perf_counter() - start
# gather并发执行
start = time.perf_counter()
gather_results = await asyncio.gather(
*[mock_query(i) for i in range(10)]
)
gather_time = time.perf_counter() - start
# TaskGroup并发执行
start = time.perf_counter()
tg_results = []
async with asyncio.TaskGroup() as tg:
tasks = [tg.create_task(mock_query(i)) for i in range(10)]
for t in tasks:
tg_results.append(await t)
tg_time = time.perf_counter() - start
print(f"串行耗时: {serial_time:.3f}s")
print(f"gather耗时: {gather_time:.3f}s")
print(f"TaskGroup耗时: {tg_time:.3f}s")
print(f"并发提升: {serial_time/tg_time:.1f}x")
asyncio.run(benchmark())
# 预期输出: 串行耗时: 1.000s
# gather耗时: 0.100s
# TaskGroup耗时: 0.100s
# 并发提升: 10.0x在纯IO密集型任务中，TaskGroup与gather的性能几乎一致，因为它们共享相同的底层事件循环调度。TaskGroup的优势体现在异常安全性和代码结构上，而非原始性能。但在实际场景中，由于TaskGroup能更早地取消失败任务，可以节省不必要的等待时间。
十、最佳实践与注意事项
- 任务内部处理异常：将业务级别的错误（如查询失败、API超时）在任务内部捕获并记录，只让不可恢复的系统级异常冒泡到TaskGroup层。
- 使用Semaphore控制并发上限：TaskGroup本身不限制并发数，务必配合信号量或连接池限制对外部资源的并发访问。
- 避免在TaskGroup内使用asyncio.timeout嵌套：超时控制应放在TaskGroup外部或任务内部，避免复杂的嵌套取消行为。
- Python版本要求：TaskGroup需要Python 3.11+，生产环境建议使用3.12以获得更稳定的实现。
- 与FastAPI原生兼容：TaskGroup可在任何asyncio事件循环中运行，与FastAPI、Starlette等框架完全兼容。
十一、总结
asyncio.TaskGroup代表了Python异步编程向”结构化并发”迈进的重要一步。通过本文的完整实战，我们从数据库连接池搭建到服务层编排，再到FastAPI路由集成，完整展示了如何利用TaskGroup构建一个高性能、健壮且易于维护的异步Web服务。其核心优势在于：自动化的任务生命周期管理、清晰的异常处理语义，以及与现有asyncio生态的无缝集成。随着Python 3.12的普及，TaskGroup必将成为异步Python开发者的标配工具。
