Python异步队列任务实现咨询：异步执行函数并批量处理结果

核心实现思路 #

使用asyncio异步队列（asyncio.Queue）作为结果容器，并发执行所有目标函数并将结果推入队列；同时启动一个消费者任务，定期从队列中批量取出最多512条结果进行处理，直到所有函数执行完成且队列清空。

完整代码示例 #

import asyncio

# ---------------------- 替换为你的实际函数 ----------------------
async def your_async_func(param):
    # 模拟耗时异步操作（如API请求、IO读写等）
    await asyncio.sleep(0.05)
    return f"Result of task {param}"

def process_batch(items):
    # 批量处理逻辑，替换为你的实际业务代码
    print(f"Processed {len(items)} tasks: {items[:3]}...")
# -------------------------------------------------------------

async def consumer(queue, total_task_count):
    processed = 0
    while processed < total_task_count:
        batch = []
        try:
            # 等待1秒尝试获取队列元素，避免空循环阻塞事件循环
            first_item = await asyncio.wait_for(queue.get(), timeout=1)
            batch.append(first_item)
            
            # 批量获取剩余元素，最多取到512条
            while not queue.empty() and len(batch) < 512:
                batch.append(queue.get_nowait())
        except asyncio.TimeoutError:
            # 超时说明队列暂时无数据，继续循环等待
            continue
        
        # 处理当前批次
        process_batch(batch)
        processed += len(batch)
        
        # 标记队列中这些元素已处理完成
        for _ in batch:
            queue.task_done()

async def main():
    # 生成你的函数列表（这里用带参数的异步函数示例）
    task_list = [your_async_func(i) for i in range(1500)]  # 1500个示例异步任务
    result_queue = asyncio.Queue(maxsize=2000)  # 队列大小可根据内存情况调整

    # 创建所有生产者任务：并发执行函数，将结果推入队列
    producer_tasks = []
    for coro in task_list:
        async def produce(coro):
            result = await coro
            await result_queue.put(result)
        producer_tasks.append(asyncio.create_task(produce(coro)))

    # 启动消费者任务
    consumer_task = asyncio.create_task(consumer(result_queue, len(task_list)))

    # 等待所有生产者任务执行完毕
    await asyncio.gather(*producer_tasks)
    # 等待队列中所有元素都被处理完成
    await result_queue.join()
    # 终止消费者任务（此时所有任务已处理完毕，循环会自动退出，也可省略取消步骤）
    consumer_task.cancel()
    try:
        await consumer_task
    except asyncio.CancelledError:
        pass

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

关键细节说明 #

1. 异步队列的必要性：asyncio.Queue是异步安全的，支持await put()和await get()操作，避免了线程安全问题，适配异步事件循环模型。
2. 并发执行函数：每个目标函数被包装成独立的异步任务，通过asyncio.create_task启动，实现真正的并发执行（而非伪代码中的顺序执行）。
3. 批量获取逻辑：先用asyncio.wait_for设置1秒超时等待队列元素，避免空循环阻塞；再用get_nowait批量获取剩余元素，最多取512条，符合你的需求。
4. 结束条件控制：
   • 等待所有生产者任务完成，确保没有新的结果推入队列；
   • 调用queue.join()等待队列中所有元素被处理完毕；
   • 最后取消消费者任务，完成整个流程。

常见误区规避 #

• 不要在异步代码中使用time.sleep()：它会阻塞整个事件循环，导致所有异步任务暂停，需用asyncio.sleep()或超时机制替代。
• 不要直接同步调用异步函数：必须用await或create_task启动，否则函数不会真正执行。

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