在 Python 中使用多处理的简单流程管理器

Question

我有很多任务（彼此独立，由 Python 中的一些代码表示）需要执行。它们的执行时间各不相同。我也有有限的资源，所以最多N 个任务可以同时运行。目标是尽快完成整个任务堆栈的执行。

似乎我正在寻找某种管理器，当资源可用时开始新任务并收集已完成的任务。

• 是否有任何已经制定的解决方案或者我应该自己编写代码？
• 有什么需要注意的地方吗？

Answer 1

据我所知，您的 main 会变成：

def main():
    tasks = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
    with multiprocessing.Pool(POOL_SIZE) as pool:
        pool.map(sleep, tasks)

即您刚刚重新实现了一个池，但效率低下（Pool 在可能的情况下重用了Processes）并且不那么安全，Pool 付出了很多努力来清理异常

Answer 2

这是一个简单的代码 sn-p 应该符合要求：

import multiprocessing
import time

POOL_SIZE = 4
STEP = 1


def sleep(seconds: int):
    time.sleep(seconds)


def main():
    tasks = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
    pool = [None] * POOL_SIZE

    while tasks or [item for item in pool if item is not None]:
        for i in range(len(pool)):
            if pool[i] is not None and not pool[i].is_alive():
                # Finished task. Clear the resource.
                pool[i] = None

            if pool[i] is None:
                # Free resource. Start new task if any are left.
                if tasks:
                    task = tasks.pop(0)
                    pool[i] = multiprocessing.Process(target=sleep, args=(task,))
                    pool[i].start()

        time.sleep(STEP)


if __name__ == '__main__':
    main()

管理器有一个任意长度的tasks 列表，为了简单起见，这里是用整数表示的任务，它们被放置为sleep 函数的参数。它还有一个pool 列表，最初为空，表示可用资源。

管理器定期访问所有当前正在运行的进程并检查它们是否已完成。如果资源可用，它还会启动新进程。整个循环一直在重复，直到没有任务并且没有当前正在运行的进程。 STEP 值在这里是为了节省计算能力——你通常不需要每毫秒检查一次正在运行的进程。

至于注意事项，there are some guidelines 在使用多处理时应牢记。