本文目录导读：
- 基础方案：asyncio.gather
- 控制并发数：Semaphore信号量
- 使用asyncio.TaskGroup (Python 3.11+)
- 生产级方案：完整的任务队列系统
- 极简方案：使用asyncio.Queue
我来详细介绍Python中实现协程批量任务执行的几种常用方法。
基础方案：asyncio.gather
import asyncio
import time
from typing import List, Any
async def task(name: str, delay: float) -> str:
"""模拟异步任务"""
print(f"开始执行任务: {name}")
await asyncio.sleep(delay)
print(f"任务完成: {name}")
return f"{name} 的结果"
async def batch_execute_gather():
"""使用gather批量执行"""
tasks = []
for i in range(5):
tasks.append(task(f"任务{i}", i * 0.5))
# 同时执行所有任务
results = await asyncio.gather(*tasks)
print(f"所有任务完成: {results}")
return results
# 运行
asyncio.run(batch_execute_gather())
控制并发数：Semaphore信号量
import asyncio
import aiohttp
from typing import List
class BatchExecutor:
def __init__(self, max_concurrency: int = 3):
self.semaphore = asyncio.Semaphore(max_concurrency)
async def limited_task(self, task_id: int, url: str) -> dict:
"""带并发限制的任务执行"""
async with self.semaphore:
print(f"开始任务 {task_id}，当前并发: {self.semaphore._value}")
try:
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
return await response.json()
except Exception as e:
print(f"任务 {task_id} 失败: {e}")
return None
finally:
print(f"完成任务 {task_id}")
async def execute_batch(self, tasks: List[tuple]) -> List[Any]:
"""批量执行带并发控制的任务"""
coroutines = [
self.limited_task(task_id, url)
for task_id, url in tasks
]
return await asyncio.gather(*coroutines)
# 并行执行带限制的HTTP请求
async def main():
executor = BatchExecutor(max_concurrency=2)
tasks = [
(1, "http://example.com/api/1"),
(2, "http://example.com/api/2"),
(3, "http://example.com/api/3"),
(4, "http://example.com/api/4"),
]
results = await executor.execute_batch(tasks)
print(f"最终结果: {results}")
# asyncio.run(main())
使用asyncio.TaskGroup (Python 3.11+)
import asyncio
from typing import List
class AdvancedBatchExecutor:
async def process_item(self, item: dict) -> dict:
"""处理单个项目"""
await asyncio.sleep(0.1) # 模拟IO操作
return {"processed": item["id"], "result": item["value"] * 2}
async def batch_process(self, items: List[dict]) -> List[dict]:
"""使用TaskGroup批量处理"""
results = []
# Python 3.11+ 的 TaskGroup
async with asyncio.TaskGroup() as tg:
# 创建任务并保存引用
tasks = [tg.create_task(self.process_item(item)) for item in items]
# 等待所有任务完成（自动管理异常）
# TaskGroup会自动等待所有任务
# 收集结果
results = [task.result() for task in tasks]
return results
# 使用示例
async def main():
executor = AdvancedBatchExecutor()
items = [
{"id": i, "value": i * 10}
for i in range(10)
]
results = await executor.batch_process(items)
print(f"处理结果: {results}")
# asyncio.run(main())
生产级方案：完整的任务队列系统
import asyncio
import logging
from typing import Callable, Any, List
from dataclasses import dataclass
from datetime import datetime
@dataclass
class TaskResult:
task_id: str
status: str # 'success', 'failed'
result: Any = None
error: str = None
class BatchTaskManager:
def __init__(self,
max_concurrency: int = 5,
retry_times: int = 3):
self.semaphore = asyncio.Semaphore(max_concurrency)
self.retry_times = retry_times
self.logger = logging.getLogger(__name__)
async def execute_with_retry(self,
task_func: Callable,
task_id: str,
*args,
**kwargs) -> TaskResult:
"""带重试机制的任务执行"""
for attempt in range(self.retry_times):
try:
async with self.semaphore:
self.logger.info(f"执行任务 {task_id}, 第{attempt+1}次尝试")
result = await task_func(*args, **kwargs)
return TaskResult(
task_id=task_id,
status='success',
result=result
)
except Exception as e:
self.logger.error(f"任务 {task_id} 第{attempt+1}次失败: {e}")
if attempt == self.retry_times - 1:
return TaskResult(
task_id=task_id,
status='failed',
error=str(e)
)
await asyncio.sleep(1 * (attempt + 1)) # 退避等待
async def process_batch(self,
tasks: List[tuple],
progress_callback: Callable = None) -> List[TaskResult]:
"""批量处理任务"""
tasks_list = []
total = len(tasks)
for i, (task_id, task_func, args, kwargs) in enumerate(tasks):
task = self.execute_with_retry(task_func, task_id, *args, **kwargs)
tasks_list.append(task)
# 使用as_completed获取任务进度
results = []
completed = 0
for coro in asyncio.as_completed(tasks_list):
result = await coro
results.append(result)
completed += 1
if progress_callback:
progress_callback(completed, total, result)
return results
# 使用示例
async def example_task(task_id: str, data: dict) -> dict:
"""示例任务函数"""
await asyncio.sleep(0.5)
if data.get("fail"):
raise ValueError("模拟失败")
return {"id": task_id, "result": data["value"]}
def progress_handler(completed, total, result):
"""进度回调"""
print(f"进度: {completed}/{total}, 任务 {result.task_id}: {result.status}")
async def main():
# 配置日志
logging.basicConfig(level=logging.INFO)
manager = BatchTaskManager(max_concurrency=3)
# 准备任务列表
tasks = []
for i in range(10):
task_id = f"task_{i}"
task_func = example_task
args = (task_id, {"value": i * 10, "fail": i == 5}) # 第6个任务会失败
kwargs = {}
tasks.append((task_id, task_func, args, kwargs))
# 执行批量任务
results = await manager.process_batch(tasks, progress_handler)
# 统计结果
success = sum(1 for r in results if r.status == 'success')
failed = sum(1 for r in results if r.status == 'failed')
print(f"完成: {len(results)} 个任务, 成功: {success}, 失败: {failed}")
# asyncio.run(main())
极简方案：使用asyncio.Queue
import asyncio
import random
class SimpleTaskQueue:
async def worker(self, worker_id: int, queue: asyncio.Queue):
"""工作协程"""
while True:
task = await queue.get()
if task is None: # 结束信号
break
task_id, func, args = task
print(f"Worker {worker_id} 处理任务 {task_id}")
try:
result = await func(*args)
print(f"Worker {worker_id} 完成任务 {task_id}: {result}")
except Exception as e:
print(f"Worker {worker_id} 任务 {task_id} 失败: {e}")
finally:
queue.task_done()
async def run(self, tasks: list, num_workers: int = 3):
"""运行任务队列"""
queue = asyncio.Queue()
# 添加任务
for task in tasks:
await queue.put(task)
# 添加结束信号
for _ in range(num_workers):
await queue.put(None)
# 创建工作协程
workers = [
asyncio.create_task(self.worker(i, queue))
for i in range(num_workers)
]
# 等待所有任务完成
await queue.join()
# 取消工作协程
for w in workers:
w.cancel()
print("所有任务完成")
# 使用示例
async def simple_task(data: str) -> str:
await asyncio.sleep(random.uniform(0.1, 0.5))
return f"处理结果: {data}"
async def main():
queue_runner = SimpleTaskQueue()
tasks = [
(i, simple_task, (f"数据{i}",))
for i in range(10)
]
await queue_runner.run(tasks, num_workers=3)
# asyncio.run(main())
- asyncio.gather: 最简单的方式，同时执行所有任务
- 信号量控制: Semaphore限制并发数
- TaskGroup (Python 3.11+): 结构化管理异常
- 任务队列: 适用于动态任务分配
- as_completed: 按完成顺序处理结果
- 重试机制: 提高任务执行成功率
选择哪种方案取决于：
- 任务数量
- 是否有并发限制
- 是否需要重试
- 是否需要进度回调
- Python版本要求
