如何在Python中并行化多文件的链式API调用流程

刚好我之前也遇到过类似的需求，这种流水线式的并行处理在IO密集型的API调用场景里特别实用——既能最大化利用等待API响应的空闲时间，又能保证单个文件的API调用严格按顺序执行，哪个文件的前一步走完就立刻启动下一步，完全不用等其他文件的同阶段任务。

下面给你两种用Python原生库实现的方案，不用装额外依赖，按需选就行：

方法一：单文件完整任务链并行（简单易上手） #

这种方案最适合你的初始代码改造，逻辑和原来的串行代码几乎一致，只是把单个文件的完整处理流程包装成函数，然后提交到线程池里并行执行。

代码示例 #

import time
from concurrent.futures import ThreadPoolExecutor, as_completed

# ----------------------
# 先模拟你的真实API调用，实际用的时候替换成你自己的函数
# ----------------------
def call_api_1(file):
    print(f"启动 {file} 的API 1调用")
    time.sleep(2)  # 模拟API的IO等待时间
    return f"api1_result_{file}"

def call_api_2(file, api1_output):
    print(f"启动 {file} 的API 2调用，输入：{api1_output}")
    time.sleep(1.5)
    return f"api2_result_{file}_{api1_output}"

def call_api_3(file, api2_output):
    print(f"启动 {file} 的API 3调用，输入：{api2_output}")
    time.sleep(1)
    return f"final_result_{file}"

def save_final_output(result):
    print(f"保存最终结果：{result}")

# ----------------------
# 核心：单个文件的串行处理链
# ----------------------
def process_single_file(file):
    # 单个文件的API调用严格按顺序走
    out1 = call_api_1(file)
    out2 = call_api_2(file, out1)
    final_out = call_api_3(file, out2)
    save_final_output(final_out)
    return final_out

# ----------------------
# 主程序：并行处理所有文件
# ----------------------
if __name__ == "__main__":
    files = ["file_1.png", "file_2.png", "file_3.png"]
    
    # 线程池大小可以根据API的并发限制调整，比如API允许同时5次调用就设为5
    with ThreadPoolExecutor(max_workers=3) as executor:
        # 给每个文件提交一个处理任务
        task_futures = [executor.submit(process_single_file, file) for file in files]
        
        # 实时监控任务完成情况，也可以直接用executor.map简化
        for future in as_completed(task_futures):
            try:
                result = future.result()
                print(f"✅ 文件处理完成：{result}")
            except Exception as e:
                # 捕获单个文件的处理异常，不影响其他文件
                print(f"❌ 文件处理出错：{str(e)}")

为什么这能满足你的需求？ #

每个文件的process_single_file函数是串行执行API1→API2→API3，但因为我们把这些函数提交到线程池，多个文件的处理流程是并行推进的：

• 一开始会同时启动所有文件的API1调用
• 当某一个文件的API1调用完成，会立刻启动它的API2，完全不用等其他文件的API1跑完
• 以此类推，完全符合你要的“流水线式”并行效果

这种方案的优势是改动极小，和你原来的串行代码逻辑对齐，调试和维护都特别简单。

方法二：分阶段流水线并行（精细控并发） #

如果你的不同API有不同的并发限制（比如API1允许同时跑5次，API2只允许同时跑2次），可以用这种分阶段的方式，给每个API步骤单独配线程池，精准控制每个阶段的并发数。

代码示例 #

import time
from concurrent.futures import ThreadPoolExecutor, as_completed

# ----------------------
# 同样先模拟你的API调用
# ----------------------
def call_api_1(file):
    print(f"启动 {file} 的API 1调用")
    time.sleep(2)
    return f"api1_result_{file}"

def call_api_2(file, api1_output):
    print(f"启动 {file} 的API 2调用，输入：{api1_output}")
    time.sleep(1.5)
    return f"api2_result_{file}_{api1_output}"

def call_api_3(file, api2_output):
    print(f"启动 {file} 的API 3调用，输入：{api2_output}")
    time.sleep(1)
    return f"final_result_{file}"

def save_final_output(result):
    print(f"保存最终结果：{result}")

# ----------------------
# 每个阶段的处理函数
# ----------------------
def stage_api1(file):
    # 返回文件名+API结果，方便后续阶段关联到原文件
    return file, call_api_1(file)

def stage_api2(api1_result):
    file, api1_out = api1_result
    return file, call_api_2(file, api1_out)

def stage_api3(api2_result):
    file, api2_out = api2_result
    final_result = call_api_3(file, api2_out)
    save_final_output(final_result)
    return final_result

# ----------------------
# 主程序：分阶段提交任务
# ----------------------
if __name__ == "__main__":
    files = ["file_1.png", "file_2.png", "file_3.png"]
    
    # 给每个API阶段单独配线程池，按需设置并发数
    with ThreadPoolExecutor(max_workers=3) as executor_api1, \
         ThreadPoolExecutor(max_workers=2) as executor_api2, \
         ThreadPoolExecutor(max_workers=2) as executor_api3:
        
        # 第一阶段：提交所有文件的API1任务
        futures_api1 = [executor_api1.submit(stage_api1, file) for file in files]
        
        # 实时处理API1的完成结果，立刻提交到API2阶段
        futures_api2 = []
        for future in as_completed(futures_api1):
            try:
                api1_res = future.result()
                futures_api2.append(executor_api2.submit(stage_api2, api1_res))
            except Exception as e:
                print(f"❌ API1处理出错：{str(e)}")
        
        # 实时处理API2的完成结果，立刻提交到API3阶段
        futures_api3 = []
        for future in as_completed(futures_api2):
            try:
                api2_res = future.result()
                futures_api3.append(executor_api3.submit(stage_api3, api2_res))
            except Exception as e:
                print(f"❌ API2处理出错：{str(e)}")
        
        # 等待所有最终任务完成
        for future in as_completed(futures_api3):
            try:
                final_res = future.result()
                print(f"✅ 全流程完成：{final_res}")
            except Exception as e:
                print(f"❌ API3处理出错：{str(e)}")

这种方案的优势 #

• 可以给每个API阶段单独设置并发数，完美适配不同API的调用限制
• 完全是流水线式推进，哪个文件的前阶段完成就立刻进入下阶段，没有任何不必要的等待
• 每个阶段的逻辑独立，后续要加新的API步骤也很方便

一些实用的注意事项 #

• 线程/进程池选择：API调用基本都是IO密集型任务，用ThreadPoolExecutor就够了，开销远小于ProcessPoolExecutor；如果你的API调用里有大量本地CPU计算，再考虑用进程池。
• 异常处理一定要加：API调用很容易因为网络、限流等原因出错，一定要在任务函数或者处理future的时候捕获异常，避免单个文件的错误导致整个流程卡住。
• 并发数别设太满：一定要参考API服务商的并发限制，别把线程池设得太大，不然容易触发限流甚至被封IP。
• 如果是HTTP API，也可以用asyncio：如果你的API是HTTP接口，用asyncio配合aiohttp做异步请求，效率会更高，逻辑和上面的分阶段方案类似，只是用协程代替线程。

备注：内容来源于stack exchange，提问作者Omega