如何并行调用耗时函数并维持固定3个并行执行实例？

要始终维持3个耗时函数实例并行执行，无论每个函数的执行时间（0-300秒）如何变化，Python里最直接的方式是用线程池或进程池来管理任务，自动维持固定数量的并行实例。以下是两种常用实现：

一、线程池方案（IO密集型场景优先） #

如果你的耗时函数是IO密集型（比如网络请求、文件操作），线程池的开销更低，推荐用concurrent.futures.ThreadPoolExecutor：

import time
import threading
from concurrent.futures import ThreadPoolExecutor
import random

def f(x, j):
    # 格式化线程名称，匹配示例输出格式
    thread_id = threading.current_thread().name.split('_')[-1]
    print(f"thread {thread_id} : {x*x}, {j}")
    # 模拟0-300秒的随机执行时间
    time.sleep(random.randint(0, 300))

if __name__ == "__main__":
    # 模拟JSON列表，实际场景可替换为json.load读取的列表
    json_list = list(range(10))  # 对应10次调用
    
    # 初始化线程池，固定3个并行线程
    with ThreadPoolExecutor(max_workers=3) as executor:
        task_index = 0
        for item in json_list:
            task_index += 1
            # 提交任务到线程池，自动调度执行
            executor.submit(f, item, task_index)

代码说明 #

• ThreadPoolExecutor(max_workers=3)：指定最多3个工作线程，线程池会始终维持这个数量的并行任务，当某个线程完成任务后，立刻从任务队列中取下一个任务执行。
• executor.submit()：将每个任务提交到线程池，无需手动管理线程的创建和销毁。
• 线程名称格式化：通过threading.current_thread().name获取线程标识，处理后输出和示例一致的thread X格式。

二、进程池方案（CPU密集型场景） #

如果耗时函数是CPU密集型（比如大量计算），由于Python的GIL限制，推荐用multiprocessing.Pool创建进程池：

import time
import multiprocessing
import random

def f(task_args):
    x, j = task_args
    # 格式化进程名称，匹配示例输出格式
    process_id = multiprocessing.current_process().name.split('-')[-1]
    print(f"thread {process_id} : {x*x}, {j}")
    # 模拟0-300秒的随机执行时间
    time.sleep(random.randint(0, 300))

if __name__ == "__main__":
    # 模拟JSON列表
    json_list = list(range(10))
    
    # 初始化进程池，固定3个并行进程
    with multiprocessing.Pool(processes=3) as pool:
        # 构造任务参数列表
        tasks = [(item, idx+1) for idx, item in enumerate(json_list)]
        # 批量提交任务，进程池自动维持3个并行实例
        pool.map(f, tasks)

代码说明 #

• multiprocessing.Pool(processes=3)：创建包含3个进程的进程池，每个进程独立执行任务，不受GIL限制。
• pool.map()：将任务列表批量分配给进程池，自动调度任务，始终维持3个并行进程。
• 进程间参数传递：由于进程间内存隔离，需要将任务参数打包成元组传递。

关键注意事项 #

1. 执行顺序：由于每个任务的执行时间随机，线程/进程的结束顺序可能不同，但线程池/进程池会始终保证同时有3个实例在运行，符合需求。
2. JSON数据处理：实际场景中，只需将json_list替换为json.load(open('your_data.json'))读取的列表即可，调用次数会自动匹配列表长度。
3. Windows兼容性：使用多进程时，必须把代码放在if __name__ == "__main__":块内，避免进程启动时的递归问题。

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