Python中requests.futures批量请求重试机制及结果合并实现问题

核心思路 #

不用单独维护原响应列表和重试列表，而是维护一个待请求的任务队列，每次执行后分离成功/失败任务，失败任务重新入队循环处理，直到所有任务成功或达到设定的最大重试次数。这样自然就能把所有成功响应合并到最终列表里。

代码实现示例 #

from requests_futures.sessions import FuturesSession
import requests

def validate_response(response):
    # 自定义响应验证逻辑：检查状态码200且内容不含"Server Error"
    if response.status_code != 200:
        return False
    if "Server Error" in response.text:
        return False
    return True

def send_single_request(session, url):
    # 封装单个请求的发送逻辑，可按需添加headers、参数等
    return session.get(url)

def send_bulk_requests_with_retry(urls, max_retries=3):
    session = FuturesSession()
    final_responses = []
    # 初始化待处理队列：存储(url, 已重试次数)元组
    pending_tasks = [(url, 0) for url in urls]

    while pending_tasks:
        current_batch = pending_tasks
        pending_tasks = []
        # 批量发送当前批次请求
        futures = [send_single_request(session, url) for url, _ in current_batch]
        # 遍历结果，分离成功与需重试任务
        for future, (url, retry_count) in zip(futures, current_batch):
            try:
                response = future.result()
                if validate_response(response):
                    final_responses.append(response)
                else:
                    if retry_count < max_retries:
                        pending_tasks.append((url, retry_count + 1))
                        print(f"URL {url} 验证失败，将进行第{retry_count+1}次重试")
                    else:
                        print(f"URL {url} 达到最大重试次数，放弃")
            except requests.exceptions.RequestException as e:
                # 处理请求异常（超时、连接错误等）
                if retry_count < max_retries:
                    pending_tasks.append((url, retry_count + 1))
                    print(f"URL {url} 请求出错: {str(e)}，将进行第{retry_count+1}次重试")
                else:
                    print(f"URL {url} 达到最大重试次数，放弃")
    return final_responses

# 调用示例
if __name__ == "__main__":
    target_urls = ["https://example.com/api/1", "https://example.com/api/2"]
    responses = send_bulk_requests_with_retry(target_urls, max_retries=3)
    # 处理最终成功响应
    for resp in responses:
        print(f"URL {resp.url} 处理成功")

关键细节说明 #

• 任务队列设计：用(url, 已重试次数)元组跟踪每个任务的重试状态，避免无限重试。
• 可复用验证逻辑：validate_response函数单独抽离，可根据实际需求修改（比如检查JSON字段、特定HTML标签等）。
• 异常覆盖：除了验证响应内容，还处理了请求过程中的超时、连接失败等异常，让重试逻辑更健壮。
• 自动合并响应：所有验证通过的响应直接追加到final_responses，无需手动合并多轮请求结果。

可选优化 #

• 添加指数退避：重试前加入延迟，避免频繁请求给服务器造成压力：

  import time
  # 在重试任务入队前添加延迟
  time.sleep(2 ** retry_count)
  

• 扩展任务参数：如果请求需要自定义headers、参数，可将这些信息加入任务元组，比如(url, headers, params, retry_count)。

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