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` 通常是比较好的选择。