Python异步并发控制asyncio.gather 与 Semaphore 协同设计解析

import asyncio
import aiohttp

# 通用异步任务：调用第三方接口获取数据
async def fetch_data(api_path: str):
    async with aiohttp.ClientSession() as session:
        # 模拟调用各类第三方接口（如支付、短信、数据查询）
        resp = await session.get(f"https://api.example.com{api_path}")
        return await resp.json()

async def batch_fetch(api_paths: list):
    # 直接创建所有任务并并发执行
    tasks = [fetch_data(path) for path in api_paths]
    # 无并发限制：所有任务同时发起请求
    results = await asyncio.gather(*tasks)
    return results

# 执行：一次性发起500个HTTP请求
if __name__ == "__main__":
    api_list = [f"/data/{i}" for i in range(500)]
    asyncio.run(batch_fetch(api_list))

import asyncio
import aiohttp

async def fetch_data(api_path: str, sem: asyncio.Semaphore):
    # 信号量控制：同一时间仅10个任务执行该逻辑
    async with sem:
        async with aiohttp.ClientSession() as session:
            resp = await session.get(f"https://api.example.com{api_path}")
            return await resp.json()

async def batch_fetch(api_paths: list):
    # 初始化信号量，设置并发上限10
    sem = asyncio.Semaphore(10)
    # 任务创建时传入信号量
    tasks = [fetch_data(path, sem) for path in api_paths]
    results = await asyncio.gather(*tasks)
    return results

# 执行：始终保持10个请求并行，剩余任务排队
if __name__ == "__main__":
    api_list = [f"/data/{i}" for i in range(500)]
    asyncio.run(batch_fetch(api_list))

import asyncio
import logging
from typing import List, Any, Dict

# 初始化通用日志器
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class AsyncBatchProcessor:
    """通用异步批量处理器，结合gather与Semaphore实现可控并发"""
    def __init__(self, concurrency_limit: int = 10):
        """
        初始化处理器
        :param concurrency_limit: 并发上限，默认10（可根据业务调整）
        """
        self.concurrency_limit = concurrency_limit
        self.semaphore = asyncio.Semaphore(concurrency_limit)

    async def _process_single(self, item: Any) -> Dict[str, Any]:
        """
        处理单个任务（核心业务逻辑，可按需重写）
        :param item: 单个任务参数（如接口路径、文件路径、数据库ID等）
        :return: 任务执行结果
        """
        async with self.semaphore:  # 信号量控制并发规模
            try:
                # 替换为实际业务逻辑（如接口调用、文件读写、数据计算）
                await asyncio.sleep(0.1)  # 模拟异步操作
                logger.info(f"任务{item}执行完成")
                return {"status": "success", "item": item, "result": "ok"}
            except Exception as e:
                logger.error(f"任务{item}执行失败: {str(e)}")
                return {"status": "failed", "item": item, "error": str(e)}

    async def batch_process(self, items: List[Any]) -> List[Dict[str, Any]]:
        """
        批量处理任务
        :param items: 任务列表（通用化参数，适配任意类型）
        :return: 所有任务的执行结果
        """
        # 创建所有异步任务
        tasks = [self._process_single(item) for item in items]
        # 并发执行（由Semaphore控制并发上限）
        results = await asyncio.gather(*tasks, return_exceptions=False)
        return results

# 通用调用示例
async def main():
    # 初始化处理器，设置并发上限5
    processor = AsyncBatchProcessor(concurrency_limit=5)
    # 模拟任意类型的任务列表（接口路径、ID、文件路径等）
    task_list = [f"task_{i}" for i in range(20)]
    # 批量执行任务
    results = await processor.batch_process(task_list)
    # 统计执行结果
    success_count = len([r for r in results if r["status"] == "success"])
    failed_count = len([r for r in results if r["status"] == "failed"])
    logger.info(f"批量处理完成：成功{success_count}个，失败{failed_count}个")

if __name__ == "__main__":
    asyncio.run(main())