multiprocessing.Pool.map无法并行工作，如何实现进程-线程嵌套并行？

兄弟，我一眼就瞅出你代码里的关键问题啦！你写的process_pool.map(do_stuff, [drain_queue()])里，第二个参数是一个包裹着任务列表的单元素列表——这就导致map只会把整个任务列表丢给唯一的一个进程去处理，其他进程全程摸鱼，自然没法实现你想要的多进程并行效果。

核心修正思路 #

要让进程池里的每个进程都动起来，你需要把map的第二个参数改成任务的直接迭代器：也就是直接传drain_queue()返回的任务列表，而不是把它套在另一个列表里。这样进程池会自动把任务分配给不同的进程，每个进程再启动16个线程来处理自己拿到的任务（如果任务量很大，也可以先把任务分成和进程数匹配的批次，让每个进程处理一批）。

修正后的代码示例 #

场景1：每个进程处理单个任务，任务内嵌套线程 #

如果drain_queue()返回的是单个任务的列表，直接把这个列表传给map即可：

import os
import multiprocessing
from concurrent.futures import ThreadPoolExecutor

def do_stuff(task):
    print(f'> PID: {os.getpid()} 正在处理任务: {task}')
    # 进程内启动16线程处理当前任务的子操作
    with ThreadPoolExecutor(max_workers=16) as executor:
        # 假设每个task包含多个子任务，比如task.sub_tasks
        sub_results = list(executor.map(process_sub_task, task["sub_tasks"]))
    return sub_results

def process_sub_task(sub_task):
    # 子任务的具体执行逻辑
    return sub_task * 2

def drain_queue():
    # 模拟返回任务列表，每个任务包含若干子任务
    return [{"sub_tasks": [i, i+1, i+2]} for i in range(8)]

if __name__ == '__main__':
    # 启动4个进程的进程池
    with multiprocessing.Pool(4) as process_pool:
        # 直接传入任务列表，进程池自动分配任务给不同进程
        final_results = process_pool.map(do_stuff, drain_queue())
    print('所有任务执行完成！')

场景2：每个进程处理一批任务，批次内嵌套线程 #

如果任务量极大，不想让进程池频繁分配单个任务，可以先把任务分成和进程数对应的批次：

import os
import multiprocessing
from concurrent.futures import ThreadPoolExecutor
from itertools import islice

def do_stuff(task_batch):
    print(f'> PID: {os.getpid()} 处理批次任务，共{len(task_batch)}个任务')
    with ThreadPoolExecutor(max_workers=16) as executor:
        batch_results = list(executor.map(process_single_task, task_batch))
    return batch_results

def process_single_task(task):
    # 单个任务的具体执行逻辑
    return task * 2

def drain_queue():
    # 模拟返回大量任务
    return list(range(100))

def split_tasks(tasks, num_processes):
    # 把任务分成num_processes个均等批次
    batch_size = len(tasks) // num_processes
    batches = []
    task_iter = iter(tasks)
    for _ in range(num_processes):
        batch = list(islice(task_iter, batch_size))
        if batch:
            batches.append(batch)
    # 处理剩余的零散任务
    remaining_tasks = list(task_iter)
    if remaining_tasks:
        batches[-1].extend(remaining_tasks)
    return batches

if __name__ == '__main__':
    process_count = 4
    all_tasks = drain_queue()
    task_batches = split_tasks(all_tasks, process_count)
    
    with multiprocessing.Pool(process_count) as process_pool:
        final_results = process_pool.map(do_stuff, task_batches)
    print('所有批次任务执行完成！')

额外注意事项 #

• 务必把多进程相关代码放在if __name__ == '__main__':块内，这是Windows系统下多进程运行的必要条件，能避免无限创建子进程的坑。
• 如果任务涉及共享资源（比如数据库连接），每个进程要单独创建连接，别在主进程创建后传给子进程，会出现资源冲突问题。
• 线程池大小（16）要根据任务类型调整：IO密集型任务可以设大一点，CPU密集型任务别超过CPU核心数，不然会增加线程切换开销。

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