如何用队列与线程重构代码实现API并发请求？

直接上重构后的代码，核心思路是用队列分发请求任务，多线程并发处理API调用，彻底解决同步等待的效率问题：

import queue
import threading
import requests
import pandas as pd
import yaml
# 导入你已有的DataExtractor模块
from your_module import DataExtractor

def worker(task_queue, result_queue, headers):
    """工作线程：从任务队列取URL，发送请求并将结果存入结果队列"""
    while True:
        url = task_queue.get()
        # 收到结束信号则退出循环
        if url is None:
            break
        try:
            response = requests.get(url=url, headers=headers)
            response.raise_for_status()  # 触发HTTP错误异常
            result_queue.put(response.json())
        except Exception as e:
            # 可根据需求扩展错误处理，比如写入日志或重试
            print(f"请求URL {url} 失败: {str(e)}")
        finally:
            # 标记当前任务处理完成，让队列感知进度
            task_queue.task_done()

def retrieve_stores_data(api_endpoint: str, headers=None):
    number_of_stores_endpoint = 'https://aqj7u5id95.execute-api.eu-west-1.amazonaws.com/prod/number_stores'
    
    # 加载请求头
    if headers is None:
        with open('API_Keys.yaml', 'r') as stream:
            headers = yaml.safe_load(stream)
    
    # 获取门店总数
    number_of_stores = DataExtractor.list_number_of_stores(number_of_stores_endpoint, headers)
    
    # 初始化任务队列（存待请求URL）和结果队列（存API返回数据）
    task_queue = queue.Queue()
    result_queue = queue.Queue()
    
    # 填充任务队列
    for store_id in range(number_of_stores):
        url = api_endpoint + f'{store_id}'
        task_queue.put(url)
    
    # 设置线程数：根据API速率限制调整，建议10-20之间
    thread_count = 15
    threads = []
    
    # 创建并启动工作线程
    for _ in range(thread_count):
        thread = threading.Thread(target=worker, args=(task_queue, result_queue, headers))
        thread.start()
        threads.append(thread)
    
    # 等待队列中所有任务处理完成
    task_queue.join()
    
    # 向队列发送结束信号，每个线程对应一个None
    for _ in range(thread_count):
        task_queue.put(None)
    
    # 等待所有线程退出
    for thread in threads:
        thread.join()
    
    # 从结果队列提取所有数据并转为DataFrame
    stores_data = []
    while not result_queue.empty():
        stores_data.append(result_queue.get())
    
    return pd.DataFrame(stores_data)

核心优化说明 #

• 任务队列替代列表：用queue.Queue自动处理任务分发，避免手动遍历和同步阻塞，天然支持线程安全的任务传递。
• 多线程并发请求：多个线程同时处理API调用，把原本串行等待响应的时间转化为并行处理，总耗时大幅降低。
• 线程安全的结果存储：result_queue是线程安全容器，无需额外加锁即可安全存储多线程返回的数据。
• 优雅的线程管理：通过task_done()和join()跟踪任务进度，发送None信号让线程有序退出，避免资源泄漏。

注意事项 #

• 线程数需根据目标API的速率限制调整，避免因请求过于频繁触发限流或封禁；如果API有明确QPS限制，可适当降低线程数或在worker中添加短暂延迟（time.sleep(0.1)）。
• 如果需要保证结果顺序与门店ID一致，可以在请求结果中附带store_id，后续再按ID排序DataFrame。

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