Python自学成才之路 进程间通信

程序在创建子进程时，会完全复制一份主进程的环境，包括变量，函数，类等。所以在子进程中使用的变量，函数，类和主进程之间隔离的，子进程之间也是隔离的。
看下面这个案例：

from multiprocessing import ProcessAGE = 1def hello():print('hello')def greet(names):global AGEAGE += 1names.append('jack')print('======子进程代码=======')print('AGE = %s, id = %s'%(AGE, id(AGE)))print('names = %s , id = %s'%(names, id(names)))print('hello id = %s'%(id(hello)))print('======子进程代码=======')if __name__ == '__main__':names = ['peter']p = Process(target=greet, args=(names,))p.start()p.join()print('第二次执行子进程')p2 = Process(target=greet, args=(names,))p2.start()p2.join()print('======父进程代码=======')print('AGE = %s, id = %s'%(AGE, id(AGE)))print('names = %s , id = %s'%(names, id(names)))print('hello id = %s' % (id(hello)))print('======父进程代码=======')
输出：
======子进程代码=======
AGE = 2, id = 140736619595456
names = ['peter', 'jack'] , id = 2672604171264
hello id = 2672609140448
======子进程代码=======
第二次执行子进程
======子进程代码=======
AGE = 2, id = 140736619595456
names = ['peter', 'jack'] , id = 1613780831232
hello id = 1613785865952
======子进程代码=======
======父进程代码=======
AGE = 1, id = 140736619595424
names = ['peter'] , id = 2006331253248
hello id = 2006329121232
======父进程代码=======

从案例中可以看到，进程把所有的变量，函数都新复制了一份，即使是全局变量进程之间也是隔离的。所以进程之间要想共享数据需要使用进程间的通信，进程间通行有两种方式，第一种是管道（Pipe），第二种是队列。

Pipe

1. pipe常用于两个进程，两个进程分别位于管道的两端
2. Pipe方法返回管道的两端conn1和conn2，Pipe有一个duplex参数，默认为True，即全双工模式，若为false，conn1只能接收消息，conn2只能发送消息。

class MyProcess(Process):def __init__(self, threadname, conn):super().__init__()self.threadname = threadnameself.conn = conndef run(self):self.conn.send('我是' + self.threadname)print(self.threadname + '-接受消息：' + self.conn.recv())self.conn.close()if __name__ == '__main__':# 建立管道，拿到管道的两端，双工通信方式，两端都可以收发消息conn1, conn2 = Pipe()myProcess1 = MyProcess('thread1', conn1)myProcess2 = MyProcess('thread2', conn2)myProcess1.start()myProcess2.start()
输出：
thread2-接受消息：我是thread1
thread1-接受消息：我是thread2

Queue
Queue的一些常用方法的：
Queue(n)：初始化一个消息队列，并指定这个队列中最多能够容纳多少条消息。
put(obj,[block[,timeout]])：推入一条消息到这个队列中。默认是阻塞的，也就是说如果这个消息队列中已经满了，那么会会一直等待，将这个消息添加到消息队列中。timeout可以指定这个阻塞最长的时间，如果超过这个时间还是满的，就会抛出异常。
put_nowait() ：非阻塞的推入一条消息，如果这个队列已经满了，那么会立马抛出异常。
qsize()：获取这个消息队列消息的数量。
full()：判断这个消息队列是否满了。
empty()：判断这个消息队列是否空了。
get([block[,timeout]])：获取队列中的一条消息，然后将其从队列中移除，block默认为True。如果设置block为False，那么如果没值，会立马抛出异常。timeout指定如果多久没有获取到值后会抛出异常。

看个栗子：

from multiprocessing import Queue# 初始化一个Queue对象，最多只能存放三条消息
q = Queue(3)# 存放第一条消息
q.put('m1')
# 存放第二条消息
q.put('m2')# 判断这个队列中是否已经满了
print(q.full())# 存放第三条消息
q.put('m3')# 判断这个队列中的消息是否已经满了
print(q.full())# 因为如果消息队列已经满了，那么再put进去的时候就会报错
try:# q.put('m4', block=False)# 队列满了立即抛出异常# q.put('m4',block=True, timeout=2) # 可以被阻塞，等待时间超过两秒抛出异常q.put_nowait('m4')
except:print('消息队列已经满了，现有消息数量：%s' % q.qsize())print(q.get())输出：
False
True
消息队列已经满了，现有消息数量：3
m1

利用queue作为进程间通信来实现生产者和消费者

from multiprocessing import Process, Queue
import osdef write(q):for x in ['m1', 'm2', 'm3']:q.put(x)print('子进程已经存放了消息%s, id : %s' % (x, os.getpid()))def read(q):while True:try:msg = q.get(block=True, timeout=3)print('子进程已经读出了消息%s, id : %s' % (msg, os.getpid()))except:print('所有消息已经读完了')breakif __name__ == '__main__':q = Queue()pw = Process(target=write, args=(q,))pr = Process(target=read, args=(q,))pw.start()pr.start()pw.join()
输出：
子进程已经存放了消息m1, id : 28740
子进程已经存放了消息m2, id : 28740
子进程已经读出了消息m1, id : 17936
子进程已经读出了消息m2, id : 17936
所有消息已经读完了

进程池间通信
进程池间的通信使用Manager().Queue()，不能使用Queue（会报错，Queue objects should only be shared between processes through inheritance），Manager().Queue()和Queue的使用方法是一样的。

from multiprocessing import Pool, Manager
import osdef write(q):for x in ['m1', 'm2']:q.put(x)print('子进程已经存放了消息%s, id : %s' % (x, os.getpid()))def read(q):while True:try:msg = q.get(block=True, timeout=3)print('子进程已经读出了消息%s, id : %s' % (msg, os.getpid()))except:print('所有消息已经读完了')breakif __name__ == '__main__':q = Manager().Queue()pool = Pool(2)pool.apply(write, args=(q,))pool.apply(read, args=(q,))pool.close()pool.join()
输出：
子进程已经存放了消息m1, id : 3052
子进程已经存放了消息m2, id : 3052
子进程已经读出了消息m1, id : 22556
子进程已经读出了消息m2, id : 22556
所有消息已经读完了

做一下小总结：Python进程间数据是不共享的，所有的函数，变量，类都会被重新复制一份，要想让进程间可以共享数据，需要用到进程通信技术。比如pipe，和queue。Pipe常用于两个进程间的两端通信，实际用得较少。通常都是通过queue来实现进程间通信，进程池间通行是通过Manager.Queue。

本人是做大数据开发的，在微信上开了个个人号，会经常在上面分享一些学习心得，原创文章都会首发到公众号上，感兴趣的盆友可以关注下哦！
备注：微信公众号搜索‘大数据入坑指南