189 8069 5689
这篇文章主要介绍“Python中怎么使用multiprocessing实现进程间通信”的相关知识，小编通过实际案例向大家展示操作过程，操作方法简单快捷，实用性强，希望这篇“Python中怎么使用multiprocessing实现进程间通信”文章能帮助大家解决问题。
创新互联建站为您提适合企业的网站设计 让您的网站在搜索引擎具有高度排名，让您的网站具备超强的网络竞争力！结合企业自身，进行网站设计及把握，最后结合企业文化和具体宗旨等，才能创作出一份性化解决方案。从网站策划到成都网站设计、成都网站制作， 我们的网页设计师为您提供的解决方案。
python的多线程代码效率由于受制于GIL，不能利用多核CPU来加速，而多进程方式可以绕过GIL, 发挥多CPU加速的优势，能够明显提高程序的性能
但进程间通信却是不得不考虑的问题。 进程不同于线程，进程有自己的独立内存空间，不能使用全局变量在进程间传递数据。
实际项目需求中，常常存在密集计算、或实时性任务，进程之间有时需要传递大量数据，如图片、大对象等，传递数据如果通过文件序列化、或网络接口来进行，难以满足实时性要求，采用redis，或者kaffka, rabbitMQ 之第3方消息队列包，又使系统复杂化了。
Python multiprocessing 模块本身就提供了消息机制、同步机制、共享内存等各种非常高效的进程间通信方式。
了解并掌握 python 进程间通信的各类方式的使用，以及安全机制，可以帮助大幅提升程序运行性能。
进程间通信的主要方式总结如下
关于进程间通信的内存安全
内存安全意味着，多进程间可能会因同抢，意外销毁等原因造成共享变量异常。
Multiprocessing 模块提供的Queue, Pipe, Lock, Event 对象，都已实现了进程间通信安全机制。
采用共享内存方式通信，需要在代码中自已来跟踪、销毁这些共享内存变量，否则可能会出同抢、未正常销毁等。造成系统异常。 除非开发者很清楚共享内存使用特点，否则不建议直接使用此共享内存，而是通过Manager管理器来使用共享内存。
内存管理器Manager
Multiprocessing提供了内存管理器Manager类，可统一解决进程通信的内存安全问题，可以将各种共享数据加入管理器，包括 list, dict, Queue, Lock, Event, Shared Memory 等，由其统一跟踪与销毁。
类似于1上简单的socket通道，双端均可收发消息。
Pipe 对象的构建方法：
parent_conn, child_conn = Pipe(duplex=True/False)
参数说明
duplex=True, 管道为双向通信
duplex=False, 管道为单向通信，只有child_conn可以发消息，parent_conn只能接收。
示例代码：
from multiprocessing import Process, Pipe def myfunction(conn): conn.send(['hi!! I am Python']) conn.close() if __name__ == '__main__': parent_conn, child_conn = Pipe() p = Process(target=myfunction, args=(child_conn,)) p.start() print (parent_conn.recv() ) p.join()
Multiprocessing 的Queue 类，是在python queue 3.0版本上修改的， 可以很容易实现生产者 – 消息者间传递数据，而且Multiprocessing的Queue 模块实现了lock安全机制。
Queue模块共提供了3种类型的队列。
(1) FIFO queue , 先进先出，
class queue.Queue(maxsize=0)
(2) LIFO queue, 后进先出， 实际上就是堆栈
class queue.LifoQueue(maxsize=0)
(3) 带优先级队列， 优先级最低entry value lowest 先了列
class queue.PriorityQueue(maxsize=0)
Multiprocessing.Queue类的主要方法：
| method | Description |
|---|---|
| queue.qsize() | 返回队列长度 |
| queue.full() | 队列满，返回 True, 否则返回False |
| queue.empty() | 队列空，返回 True, 否则返回False |
| queue.put(item) | 将数据写入队列 |
| queue.get() | 将数据抛出队列 ， |
| queue.put_nowait(item), queue.get_nowait() | 无等待写入或抛出 |
说明：
put(), get() 是阻塞方法， 而put_notwait(), get_nowait()是非阻塞方法。
Multiprocessing 的Queue类没有提供Task_done, join方法
Queue模块的其它队列类：
(1) SimpleQueue
简洁版的FIFO队列, 适事简单场景使用
(2) JoinableQueue子类
Python 3.5 后新增的 Queue的子类，拥有 task_done(), join() 方法
task_done()表示，最近读出的1个任务已经完成。
join()阻塞队列，直到queue中的所有任务都已完成。
producer – consumer 场景，使用Queue的示例
import multiprocessing
def producer(numbers, q):
for x in numbers:
if x % 2 == 0:
if q.full():
print("queue is full")
break
q.put(x)
print(f"put {x} in queue by producer")
return None
def consumer(q):
while not q.empty():
print(f"take data {q.get()} from queue by consumer")
return None
if __name__ == "__main__":
# 设置1个queue对象，最大长度为5
qu = multiprocessing.Queue(maxsize=5,)
# 创建producer子进程，把queue做为其中1个参数传给它，该进程负责写
p5 = multiprocessing.Process(
name="producer-1",
target=producer,
args=([random.randint(1, 100) for i in range(0, 10)], qu)
)
p5.start()
p5.join()
#创建consumer子进程，把queue做为1个参数传给它，该进程中队列中读
p6 = multiprocessing.Process(
name="consumer-1",
target=consumer,
args=(qu,)
)
p6.start()
p6.join()
print(qu.qsize())Multiprocessing也提供了与threading 类似的同步锁机制，确保某个时刻只有1个子进程可以访问某个资源或执行某项任务, 以避免同抢。
例如：多个子进程同时访问数据库表时，如果没有同步锁，用户A修改1条数据后，还未提交，此时，用户B也进行了修改，可以预见，用户A提交的将是B个修改的数据。
添加了同步锁，可以确保同时只有1个子进程能够进行写入数据库与提交操作。
如下面的示例，同时只有1个进程可以执行打印操作。
from multiprocessing import Process, Lock
def f(l, i):
l.acquire()
try:
print('hello world', i)
finally:
l.release()
if __name__ == '__main__':
lock = Lock()
for num in range(10):
Process(target=f, args=(lock, num)).start()Event 机制的工作原理：
1个event 对象实例管理着1个 flag标记, 可以用set()方法将其置为true, 用clear()方法将其置为false, 使用wait()将阻塞当前子进程，直至flag被置为true.
这样由1个进程通过event flag 就可以控制、协调各子进程运行。
Event object的使用方法：
1）主函数： 创建1个event 对象， flag = multiprocessing.Event() , 做为参数传给各子进程
2) 子进程A: 不受event影响,通过event 控制其它进程的运行
o 先clear()，将event 置为False, 占用运行权.
o 完成工作后，用set()把flag置为True。
3) 子进程B, C: 受event 影响
o 设置 wait() 状态，暂停运行
o 直到flag重新变为True，恢复运行
主要方法：
set(), clear()设置 True/False,
wait() 使进程等待，直到flag被改为true.
is_set()， Return True if and only if the internal flag is true.
验证进程间通信 – Event
import multiprocessing
import time
import random
def joo_a(q, ev):
print("subprocess joo_a start")
if not ev.is_set():
ev.wait()
q.put(random.randint(1, 100))
print("subprocess joo_a ended")
def joo_b(q, ev):
print("subprocess joo_b start")
ev.clear()
time.sleep(2)
q.put(random.randint(200, 300))
ev.set()
print("subprocess joo_b ended")
def main_event():
qu = multiprocessing.Queue()
ev = multiprocessing.Event()
sub_a = multiprocessing.Process(target=joo_a, args=(qu, ev))
sub_b = multiprocessing.Process(target=joo_b, args=(qu, ev,))
sub_a.start()
sub_b.start()
# ev.set()
sub_a.join()
sub_b.join()
while not qu.empty():
print(qu.get())
if __name__ == "__main__":
main_event()子进程之间共存内存变量，要用 multiprocessing.Value(), Array() 来定义变量。 实际上是ctypes 类型，由multiprocessing.sharedctypes模块提供相关功能
注意 使用 share memory 要考虑同抢等问题，释放等问题，需要手工实现。因此在使用共享变量时，建议使用Manager管程来管理这些共享变量。
def func(num):
num.value=10.78 #子进程改变数值的值，主进程跟着改变
if __name__=="__main__":
num = multiprocessing.Value("d", 10.0)
# d表示数值,主进程与子进程可共享这个变量。
p=multiprocessing.Process(target=func,args=(num,))
p.start()
p.join()
print(num.value)进程之间共享数据(数组型)：
import multiprocessing
def func(num):
num[2]=9999 #子进程改变数组，主进程跟着改变
if __name__=="__main__":
num=multiprocessing.Array("i",[1,2,3,4,5])
p=multiprocessing.Process(target=func,args=(num,))
p.start()
p.join()
print(num[:])如果进程间需要共享对象数据，或共享内容，数据较大，multiprocessing 提供了SharedMemory类来实现进程间实时通信，不需要通过发消息，读写磁盘文件来实现，速度更快。
注意：直接使用SharedMemory 存在着同抢、泄露隐患，应通过SharedMemory Manager 管程类来使用, 以确保内存安全。
创建共享内存区：
multiprocessing.shared_memory.SharedMemory(name=none, create=False, size=0)
方法：
父进程创建shared_memory 后，子进程可以使用它，当不再需要后，使用close(), 删除使用unlink()方法
相关属性：
获取内存区内容： shm.buf
获取内存区名称： shm.name
获取内存区字节数: shm.size
示例：
