import asyncio
import aiohttp

# 通用異步任務(wù)：調(diào)用第三方接口獲取數(shù)據(jù)
async def fetch_data(api_path: str):
    async with aiohttp.ClientSession() as session:
        # 模擬調(diào)用各類第三方接口（如支付、短信、數(shù)據(jù)查詢）
        resp = await session.get(f"https://api.example.com{api_path}")
        return await resp.json()

async def batch_fetch(api_paths: list):
    # 直接創(chuàng)建所有任務(wù)并并發(fā)執(zhí)行
    tasks = [fetch_data(path) for path in api_paths]
    # 無并發(fā)限制：所有任務(wù)同時(shí)發(fā)起請(qǐng)求
    results = await asyncio.gather(*tasks)
    return results

# 執(zhí)行：一次性發(fā)起500個(gè)HTTP請(qǐng)求
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):
    # 信號(hào)量控制：同一時(shí)間僅10個(gè)任務(wù)執(zhí)行該邏輯
    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):
    # 初始化信號(hào)量，設(shè)置并發(fā)上限10
    sem = asyncio.Semaphore(10)
    # 任務(wù)創(chuàng)建時(shí)傳入信號(hào)量
    tasks = [fetch_data(path, sem) for path in api_paths]
    results = await asyncio.gather(*tasks)
    return results

# 執(zhí)行：始終保持10個(gè)請(qǐng)求并行，剩余任務(wù)排隊(duì)
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:
    """通用異步批量處理器，結(jié)合gather與Semaphore實(shí)現(xiàn)可控并發(fā)"""
    def __init__(self, concurrency_limit: int = 10):
        """
        初始化處理器
        :param concurrency_limit: 并發(fā)上限，默認(rèn)10（可根據(jù)業(yè)務(wù)調(diào)整）
        """
        self.concurrency_limit = concurrency_limit
        self.semaphore = asyncio.Semaphore(concurrency_limit)

    async def _process_single(self, item: Any) -> Dict[str, Any]:
        """
        處理單個(gè)任務(wù)（核心業(yè)務(wù)邏輯，可按需重寫）
        :param item: 單個(gè)任務(wù)參數(shù)（如接口路徑、文件路徑、數(shù)據(jù)庫ID等）
        :return: 任務(wù)執(zhí)行結(jié)果
        """
        async with self.semaphore:  # 信號(hào)量控制并發(fā)規(guī)模
            try:
                # 替換為實(shí)際業(yè)務(wù)邏輯（如接口調(diào)用、文件讀寫、數(shù)據(jù)計(jì)算）
                await asyncio.sleep(0.1)  # 模擬異步操作
                logger.info(f"任務(wù){(diào)item}執(zhí)行完成")
                return {"status": "success", "item": item, "result": "ok"}
            except Exception as e:
                logger.error(f"任務(wù){(diào)item}執(zhí)行失敗: {str(e)}")
                return {"status": "failed", "item": item, "error": str(e)}

    async def batch_process(self, items: List[Any]) -> List[Dict[str, Any]]:
        """
        批量處理任務(wù)
        :param items: 任務(wù)列表（通用化參數(shù)，適配任意類型）
        :return: 所有任務(wù)的執(zhí)行結(jié)果
        """
        # 創(chuàng)建所有異步任務(wù)
        tasks = [self._process_single(item) for item in items]
        # 并發(fā)執(zhí)行（由Semaphore控制并發(fā)上限）
        results = await asyncio.gather(*tasks, return_exceptions=False)
        return results

# 通用調(diào)用示例
async def main():
    # 初始化處理器，設(shè)置并發(fā)上限5
    processor = AsyncBatchProcessor(concurrency_limit=5)
    # 模擬任意類型的任務(wù)列表（接口路徑、ID、文件路徑等）
    task_list = [f"task_{i}" for i in range(20)]
    # 批量執(zhí)行任務(wù)
    results = await processor.batch_process(task_list)
    # 統(tǒng)計(jì)執(zhí)行結(jié)果
    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}個(gè)，失敗{failed_count}個(gè)")

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