基于aioboto3优化AWS CloudWatch日志异步批量下载方案

1. 修复重试机制失效问题 #

当前超过25协程时触发限流异常后无法重试，核心原因有两点：一是每个协程单独创建logs_client，导致连接资源浪费且重试逻辑未受并发控制；二是standard重试模式对限流场景的适配性不足。

优化方案：

• 全局复用一个logs_client，避免每个协程重复创建客户端，减少连接开销。
• 改用adaptive重试模式，botocore会根据限流情况动态调整重试间隔，比standard更适合应对AWS的ThrottlingException。
• 确保重试逻辑被包裹在信号量控制范围内，避免重试时突破并发限制，同时对限流异常单独处理，支持任务重入重试。

修改后的核心代码示例：

async def fetch_logs(semaphore, logs_client, start_time, end_time, log_group_name, retry_count=0):
    async with semaphore:
        try:
            start_query_response = await logs_client.start_query(
                logGroupName=log_group_name,
                startTime=int(start_time.timestamp()),
                endTime=int(end_time.timestamp()),
                queryString="some_query")

            query_id = start_query_response['queryId']
            response = None
            # 指数退避轮询，减少无效请求
            backoff = 1
            max_backoff = 16
            while response is None or response['status'] == 'Running':
                await asyncio.sleep(backoff)
                response = await logs_client.get_query_results(queryId=query_id)
                backoff = min(backoff * 2, max_backoff)

        except ClientError as e:
            if e.response['Error']['Code'] == 'ThrottlingException' and retry_count < 5:
                print(f"限流触发，重试任务: {log_group_name} {start_time} (第{retry_count+1}次)")
                await fetch_logs(semaphore, logs_client, start_time, end_time, log_group_name, retry_count+1)
            else:
                print(f"任务失败: {log_group_name} {start_time}, 错误: {e}")
                # 可将失败任务写入文件，后续补采
                with open("failed_tasks.txt", "a") as f:
                    f.write(f"{log_group_name},{start_time},{end_time}\n")

async def main():
    session = aioboto3.Session(profile_name="profile_name", region_name="us-east-1")
    # 全局创建一个日志客户端，复用连接池
    async with session.client(service_name='logs', region_name='us-east-1',
                              config=Config(
                                  {"retries": {"max_attempts": 6, "mode": "adaptive"},
                                   "max_pool_connections": 10})) as logs_client:
        log_groups = get_log_groups()
        if log_groups:
            semaphore = asyncio.Semaphore(25)
            all_tasks = []
            today = datetime.now()
            absolute_start_date = today - timedelta(days=730)
            
            # 生成所有查询任务
            for log_group in log_groups:
                current_end_date = today
                while current_end_date > absolute_start_date:
                    current_start_date = current_end_date - timedelta(days=1)
                    if current_start_date <= absolute_start_date:
                        break
                    all_tasks.append(
                        fetch_logs(semaphore, logs_client, current_start_date, current_end_date, log_group)
                    )
                    current_end_date = current_start_date
            
            # 并发执行所有任务
            await asyncio.gather(*all_tasks)

2. 理解max_pool_connections=10更优的原因 #

原代码中每个协程都创建独立的logs_client，每个客户端默认拥有10个连接的连接池，25个协程就会产生250个连接，远超实际需求，导致连接竞争、资源浪费，反而拖慢效率。

优化后全局复用一个logs_client，10个连接的池足够应对25个并发协程——因为异步请求的连接会被复用，无需为每个协程单独分配连接，所以资源利用率更高，速度更快。

3. 替换固定批量为异步队列，提升灵活性 #

固定每批25个任务的方式不够灵活，遇到任务失败时需要手动重新分批。改用asyncio.Queue实现生产者-消费者模式，能动态控制并发，自动处理任务分发与重试：

• 生产者：将所有日期-日志组的查询任务放入队列
• 消费者：启动25个协程从队列取任务执行，任务完成后自动处理下一个

示例代码片段：

async def worker(semaphore, logs_client, queue):
    while not queue.empty():
        task_info = await queue.get()
        try:
            await fetch_logs(semaphore, logs_client, *task_info)
        finally:
            queue.task_done()

async def main():
    # ... 省略session和client创建 ...
    task_queue = asyncio.Queue()
    today = datetime.now()
    absolute_start_date = today - timedelta(days=730)
    
    # 生产者：填充任务队列
    for log_group in log_groups:
        current_end_date = today
        while current_end_date > absolute_start_date:
            current_start_date = current_end_date - timedelta(days=1)
            if current_start_date <= absolute_start_date:
                break
            await task_queue.put((current_start_date, current_end_date, log_group))
            current_end_date = current_start_date
    
    # 启动25个消费者协程
    semaphore = asyncio.Semaphore(25)
    workers = [asyncio.create_task(worker(semaphore, logs_client, task_queue)) for _ in range(25)]
    await task_queue.join()
    
    # 终止所有工作协程
    for w in workers:
        w.cancel()
    await asyncio.gather(*workers, return_exceptions=True)

4. 多进程方案的适用性分析 #

AWS的限流是针对账号/区域的全局阈值（每秒30次请求），多进程无法绕过该限制，反而会增加进程间通信的开销。由于脚本属于IO密集型（核心是网络请求），asyncio的协程模型已经足够高效，GIL不会成为瓶颈。只有当日志数据的解析、处理是CPU密集型操作时，才适合用multiprocessing.Pool处理数据，但请求部分仍建议保留asyncio实现。

5. 其他细节优化 #

• 轮询策略：将固定1秒休眠改为指数退避（1s→2s→4s→8s，最大到16s），减少不必要的请求次数，降低触发限流的概率。
• 失败任务记录：对无法重试的错误，将任务信息写入文件，避免数据丢失，方便后续手动补采。
• 动态并发调整：可以通过监控请求成功率，自动在20-25之间调整信号量大小，找到当前网络环境下的最优并发数。

内容的提问来源于stack exchange，提问作者eran-avrahami