Python 中的并发 —— 进程间通信
摘要：本文介绍了Python中实现进程间通信(IPC)的多种机制：1. 队列(Queue) - 提供线程和进程安全的FIFO通信方式；2. 管道(Pipe) - 通过双向连接对象实现进程间数据传输；3. 管理器(Manager) - 控制共享对象服务进程，支持命名空间(Namespace)实现多属性共享；4. Ctypes - 使用共享内存中的Array和Value对象；5. 通信顺序进程(CSP) - 通过PyCSP库实现基于消息传递的并发模型。每种机制都配有简明示例代码，展示其基本使用方法。这些IPC技术为开发并行应用程序提供了灵活的数据交换方案。
目录
Python 中的并发 —— 进程间通信
通信方式
各类通信机制
队列
队列使用示例
输出结果
管道
管道使用示例
输出结果
管理器
管理器使用示例
输出结果
管理器中的命名空间概念
命名空间使用示例
输出结果
Ctypes 数组与值
Ctypes 数组使用示例
输出结果
通信顺序进程（CSP）
Python 库：PyCSP
安装 PyCSP
并行运行进程示例
输出结果
Python 中的并发 —— 进程间通信
进程间通信指的是进程之间的数据交换，在开发并行应用程序时，进程间的数据交换是必不可少的。下图展示了多个子进程间实现同步的各类通信机制：
通信方式
队列（Queues）管道（Pipes）管理器（Manager）C 类型（Ctypes）
各类通信机制
本节将介绍多种进程间通信机制，具体说明如下：
队列
队列可用于多进程程序中，multiprocessing模块的Queue类与Queue.Queue类功能相似，因此可使用相同的应用程序编程接口（API）。multiprocessing.Queue为进程间通信提供了一种线程和进程安全的先进先出（FIFO）机制。
队列使用示例
以下是取自 Python 多进程官方文档的简单示例，用于理解multiprocessing模块中Queue类的使用原理：main.py
from multiprocessing import Process, Queue import random def f(q): q.put([42, None, 'hello']) def main(): q = Queue() p = Process(target=f, args=(q,)) p.start() print(q.get()) if __name__ == '__main__': main()输出结果
运行代码并验证输出：
plaintext
[42, None, 'hello']管道
管道是一种数据结构，用于多进程程序中的进程间通信。Pipe()函数会返回一对由管道连接的连接对象，该管道默认是双向的。其工作机制如下：
- 返回一对连接对象，分别代表管道的两端；
- 每个对象都拥有
send()（发送）和recv()（接收）两个方法，用于实现进程间的通信。
管道使用示例
以下是取自 Python 多进程官方文档的简单示例，用于理解multiprocessing模块中Pipe()函数的使用原理：main.py
from multiprocessing import Process, Pipe def f(conn): conn.send([42, None, 'hello']) conn.close() if __name__ == '__main__': parent_conn, child_conn = Pipe() p = Process(target=f, args=(child_conn,)) p.start() print(parent_conn.recv()) p.join()输出结果
运行代码并验证输出：
plaintext
[42, None, 'hello']管理器
Manager是multiprocessing模块中的一个类，为所有使用者之间协调共享信息提供了一种方式。管理器对象会控制一个服务进程，该服务进程负责管理共享对象，并允许其他进程对这些对象进行操作。换句话说，管理器提供了一种创建可在不同进程间共享数据的方式。管理器对象具有以下不同特性：
- 核心特性是控制管理共享对象的服务进程；
- 另一重要特性是，当任意进程修改共享对象时，会对所有共享对象进行同步更新。
管理器使用示例
以下示例通过管理器对象在服务进程中创建一个列表记录，然后向该列表中添加一条新记录：
import multiprocessing def print_records(records): for record in records: print("名称：{0}\n分数：{1}\n".format(record[0], record[1])) def insert_record(record, records): records.append(record) print("新增一条记录\n") if __name__ == '__main__': with multiprocessing.Manager() as manager: records = manager.list([('计算机', 1), ('历史', 5), ('语文', 9)]) new_record = ('英语', 3) p1 = multiprocessing.Process(target=insert_record, args=(new_record, records)) p2 = multiprocessing.Process(target=print_records, args=(records,)) p1.start() p2.start() p1.join() p2.join()输出结果
运行代码并验证输出：
plaintext
新增一条记录 名称：计算机 分数：1 名称：历史 分数：5 名称：语文 分数：9 名称：英语 分数：3管理器中的命名空间概念
Manager类内置了命名空间（Namespaces）的概念，这是一种能在多个进程间共享多个属性的便捷方法。命名空间没有可调用的公共方法，但拥有可写的属性。
命名空间使用示例
以下 Python 脚本示例可帮助我们利用命名空间在主进程和子进程间共享数据：main.py
import multiprocessing def Mng_NaSp(using_ns): using_ns.x += 5 using_ns.y *= 10 if __name__ == '__main__': manager = multiprocessing.Manager() using_ns = manager.Namespace() using_ns.x = 1 using_ns.y = 1 print('修改前', using_ns) p = multiprocessing.Process(target=Mng_NaSp, args=(using_ns,)) p.start() p.join() print('修改后', using_ns)输出结果
运行代码并验证输出：
plaintext
修改前 Namespace(x=1, y=1) 修改后 Namespace(x=6, y=10)Ctypes 数组与值
multiprocessing模块提供了Array（数组）和Value（值）对象，用于在共享内存映射中存储数据。Array是从共享内存中分配的 C 类型数组，Value是从共享内存中分配的 C 类型对象。
使用前，需先从multiprocessing中导入Process、Value和Array。
Ctypes 数组使用示例
以下 Python 脚本是取自 Python 官方文档的示例，用于演示如何利用 Ctypes 数组和值实现进程间的数据共享：main.py
import array from multiprocessing import Process def f(n, a): n = 3.1415927 for i in range(len(a)): a[i] = -a[i] if __name__ == '__main__': num = 0.0 arr = array.array('i', range(10)) p = Process(target=f, args=(num, arr)) p.start() p.join() print(num) print(arr[:])输出结果
运行代码并验证输出：
plaintext
3.1415927 [0, -1, -2, -3, -4, -5, -6, -7, -8, -9]通信顺序进程（CSP）
通信顺序进程可用于描述带有并发模型的系统与其他系统之间的交互，它是一种通过消息传递编写并发程序的框架，因此能有效描述并发行为。
Python 库：PyCSP
为实现通信顺序进程中的核心原语，Python 提供了 PyCSP 库。该库的实现代码简洁、可读性强，易于理解。以下是 PyCSP 的基本进程网络结构：通道 2进程 1 进程 2通道 1
在上述 PyCSP 进程网络中，包含进程 1 和进程 2 两个进程，这两个进程通过通道 1 和通道 2 两条通道传递消息实现通信。
安装 PyCSP
可通过以下命令安装 Python 的 PyCSP 库：
plaintext
pip install PyCSP并行运行进程示例
以下 Python 脚本是一个简单的示例，通过 PyCSP 库实现两个进程的并行运行：main.py
from pycsp.parallel import * import time @process def P1(): time.sleep(1) print('P1 进程退出') @process def P2(): time.sleep(1) print('P2 进程退出') def main(): Parallel(P1(), P2()) print('程序终止') if __name__ == '__main__': main()在上述脚本中，定义了 P1 和 P2 两个函数，并通过@process装饰器将其转换为进程。
输出结果
运行代码并验证输出：
plaintext
P2 进程退出 P1 进程退出 程序终止