如何用Python concurrent.futures实现多进程各带线程池的任务排队？

针对你提出的多进程+单进程线程池的混合架构需求，以及动态任务分配的问题，以下是几种无需从零编写执行器的解决方案：

1. 基于concurrent.futures封装混合执行器 #

你可以自己封装一个轻量级的混合执行器，外层用ProcessPoolExecutor管理进程，每个进程内部初始化独立的ThreadPoolExecutor，通过进程安全的队列实现动态任务分配，避免提前拆分任务列表。

示例代码：

from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
from multiprocessing import Queue
import time

def _process_worker(task_queue, result_queue, thread_count):
    """每个进程内部的线程池工作逻辑"""
    with ThreadPoolExecutor(max_workers=thread_count) as thread_pool:
        while True:
            task = task_queue.get()
            if task is None:  # 终止信号
                break
            fn, args, kwargs = task
            # 提交任务到线程池，完成后把结果放入结果队列
            future = thread_pool.submit(fn, *args, **kwargs)
            result_queue.put(future.result())

class HybridExecutor:
    def __init__(self, process_count=4, threads_per_process=8):
        self.process_count = process_count
        self.threads_per_process = threads_per_process
        self.task_queue = Queue()
        self.result_queue = Queue()
        self.process_pool = ProcessPoolExecutor(max_workers=process_count)
        
        # 启动所有进程的工作线程
        for _ in range(process_count):
            self.process_pool.submit(
                _process_worker,
                self.task_queue,
                self.result_queue,
                self.threads_per_process
            )

    def submit(self, fn, *args, **kwargs):
        """提交任务到全局队列"""
        self.task_queue.put((fn, args, kwargs))

    def get_results(self):
        """迭代获取任务结果"""
        while True:
            try:
                yield self.result_queue.get(timeout=1)
            except:
                # 简单判断队列是否为空且进程池已关闭，实际可优化
                if self.task_queue.empty() and self.process_pool._shutdown:
                    break

    def shutdown(self):
        """关闭执行器，发送终止信号"""
        for _ in range(self.process_count):
            self.task_queue.put(None)
        self.process_pool.shutdown()

# 使用示例
def io_task(file_name):
    """模拟IO密集型任务（小文件/网络请求）"""
    time.sleep(0.1)
    return f"Processed {file_name}"

def cpu_task(data):
    """模拟CPU密集型任务（大文件处理）"""
    time.sleep(0.5)
    return data.upper()

if __name__ == "__main__":
    executor = HybridExecutor(process_count=2, threads_per_process=4)
    
    # 提交混合类型任务
    for i in range(10):
        executor.submit(io_task, f"small_file_{i}.txt")
    executor.submit(cpu_task, "large_file_content")
    
    # 处理结果
    for res in executor.get_results():
        print(res)
    
    executor.shutdown()

2. 第三方现成混合执行器实现 #

• loky：作为标准库ProcessPoolExecutor的增强替代，它支持更高效的跨进程任务传递，且天然兼容线程池嵌套使用。你可以直接用loky.ProcessPoolExecutor创建进程池，每个进程内部初始化ThreadPoolExecutor，任务会自动通过loky的队列动态分配，无需手动处理队列同步。

3. 关键优化点 #

• 动态任务分配：用multiprocessing.Queue作为全局任务队列，主进程持续提交任务，子进程的线程池主动拉取任务，避免预拆分任务的局限性。
• 资源配比：根据任务类型调整参数：CPU密集型任务为主时，进程数设为CPU核心数；IO密集型任务为主时，每个进程的线程数可设为10~20（根据系统IO能力调整）。
• 任务粒度控制：将大粒度CPU任务直接交给进程处理，小粒度IO任务交给进程内的线程池，最大化资源利用率。

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