Python 多进程的基本使用

1. 引言

在计算机领域，进程是指计算机中正在运行的程序的实例。一个程序可以由多个进程组成，每个进程都是独立运行的，它们有自己独立的内存空间和计算资源。多进程编程是一种利用多个进程并行处理任务的方法，能够提高程序的执行效率。

Python 是一种高级编程语言，提供了多进程编程的支持。本文将介绍 Python 中多进程编程的基本使用方法，包括如何创建和管理进程、进程间通信等内容。

2. 使用 `multiprocessing` 模块创建进程

Python 提供了 multiprocessing 模块来创建和管理进程。下面是使用 multiprocessing 模块创建进程的示例代码：

import multiprocessing

def worker():
    print("Worker")

if __name__ == "__main__":
    p = multiprocessing.Process(target=worker)
    p.start()
    p.join()
    print("Main Process")

在上面的代码中，我们首先导入了 multiprocessing 模块。然后，我们定义了一个名为 worker 的函数，该函数将作为新进程的入口点。在 worker 函数中，我们只是简单地打印出一行文本。

在 if __name__ == "__main__" 的代码块内，我们创建了一个 Process 对象 p，并将 worker 函数设置为该进程的目标函数。然后，我们通过 start() 方法启动进程。最后，我们调用 join() 方法来等待进程运行结束。这样，我们就创建了一个子进程并让它执行 worker 函数。

运行上述代码，我们将会看到如下的输出：

Worker
Main Process

可以看到，首先打印出了 Worker，然后打印出了 Main Process。这说明子进程和主进程是并行执行的。

3. 进程间通信

在多进程编程中，不同的进程之间需要进行通信，以便共享数据或协同完成任务。Python 提供了多种进程间通信的方法，如 Queue、Pipe 等。下面我们将介绍一些常用的进程间通信方式。

3.1 使用 `Queue` 进行进程间通信

Queue 是一个线程安全的队列，可以安全地存取数据。在多进程编程中，我们可以使用 Queue 来实现进程间通信。

下面是使用 Queue 进行进程间通信的示例代码：

import multiprocessing

def worker(q):
    data = q.get()
    print("Worker: {}".format(data))

if __name__ == "__main__":
    q = multiprocessing.Queue()
    q.put("Hello")

    p = multiprocessing.Process(target=worker, args=(q,))
    p.start()
    p.join()

    print("Main Process")

在上面的代码中，我们首先导入了 multiprocessing 模块。然后，我们定义了一个名为 worker 的函数，该函数将作为新进程的入口点。在 worker 函数中，我们使用 q.get() 方法从 Queue 中获取数据，并打印出来。

在 if __name__ == "__main__" 的代码块内，我们创建了一个 Queue 对象 q，并使用 q.put() 方法将数据 "Hello" 放入队列中。

然后，我们创建了一个子进程，并将 q 作为参数传递给子进程。子进程在执行时，会调用 worker 函数并将 q 作为参数传入。最后，我们调用 join() 方法来等待子进程运行结束。

运行上述代码，我们将会看到如下的输出：

Worker: Hello
Main Process

可以看到，子进程成功从队列中获取到了数据，并打印出来。

3.2 使用 `Pipe` 进行进程间通信

Pipe 是一种双向管道，可以用于进程间的双向通信。在多进程编程中，我们可以使用 Pipe 实现进程间通信。

下面是使用 Pipe 进行进程间通信的示例代码：

import multiprocessing

def worker(conn):
    data = conn.recv()
    print("Worker: {}".format(data))

    conn.send("World")

if __name__ == "__main__":
    parent_conn, child_conn = multiprocessing.Pipe()

    parent_conn.send("Hello")

    p = multiprocessing.Process(target=worker, args=(child_conn,))
    p.start()
    p.join()

    print("Main Process: {}".format(parent_conn.recv()))

在上面的代码中，我们首先导入了 multiprocessing 模块。然后，我们定义了一个名为 worker 的函数，该函数将作为新进程的入口点。在 worker 函数中，我们使用 conn.recv() 方法从 Pipe 中接收数据，并打印出来。然后，我们使用 conn.send() 方法向 Pipe 发送数据。

在 if __name__ == "__main__" 的代码块内，我们使用 multiprocessing.Pipe() 创建了一个双向管道，并返回两个连接对象 parent_conn 和 child_conn。然后，我们使用 parent_conn.send() 方法向管道发送数据。

接着，我们创建了一个子进程，并将 child_conn 作为参数传递给子进程。子进程在执行时，会调用 worker 函数并将 child_conn 作为参数传入。最后，我们调用 join() 方法来等待子进程运行结束。

运行上述代码，我们将会看到如下的输出：

Worker: Hello
Main Process: World

可以看到，子进程成功从管道中接收到了数据，并打印出来。同时，父进程也从管道中接收到了子进程发送的数据，并打印出来。

4. 进程池

在实际应用中，我们可能需要创建多个进程来处理任务，而不是仅仅一个进程。如果我们需要频繁地创建和销毁进程，将会消耗大量的系统资源。为了避免这种情况，我们可以使用进程池来复用进程。

Python 中的 multiprocessing 模块提供了 Pool 类来实现进程池。下面是使用进程池的示例代码：

import multiprocessing

def worker(x):
    return x * x

if __name__ == "__main__":
    with multiprocessing.Pool(processes=4) as pool:
        results = pool.map(worker, [1, 2, 3, 4, 5])

    print("Results: {}".format(results))

在上面的代码中，我们首先导入了 multiprocessing 模块。然后，我们定义了一个名为 worker 的函数，该函数将作为进程池中进程的任务。在 worker 函数中，我们接收一个参数，并返回该参数的平方值。

在 if __name__ == "__main__" 的代码块内，我们使用 multiprocessing.Pool() 创建了一个进程池，并指定了进程池的大小为4。然后，我们使用 pool.map() 方法来处理一个可迭代对象 [1, 2, 3, 4, 5]，其中每个元素都会被传递给 worker 函数进行处理。pool.map() 方法会返回一个包含结果的列表。

最后，我们打印出了结果。

运行上述代码，我们将会看到如下的输出：

Results: [1, 4, 9, 16, 25]

可以看到，进程池成功地对输入的每个元素进行处理，并返回了结果。

5. 进程间锁机制

在多进程编程中，如果多个进程同时访问共享资源，会导致数据不一致的问题。为了避免这种情况，我们可以使用进程间的锁机制来保证关键区域的互斥访问。

Python 中的 multiprocessing 模块提供了 Lock 类来实现进程间的锁机制。下面是使用进程间锁的示例代码：

import multiprocessing

def worker(lock, counter):
    with lock:
        for _ in range(100000):
            counter.value += 1

if __name__ == "__main__":
    lock = multiprocessing.Lock()
    counter = multiprocessing.Value('i', 0)

    processes = []
    for _ in range(4):
        p = multiprocessing.Process(target=worker, args=(lock, counter))
        processes.append(p)
        p.start()

    for p in processes:
        p.join()

    print("Counter: {}".format(counter.value))

在上面的代码中，我们首先导入了 multiprocessing 模块。然后，我们定义了一个名为 worker 的函数，该函数将作为新进程的入口点。在 worker 函数中，我们使用 with lock: 语句来获取锁，并在关键区域内对共享资源进行操作。

在 if __name__ == "__main__" 的代码块内，我们使用 multiprocessing.Lock() 创建了一个锁对象 lock，并使用 multiprocessing.Value() 创建了一个共享变量 counter。counter 的初始值为0。

然后，我们创建了一个空的列表 processes，用于存放子进程的引用。接着，我们使用一个循环创建了4个子进程，并将它们的引用添加到 processes 列表中。每个子进程都会执行 worker 函数，并传递 lock 和 counter 作为参数。

最后，我们使用另一个循环调用 join() 方法等待所有子进程执行完毕。最终，我们打印出了共享变量 counter 的值。

运行上述代码，我们将会看到如下的输出：