如何在 Python 中分别启动和停止多处理进程？答案

【问题标题】：How to separately start and stop multiprocessing processes in Python?如何在 Python 中分别启动和停止多处理进程？
【发布时间】：2022-09-23 05:29:16
【问题描述】：

我使用专用的 Python (3.8) 库通过 USB 端口控制电机驱动器。

电机控制驱动器制造商 (ODrive) 提供的 Python 库允许单个 Python 进程控制一个或多个驱动器。

但是，我想运行 3 个进程，每个进程控制 1 个驱动器。

在研究了各种选项（我首先考虑了虚拟机、Docker 容器和多线程）之后，我开始相信最简单的方法是使用multiprocessing。

我的问题是我需要一种方法来管理（即独立启动、监控和停止）多个进程。其背后的实际原因是电机连接到不同的设置。例如，如果出现故障，每个设置都必须能够单独停止和重新启动，但其他正在运行的设置不应受到此操作的影响。

在阅读了互联网和 Stack Overflow 之后，我现在了解了如何创建处理的 Pool、如何将进程与处理器内核关联、如何启动进程池以及对它们进行排队/加入（后者不需要我）。

我不知道如何独立管理它们。如何在不影响其他进程执行的情况下分别启动/停止不同进程？是否有库来管理它们（甚至可能使用 GUI）？

每个进程都可以有自己的消息队列，如果您不使用池，您可以在其中发送消息以停止等。
@MattClarke，它们也可以单独启动吗？在已经从 Python 脚本启动其他进程之后，我不确定如何实际设想如何告诉设置“开始执行功能，使用这些参数”。
如果您不使用池而是手动创建进程，那么您可以更精细地控制它们何时启动等。TBH 它并不比使用池复杂得多。
@MattClarke，感谢您的提示。在实践中，我宁愿使用一种“控制台方法”，而不是坚持使用“脚本方法”。因此，我每次都必须通过独立命令手动运行（和管理）进程。在您看来，我是否正确地看到了这一点？假设这种方法，是否有一种简单的方法/命令来监视当前正在运行的内容？例如，进程 X 在核心 2 上运行；在核心 3 上处理 Y； Core 4 是免费的吗？
我认为你的建议是可以实现的。我会给每个进程两个队列：一个用于控制，一个用于报告状态信息。然后可以查询“报告”队列以获取相应的状态（例如电机位置）。进程本身有一个is_alive 方法来查看它们是否启动。

标签： python python-3.x multiprocessing python-multiprocessing

【解决方案1】：

您可以像这样手动创建多个 multriprocessing.Process 实例：

def my_func(a, b):
    pass

p = multiprocessing.Process(target=my_func, args=(100, 200)
p.start()

并使用多处理原语Queue、Event、Condition等对其进行管理。详情请参考官方文档：https://docs.python.org/3/library/multiprocessing.html

在以下示例中，多个进程独立启动和停止。 Event 用于确定何时停止进程。 Queue 用于将结果从子进程传递到主进程。

import multiprocessing
import queue
import random
import time


def worker_process(
    process_id: int,
    results_queue: multiprocessing.Queue,
    to_stop: multiprocessing.Event,
):

    print(f"Process {process_id} is started")
    while not to_stop.is_set():
        print(f"Process {process_id} is working")
        time.sleep(0.5)
        result = random.random()
        results_queue.put((process_id, result))

    print(f"Process {process_id} exited")


process_pool = []
result_queue = multiprocessing.Queue()
while True:
    if random.random() < 0.3:
        # staring a new process
        process_id = random.randint(0, 10_000)
        to_stop = multiprocessing.Event()
        p = multiprocessing.Process(
            target=worker_process, args=(process_id, result_queue, to_stop)
        )
        p.start()
        process_pool.append((p, to_stop))

    if random.random() < 0.2:
        # closing a random process
        if process_pool:
            process, to_stop = process_pool.pop(
                random.randint(0, len(process_pool) - 1)
            )
            to_stop.set()
            process.join()

    try:
        p_id, result = result_queue.get_nowait()
        print(f"Completed: process_id={p_id} result={result}")
    except queue.Empty:
        pass

    time.sleep(1)

【讨论】：