Python multiprocessing实现分块k个task1完成后执行task2的方法

标准库multiprocessing不支持嵌套池的本质原因是池的worker进程默认是daemon进程，daemon进程不允许再启动子进程。你不需要创建嵌套进程池，全程复用主进程初始化的同一个进程池，配合异步任务回调做分块完成判定即可，完全满足所有需求。

核心逻辑 #

• 提前将原始列表lst按指定块长k切分，为每个分块维护独立的task1结果收集容器、完成计数器，注意最后一个分块长度可能不足k，要按实际长度做完成判定
• 用非阻塞的异步方式向进程池提交所有单元素的task1任务，不要用map这类会阻塞等待全量任务完成的接口
• 每个task1任务完成后触发主进程侧的回调函数，将结果存入对应分块的收集容器，判定该分块是否所有元素都完成task1处理
• 一旦某分块满足task1全量完成条件，立刻将该分块的task1结果集作为参数，向同一个进程池异步提交task2任务，无需等待其他分块的task1执行
• 进程池的worker会自动从任务队列取待执行任务，不管是task1还是task2，自然实现两类任务并行调度，不会出现等所有task1跑完才跑task2的阻塞问题。

最小实现示例 #

import multiprocessing
import time
import random

# 单元素处理逻辑
def task1(item):
    time.sleep(random.uniform(0.1, 0.5))  # 模拟实际计算/IO耗时
    return f"task1_processed_{item}"

# 分块批量处理逻辑
def task2(chunk_task1_results):
    time.sleep(random.uniform(0.2, 0.7))  # 模拟实际处理耗时
    return f"task2_finished: handled {len(chunk_task1_results)} items in chunk"

if __name__ == "__main__":
    # 初始化参数
    n = 20
    k = 4
    p = 4
    lst = list(range(n))

    # 切分原始列表
    chunks = [lst[i:i+k] for i in range(0, n, k)]
    chunk_count = len(chunks)
    # 状态追踪：每个分块的task1结果、已完成task1的计数
    chunk_task1_res = {idx: [] for idx in range(chunk_count)}
    chunk_task1_done_num = {idx: 0 for idx in range(chunk_count)}
    # 给每个元素绑定所属分块ID、分块实际长度
    task_items = []
    for chunk_id, chunk in enumerate(chunks):
        real_len = len(chunk)
        for item in chunk:
            task_items.append((chunk_id, real_len, item))

    # 收集最终task2的返回结果
    final_results = []
    def task2_done_callback(res):
        final_results.append(res)

    # 初始化唯一的进程池
    pool = multiprocessing.Pool(processes=p)

    # task1完成后的回调，负责判定分块是否完成、提交task2
    def handle_task1_done(args):
        chunk_id, chunk_real_len, task1_res = args
        chunk_task1_res[chunk_id].append(task1_res)
        chunk_task1_done_num[chunk_id] += 1
        # 分块所有task1完成，立刻提交task2
        if chunk_task1_done_num[chunk_id] == chunk_real_len:
            pool.apply_async(
                task2,
                args=(chunk_task1_res[chunk_id],),
                callback=task2_done_callback
            )

    # 异步提交所有task1任务，注意lambda用默认参数绑定当前循环的分块信息，避免闭包晚绑定问题
    for chunk_id, real_len, item in task_items:
        pool.apply_async(
            task1,
            args=(item,),
            callback=lambda res, cid=chunk_id, rlen=real_len: handle_task1_done((cid, rlen, res))
        )

    # 等待所有任务执行完成
    pool.close()
    pool.join()

    print(final_results)

注意事项 #

• 所有回调函数运行在主进程中，修改分块状态、提交新任务的操作不需要加进程锁，无并发冲突风险
• Windows/macOS默认用spawn方式启动进程，所有多进程逻辑必须放在if __name__ == "__main__":作用域下，否则会重复启动进程报错
• 如果待处理列表体量极大，不要一次性提交全部task1任务，可加简单流控：维持在途任务数为进程数p的2-3倍即可，避免任务队列占用过多内存
• 如果task2需要原始分块的输入值而非仅task1结果，在初始切分块时同步保存原始分块数据即可，核心逻辑无需改动。
• 如果你习惯用更现代的并发API，也可以替换为concurrent.futures.ProcessPoolExecutor，实现逻辑完全一致，用add_done_callback方法绑定完成回调即可。

内容的提问来源于stack exchange，提问作者A. Maman