我如何使用 Python 的 concurrent.futures 跨多个进程对任务进行排队，每个进程都有自己的线程池？答案

【问题标题】：How can I use Python's concurrent.futures to queue tasks across multiple processes each with their own thread pool?我如何使用 Python 的 concurrent.futures 跨多个进程对任务进行排队，每个进程都有自己的线程池？
【发布时间】：2023-01-30 00:27:49
【问题描述】：

我正在研究一个库函数，它使用 concurrent.futures 将网络 I/O 分布到多个线程中。由于 Python GIL，我遇到了一些工作负载（大文件）速度变慢的问题，所以我想切换到多个进程。但是，对于某些其他工作负载（许多小文件），多个进程也不太理想。我想拆分差异并拥有多个进程，每个进程都有自己的线程池。

问题是作业排队 - concurrent.futures 似乎没有设置为多个进程正确排队作业，每个进程都可以一次处理多个作业。虽然提前将作业列表分成块是一种选择，但如果作业在各个线程完成任务时异步地流向每个进程，则工作会更加顺利。

如何使用此 API 或类似 API 跨多个进程和线程有效地排队作业？除了编写我自己的执行程序之外，是否有任何我忽略的明显解决方案？或者是否有混合进程/线程执行器的现有技术？

【问题讨论】：

我认为您创建了一个队列（使用multiprocessing.Manager）来为您的流程工作人员提供工作。您创建了一个带有 N 个进程的 multiprocessing.Pool，并且在每个进程中您从队列中提取项目并将它们提交给一个带有 M 个最大工作线程的 concurrent futures.ThreadPoolExecutor。
@larsks multiprocessing.Queue 实例会更有效率。每个池进程都可以使用队列实例初始化初始值设定项和初始化参数ProcessPoolExecutor 初始值设定项的参数。

标签： python python-3.x multiprocessing concurrent.futures queueing

【解决方案1】：

如果我明白你想做什么，你基本上有很多适合多线程的工作，除了有一些 CPU 密集型工作。所以你的想法是在多个子进程中创建多个线程池，从而减少 GIL 争用。当然，在任何给定的子进程中，CPU 密集型代码只会串行执行（假设它是 Python 字节码），因此这不是一个完美的解决方案。

一种方法是创建一个非常大的多处理池（大于您拥有的内核数）。您可以创建的进程数量是有限的，而且创建它们的成本很高。但由于大部分时间他们将等待 I/O 完成，因此 I/O 部分将并发执行。

更好的方法是创建一个多处理池，其执行程序可以与其他所需参数一起传递给多线程池工作函数。这是你计划做的事情的倒置。当 worker 函数有一个 CPU 密集型工作要执行时，它可以将该工作提交给传递的多处理池执行器并阻塞返回的结果。通过这种方式，您可以获得最佳的并行性，您可以在给定您拥有的内核数量的情况下实现。这是我的建议。

但是如果你想坚持你最初的想法，也许像下面这样的东西可能会奏效：

from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor, as_completed
from multiprocessing import Queue
from queue import Empty

def init_pool_processes(q):
    global queue, thread_pool_executor

    queue = q
    thread_pool_executor = ThreadPoolExecutor(10) # or some appropriate pool size


def thread_worker(x):
    import time

    # Do something with x
    ...
    time.sleep(.1) # simulate time taken
    return x # Just for demo purposes

def process_worker(y):
    # This results in some number of threadpool jobs:
    futures = [thread_pool_executor.submit(thread_worker, y + i) for i in range(5)]
    for future in as_completed(futures):
        queue.put(future.result())


if __name__ == '__main__':
    results = []

    def get_results(result):
        try:
            while True:
                result = queue.get_no_wait()
                results.append(result)
        except Empty:
            pass

    input_args = (100, 200, 300, 400, 500)
    queue = Queue()
    with ProcessPoolExecutor(initializer=init_pool_processes, initargs=(queue,)) as executor:
        futures = [executor.submit(process_worker, input_arg) for input_arg in input_args]
        for future in as_completed(futures):
            # Every time a job submitted to the process pool completes we can
            # look for more results:
            try:
                while True:
                    result = queue.get_nowait()
                    results.append(result)
            except Empty:
                pass
    print(results)

印刷：

[102, 201, 101, 203, 103, 202, 200, 100, 104, 204, 504, 301, 404, 502, 304, 403, 302, 501, 503, 500, 402, 303, 401, 300, 400]

【讨论】：