使用ThreadPoolExecutor调用外部API时控制并发不超限额的方案

分块运行逻辑的可行性说明 #

• 手动将关键词按100个一组拆分、逐批次执行的逻辑确实可以避免触发接口并发上限，但效率存在明显缺陷：这种模式要求前一批次所有请求全部完成后才会启动下一批任务，一旦批次内存在个别响应极慢的请求，其余已经完成任务的线程会处于空等状态，无法充分用满100的并发配额。举个例子：如果一批100个请求里99个1秒就返回，剩下1个需要10秒返回，那99个空闲线程会白白等9秒才会执行下一批任务，并发能力浪费严重。

最高效的实现方案 #

不需要手动做分块处理，直接利用Python线程池原生的调度能力即可实现最优的并发控制：初始化ThreadPoolExecutor时将max_workers参数设置为外部API允许的最大并发值100即可。
线程池自带任务队列调度机制：初始化后最多同时运行100个请求，任意一个请求完成释放线程后，会自动从待执行任务队列中取下一个关键词发起请求，全程将在途请求数稳定控制在100，既不会触发限流，也不会出现线程空等的资源浪费。

可直接参考的实现代码：

import concurrent.futures
import requests
import time

# 外部API约定的最大并发上限
MAX_CONCURRENT_LIMIT = 100
# 请求超时时间，根据接口实际响应情况调整
REQUEST_TIMEOUT = 10
# 失败重试次数
MAX_RETRY = 2

def getdata(keyword):
    payload = {"keyword": keyword}
    # 带重试的请求逻辑
    for retry_cnt in range(MAX_RETRY + 1):
        try:
            res = requests.get("http://externalapi.com", params=payload, timeout=REQUEST_TIMEOUT)
            res.raise_for_status()
            # 此处写响应处理逻辑，例如存入数据库
            return {"keyword": keyword, "data": res.json()}
        except Exception as e:
            if retry_cnt >= MAX_RETRY:
                return {"keyword": keyword, "error": str(e)}
            # 重试前短暂等待，避免加剧接口压力
            time.sleep(0.5)

if __name__ == "__main__":
    # 接收前端提交的任意长度关键词列表
    keywords = ['keyword1', 'keyword2', ...] 
    with concurrent.futures.ThreadPoolExecutor(max_workers=MAX_CONCURRENT_LIMIT) as executor:
        # 如需收集执行结果可直接遍历返回值，不需要的话可直接执行不接收返回
        task_results = list(executor.map(getdata, keywords))

落地注意事项 #

• 必须配置请求超时：不要使用requests默认的无超时配置，否则遇到网络波动、接口无响应的情况，卡死的线程会长期占用并发名额，逐步耗尽线程池资源，导致整体任务停滞。
• 重试逻辑不要过度激进：一般配置2-3次重试、每次间隔0.5-1秒即可，重试间隔不要过短，避免额外增加接口压力触发更严格的限流。
• 不需要额外修改线程池的队列配置，默认的无界任务队列配合固定max_workers的配置已经完全适配动态长度的任务场景，无论前端提交几百还是上千个关键词，都能稳定控制并发不超限。

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