如何在单个Python库中结合并行化与并发加速代码？

直接用Python标准库的concurrent.futures.ThreadPoolExecutor就能搞定，不需要混合多个库。核心思路是用线程池并行发起下载任务，同时监听已完成的任务，一旦有下载结果就立刻启动转换，不用等全部下载完成。

完整代码示例 #

import concurrent.futures
import time
import random

# 模拟下载函数，随机延迟模拟不同下载耗时
def downloadedData(input_val):
    delay = random.uniform(0.5, 2.0)
    time.sleep(delay)
    print(f"下载完成: {input_val} (耗时{delay:.2f}s)")
    return f"数据_{input_val}"

# 模拟转换函数，用上一次转换结果迭代更新
def transformer(data, prev_result):
    print(f"开始转换: {data}")
    time.sleep(0.3)  # 模拟转换耗时
    if prev_result is None:
        new_result = data
    else:
        new_result = f"{prev_result} -> {data}"
    print(f"转换完成，当前结果: {new_result}")
    return new_result

def main(inputs):
    # 初始化转换结果
    final_result = None

    # 创建线程池，并行发起下载任务
    with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
        # 提交所有下载任务，得到future对象列表
        future_to_input = {executor.submit(downloadedData, inp): inp for inp in inputs}

        # 按任务完成顺序处理结果
        for future in concurrent.futures.as_completed(future_to_input):
            downloaded_data = future.result()
            # 拿到下载结果立刻执行转换
            final_result = transformer(downloaded_data, final_result)

    print(f"\n最终转换结果: {final_result}")

if __name__ == "__main__":
    inputs = ["a", "b", "c"]
    main(inputs)

关键说明 #

• 并发下载：ThreadPoolExecutor负责并行处理所有下载任务，因为下载是IO密集型操作，线程池比进程池更高效，且无需额外处理数据共享问题。
• 即时转换：concurrent.futures.as_completed会按任务实际完成的顺序返回结果，不管提交顺序，所以只要有一个下载完成，就能立刻启动转换，完全符合你“不用等其余数据下载”的需求。
• 串行转换：转换逻辑在主线程中串行执行，每次转换都会用上一次的结果，满足转换依赖要求。
• 单库实现：整个方案只用到Python标准库的concurrent.futures，不需要结合其他库。

运行代码后你会看到，下载完成一个就立刻开始转换，不用等全部下载结束，完美匹配你的需求。

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