Python脚本如何实现协程批量任务执行

本文目录导读：

1. 基础方案：asyncio.gather
2. 控制并发数：Semaphore信号量
3. 使用asyncio.TaskGroup (Python 3.11+)
4. 生产级方案：完整的任务队列系统
5. 极简方案：使用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())

1. asyncio.gather: 最简单的方式，同时执行所有任务
2. 信号量控制: Semaphore限制并发数
3. TaskGroup (Python 3.11+): 结构化管理异常
4. 任务队列: 适用于动态任务分配
5. as_completed: 按完成顺序处理结果
6. 重试机制: 提高任务执行成功率

选择哪种方案取决于：

• 任务数量
• 是否有并发限制
• 是否需要重试
• 是否需要进度回调
• Python版本要求