Python多进程的实现

在 Python 中使用多进程主要通过内置的 `multiprocessing` 模块。下面介绍几种常用的方式：

**1. 使用 `Process` 类（基础方式）**

这是最基本的方式，手动创建和管理进程。

```python

import multiprocessing

import time

import os

# 定义子进程要执行的任务函数

def worker(num):

"""子进程执行的任务"""

pid = os.getpid() # 获取当前进程ID

print(f'Worker {num} started, PID: {pid}')

time.sleep(num) # 模拟耗时操作

print(f'Worker {num} finished, PID: {pid}')

if __name__ == '__main__':

print(f'Main process started, PID: {os.getpid()}')

# 创建进程列表

processes = []

for i in range(1, 4): # 创建 3 个子进程

# 创建 Process 对象

# target=worker 指定子进程要执行的函数

# args=(i,) 是传递给 worker 函数的参数 (注意是元组)

p = multiprocessing.Process(target=worker, args=(i,))

processes.append(p)

p.start() # 启动进程

print('All processes started.')

# 等待所有子进程结束

# join() 会阻塞主进程，直到子进程执行完毕

for p in processes:

p.join()

print('All processes finished. Main process exiting.')

# --- 重要说明 ---

# if __name__ == '__main__': 这行代码非常重要！

# 在创建子进程时，子进程会导入主模块的代码。

# 如果没有这行保护，创建进程的代码会被子进程再次执行，导致无限创建进程（尤其在 Windows 上）。

# 所以，所有创建和启动进程的代码都应该放在这个 if 语句块内。

```

**工作流程:**

1. `import multiprocessing`：导入模块。

2. `def worker(num):`：定义子进程要执行的代码逻辑。

3. `if __name__ == '__main__':`：保护入口点（非常重要）。

4. `p = multiprocessing.Process(target=worker, args=(i,))`：创建一个 `Process` 对象，指定目标函数 (`target`) 和参数 (`args`)。

5. `p.start()`：启动子进程。此时子进程开始执行 `worker` 函数。

6. `p.join()`：主进程等待子进程 `p` 执行结束。如果不调用 `join()`，主进程可能会在子进程完成前就退出了。

**2. 使用 `Pool` 类（进程池，推荐用于批量任务）**

当你有很多相似的任务需要并行处理时，手动管理 `Process` 对象会很繁琐。`Pool` 可以创建一个固定数量的进程池，自动管理任务的分配和进程的复用。

```python

import multiprocessing

import time

import os

def square(x):

"""计算平方的任务"""

pid = os.getpid()

result = x * x

print(f'Task {x} processed by PID: {pid}, result: {result}')

time.sleep(1) # 模拟耗时

return result

if __name__ == '__main__':

print(f'Main process started, PID: {os.getpid()}')

# 创建一个包含 3 个进程的进程池

# 如果不指定数量，通常会根据 CPU 核心数创建

pool = multiprocessing.Pool(processes=3)

tasks = range(10) # 要处理的任务数据

# --- 常用方法 ---

# 1. map: 阻塞方式，将任务分配给进程池，等待所有结果返回

# 它会将 tasks 列表中的每个元素传递给 square 函数

print("Using pool.map():")

results_map = pool.map(square, tasks)

print(f"Map results: {results_map}")

print("-" * 20)

# 2. apply_async: 异步方式，提交任务，不阻塞主进程

# 需要手动获取结果

print("Using pool.apply_async():")

async_results = []

for task in tasks:

# 提交任务到进程池，返回一个 AsyncResult 对象

result_obj = pool.apply_async(square, args=(task,))

async_results.append(result_obj)

# 获取异步任务的结果

# result_obj.get() 会阻塞，直到该任务完成并返回结果

final_results_async = [res.get() for res in async_results]

print(f"Async results: {final_results_async}")

print("-" * 20)

# --- 关闭进程池 ---

# close() 告诉进程池不再接受新的任务

pool.close()

# join() 等待进程池中所有任务执行完毕（必须在 close() 之后调用）

pool.join()

print('All tasks finished. Main process exiting.')

```

**`Pool` 的关键点:**

* `multiprocessing.Pool(processes=N)`：创建包含 N 个工作进程的池。

* `pool.map(func, iterable)`：将 `iterable` 中的每个元素作为参数传递给 `func` 函数，并行执行，然后收集所有结果并返回一个列表。**这是阻塞的**，会等到所有任务完成。

* `pool.apply_async(func, args=(...))`：异步提交单个任务。它立即返回一个 `AsyncResult` 对象，你可以稍后通过该对象的 `get()` 方法获取结果。**这是非阻塞的**。

* `pool.close()`：关闭进程池，使其不再接受新任务。

* `pool.join()`：等待所有工作进程退出。通常在 `close()` 之后调用。

**3. 进程间通信 (IPC - Inter-Process Communication)**

由于进程拥有独立的内存空间，它们不能像线程那样直接共享变量。如果进程间需要交换数据，需要使用特殊的 IPC 机制，`multiprocessing` 模块提供了几种方式：

* **`Queue`:** 线程/进程安全的队列，用于在多个生产者和消费者进程之间传递消息（对象）。

* **`Pipe`:** 返回一对连接的 `Connection` 对象，代表管道的两端，可以用于两个进程之间的双向通信。

* **`Value` / `Array`:** 用于在进程间共享简单的 C 类型数据（如整数、浮点数、字符数组），需要配合锁（`Lock`）来保证同步。

* **`Manager`:** 提供一种更高级的方式来共享 Python 对象（如列表、字典）。它启动一个管理进程来维护这些共享对象，并允许其他进程通过代理访问它们，内部处理了同步问题。

**`Queue` 示例:**

```python

import multiprocessing

import time

def writer(q):

"""向队列写入数据"""

print(f'Writer process started (PID: {os.getpid()})')

for i in ['A', 'B', 'C', 'D']:

print(f'Putting {i} into queue')

q.put(i)

time.sleep(0.5)

q.put(None) # 发送结束信号

def reader(q):

"""从队列读取数据"""

print(f'Reader process started (PID: {os.getpid()})')

while True:

item = q.get() # 获取数据，如果队列为空会阻塞

if item is None: # 收到结束信号

print('Reader received None, exiting.')

break

print(f'Got {item} from queue')

time.sleep(1)

if __name__ == '__main__':

# 创建一个进程安全的队列

q = multiprocessing.Queue()

# 创建并启动读写进程

p_writer = multiprocessing.Process(target=writer, args=(q,))

p_reader = multiprocessing.Process(target=reader, args=(q,))

p_writer.start()

p_reader.start()

# 等待进程结束

p_writer.join()

p_reader.join()

print("Main process finished.")

```

**总结:**

* 对于简单地并行执行几个独立的任务，使用 `Process` 类。

* 对于大量相似的任务，需要高效管理和复用进程，使用 `Pool` 类（通常更方便）。

* 当进程之间需要交换数据时，使用 `Queue`、`Pipe` 或 `Manager` 等 IPC 机制。

* **永远记住** 将创建和启动进程的代码放在 `if __name__ == '__main__':` 块内。

选择哪种方式取决于你的具体需求。对于利用多核 CPU 进行计算密集型任务，`Pool` 通常是比较好的选择。