Python aiohttp场景下如何实现多ID聚合批量API请求 兼容原函数逻辑

核心实现思路是搭建全局的请求缓冲层，所有单ID的请求先在缓冲区临时挂起，靠两个触发条件统一打包发批量请求，请求完成后再把结果分发回每个挂起的调用协程，全程完全不改动原有函数的入参、返回值规则，对外行为和逐ID请求的旧逻辑完全一致。

实现的核心逻辑组件 #

不需要引入复杂的第三方依赖，用Python原生异步组件就能落地：

• 全局缓冲区：存储每一个进入的单ID请求条目，每个条目绑定一个asyncio.Future对象——后续批量请求拿到结果后，直接给对应的Future赋值，之前挂起的get_metadata协程就会自然解阻塞，拿到对应结果
• 异步互斥锁：保证缓冲区的并发读写不会出现数据竞争，避免多个协程同时操作缓冲区导致的ID遗漏、重复入队问题
• 事件触发器：用来在“缓冲区攒满50个ID”的场景下立刻唤醒批量请求流程，不用死等超时时间
• 后台批量处理协程：专门负责响应两个触发条件（攒满50个ID、单批等待满10秒），打包发HTTP请求、分发结果、处理异常

容易踩的实现坑 #

之前用信号量、队列实现容易失败，本质是没兼顾两个触发条件的响应逻辑：

• 纯队列的实现只能响应“新请求入队”的触发，没法在“到10秒阈值”的时候主动把缓冲区里攒的零散ID打包发请求，很容易出现死等、或者攒够了数量还要硬等10秒的低效问题
• 批量请求发送时如果不提前把缓冲区的内容清空，很容易出现同一个ID被重复打包到多批请求里的问题
• 必须兼容批量请求的异常场景：如果批量请求失败，要把当前批次所有挂起的get_metadata协程都设置成抛出对应异常，不能让所有调用永久阻塞，和旧逻辑里单请求抛错的行为保持一致

可直接复用的代码实现 #

import asyncio

# 全局缓冲配置
_batch_buffer: list[tuple[int, asyncio.Future]] = []
_buffer_lock = asyncio.Lock()
_trigger_event = asyncio.Event()
_batch_start_time = None
_batch_worker_started = False
MAX_BATCH_SIZE = 50
BATCH_WAIT_TIMEOUT = 10
# 复用你原有的aiohttp session对象
# session = ... 

async def _batch_worker():
    """后台常驻协程，负责批量打包发请求"""
    global _batch_start_time
    loop = asyncio.get_running_loop()
    while True:
        _trigger_event.clear()
        # 计算当前批次距离超时的剩余时间
        wait_remain = None
        if _batch_start_time is not None:
            wait_remain = BATCH_WAIT_TIMEOUT - (loop.time() - _batch_start_time)
            # 剩余时间小于等于0，直接触发请求
            if wait_remain <= 0:
                _trigger_event.set()
        
        # 等待触发事件或者超时
        try:
            await asyncio.wait_for(_trigger_event.wait(), timeout=wait_remain)
        except asyncio.TimeoutError:
            pass # 超时触发批量请求
        
        # 加锁取出当前缓冲区的所有待查询请求
        async with _buffer_lock:
            if not _batch_buffer:
                # 缓冲区空就重置计时，开始等下一批
                _batch_start_time = None
                continue
            current_batch = _batch_buffer.copy()
            _batch_buffer.clear()
            _batch_start_time = None # 处理完当前批次重置计时
        
        # 拼批量请求的URL，和原接口格式完全匹配
        id_query = "&".join([f"id={oid}" for oid, _ in current_batch])
        try:
            resp = await session.get(f"http://example.com/api/metadata?{id_query}")
            resp.raise_for_status()
            resp_results = (await resp.json())["results"]
            # 把结果分发到每个等待的Future
            for oid, fut in current_batch:
                if not fut.done():
                    fut.set_result(resp_results[oid])
        except Exception as e:
            # 批量请求失败时，所有同批次的调用抛出一致的异常，和旧逻辑行为对齐
            for _, fut in current_batch:
                if not fut.done():
                    fut.set_exception(e)


async def get_metadata(object_id):
    """完全保持原有入参、返回值签名的对外接口"""
    global _batch_start_time
    loop = asyncio.get_running_loop()
    # 为当前调用创建一个Future用来接收批量结果
    result_fut = loop.create_future()

    # 首次调用时启动后台批量worker
    if not _batch_worker_started:
        asyncio.create_task(_batch_worker())
        _batch_worker_started = True

    async with _buffer_lock:
        # 把当前请求加入缓冲区
        _batch_buffer.append( (object_id, result_fut) )
        # 如果是当前批次第一个请求，记录批次起始时间
        if _batch_start_time is None:
            _batch_start_time = loop.time()
        # 攒够最大批次大小就立刻触发批量请求
        if len(_batch_buffer) >= MAX_BATCH_SIZE:
            _trigger_event.set()
        elif not _trigger_event.is_set():
            # 没攒够数量就轻触发一次事件，保证worker重新计算等待超时时间
            _trigger_event.set()

    # 挂起等待批量请求返回结果
    return await result_fut

实现后的行为对齐说明 #

改造后所有外部调用get_metadata的逻辑完全不需要改动：

• 传入任意单个object_id，拿到的返回值和原来逐ID请求的结果完全一致
• 请求触发的HTTP错误会和旧逻辑一样向上抛出，不会出现静默失败
• 单批请求最多攒50个ID就发，不足50个的话最多等10秒就发，完全匹配接口的限流规则

内容的提问来源于stack exchange，提问作者Mitar