multiprocessing

https://docs.python.org/3/library/multiprocessing.html

multiprocessing是一个支持使用与 threading 模块类似的 API 来生成进程的包。multiprocessing包提供了本地和远程的并发性，通过使用子进程而非线程，绕过了全局解释器锁（GIL）的限制。因此，multiprocessing 模块允许程序员充分利用机器上的多个处理器。

当需要并行执行CPU密集型任务，并且计算机有多核CPU时，更适合使用multiprocessing库。如果进程将大部分时间花在等待I/O（例如网络、磁盘、数据库或键盘）上，多进程就不那么有用了，这时候更适合使用多线程或者协程。

Pool

multiprocessing 模块还引入了一些在 threading 模块中没有对应的 API。其中一个重要的例子是 Pool 对象，它提供了一种简便的方法，可以在多个输入值上并行执行函数，将输入数据分布到不同的进程中（数据并行）。下面是一个使用 Pool 进行数据并行的基本示例。

from multiprocessing import Pool


def f(x):
    return x * x


if __name__ == "__main__":
    with Pool(5) as p:
        print(p.map(f, [1, 2, 3]))
# 输出 [1, 4, 9]

进程池multiprocessing.pool.Pool隐藏了进程间通信的细节，使用Pool看起来很像函数调用：将数据传递给一个函数，它在另一个或多个进程中执行，当工作完成时，返回一个值。这依赖于背后的很多工作：一个进程中的对象使用pickle进行序列化并被传递到操作系统进程管道中。然后，另一个进程从管道中获取数据并反序列化。所需工作在子进程中完成，并产生一个结果。结果被序列化，并通过管道传递回来。最终，主进程会反序列化并返回它。总

相关内容：concurrent.futures.ProcessPoolExecutor

concurrent.futures.ProcessPoolExecutor 提供了一个更高级的接口，可将任务推送到后台进程，而不会阻塞调用进程的执行。与直接使用 Pool 接口相比，concurrent.futures API 更便于将任务提交到底层进程池，并可以与等待结果的过程分开。

Process

可以通过创建一个Process对象并调用其start()方法来生成新进程。

from multiprocessing import Process
import os


def info(title):
    print(title)
    print("module name:", __name__)
    print("parent process:", os.getppid())
    print("process id:", os.getpid())


def f(name):
    info("function f")
    print("hello", name)


if __name__ == "__main__":
    info("main line")
    p = Process(target=f, args=("bob",))
    p.start()
    p.join()

进程间通信和`multiprocessing.Queue`

如果需要对进程间的通信进行更多的控制，可以使用数据结构multiprocessing.Queue。它提供了几种从一个进程向另一个或多个其他进程发送消息的方法。任何可以被pickle序列化的对象都可以被发送到一个Queue中，但是pickle可能是一个开销很大的操作，所以应尽量让这种对象很小。

from multiprocessing import Process, Queue
import time


def producer(queue):
    """生产者进程，将数据放入队列"""
    for i in range(5):
        item = f"数据 {i}"
        print(f"生产者：将 {item} 放入队列")
        queue.put(item)
        time.sleep(1)  # 模拟生产延迟
    queue.put(None)  # 表示结束信号


def consumer(queue):
    """消费者进程，从队列中获取数据"""
    while True:
        item = queue.get()
        if item is None:  # 检查是否为结束信号
            break
        print(f"消费者：从队列中取出 {item}")
        time.sleep(2)  # 模拟处理延迟


if __name__ == "__main__":
    # 创建一个队列用于进程间通信
    queue = Queue()

    # 创建生产者和消费者进程
    producer_process = Process(target=producer, args=(queue,))
    consumer_process = Process(target=consumer, args=(queue,))

    # 启动进程
    producer_process.start()
    consumer_process.start()

    # 等待进程完成
    producer_process.join()
    consumer_process.join()

    print("所有进程已完成")