37、进程之间的通信

1、生产者消费者模型Queue：队列（管道+锁，双向通信）

　　　　为了解决数据供需不平衡的情况（即生产者生产的多或者消费者消费的多造成的失衡），有一种解决的方法：子进程生产数据，子进程处理数据。

生产者消费者模型：

　　1、消费者要处理多少数据是不确定的

　　2、所以只能用while循环来处理数据，但是while循环无法结束

　　3、需要生产者发送信号

　　4、有多少个消费者，就需要发送多少个信号

　　5、但是发送的信号数量需要根据生产者和消费者的数量进行计算，所以非常不方便。　　　

import time
import random
from multiprocessing import Process
from multiprocessing import Queue

def producer(q,food):
    for i in range(5):
        q.put('%s-%s'%(food,i))
        print('生产了%s'%(food))
        time.sleep(random.random())
    q.put(None)
    q.put(None)
    q.put(None)

def consumer(q,name):
    while True:
        food=q.get()
        if food == None:break
        print('%s吃了%s'%(name,food))


if __name__ =='__main__':
    q=Queue()
    p1=Process(target=producer,args=(q,'骨头'))
    p1.start()
    p2=Process(target=producer,args=(q,'泔水'))
    p2.start()
    p3 = Process(target=consumer, args=(q, 'alex'))
    p3.start()
    p4 = Process(target=consumer, args=(q, 'egon'))
    p4.start()
    p5 = Process(target=consumer, args=(q, 'jin'))
    p5.start()
    

2、joinableQueue：顺序为：生产者生产的数据全部被消费------->生产者进程结束------->主进程执行代码结束—-->消费者守护进程结束。

　　put和get的一个计数机制，每次get数据之后发送task_done，put端接收到计数-1，直到计数为0就能感知到。

import time
import random
from multiprocessing import Process
from multiprocessing import JoinableQueue
# put   q.join
# get  处理数据 task_done 消费完了
def producer(q,food):
    for i in range(5):
        q.put('%s-%s'%(food,i))
        print('生产了%s'%food)
        # time.sleep(random.randint(1,3))
        time.sleep(random.random())
    q.join()  # 等待 消费者 把所有的数据都处理完

def consumer(q,name):
    while True:
        food = q.get()   # 生产者不生产还是生产的慢
        print('%s 吃了 %s'%(name,food))
        q.task_done()

if __name__ == '__main__':
    q = JoinableQueue()
    p1 = Process(target=producer,args=(q,'泔水'))
    p1.start()
    p2 = Process(target=producer, args=(q, '骨头'))
    p2.start()
    c1 = Process(target=consumer, args=(q, 'alex'))
    c1.daemon = True
    c1.start()
    c2 = Process(target=consumer, args=(q, 'egon'))
    c2.daemon = True
    c2.start()
    c3 = Process(target=consumer, args=(q, 'jin'))
    c3.daemon = True
    c3.start()

    p1.join()  # 等待p1执行完毕
    p2.join()  # 等待p2执行完毕

3、ipc(inter-process-commnication内部进程交流)——

　　pipe(管道）：支持双向通信，数据不安全。

　　首先来看管道是怎么用的

　　

from multiprocessing import  Pipe
p1,p2=Pipe()
p1.send('hello')        #一个发送，一个接收
print(p2.recv())
print('————————————————————————')
p2.send('hihihi')
p2.close()              #关闭p2发送端
print(p1.recv())
#print(p1.recv())     #放开的话，继续接收，会报错，报错类型：EOFError

管道在多进程中的作用：主进程中发送到管道，然后在子进程中接收。

应该特别注意管道端点的正确管理问题。如果是生产者或消费者中都没有使用管道的某个端点，就应将它关闭。这也说明了为何在生产者中关闭了管道的输出端，在消费者中关闭管道的输入端。如果忘记执行这些步骤，程序可能在消费者中的recv（）操作上挂起。管道是由操作系统进行引用计数的，必须在所有进程中关闭管道后才能生成EOFError异常。因此，在生产者中关闭管道不会有任何效果，除非消费者也关闭了相同的管道端点。 

　　

from multiprocessing import Process
from multiprocessing import Pipe,Lock
def func(p):
    foo,son=p
    foo.close()             #只用son，就把foo给关了
    while True:
        try:
            print(son.recv())
        except EOFError:        #异常处理
            break

if __name__ =='__main__':
    foo,son=Pipe()
    p=Process(target=func,args=((foo,son),))
    p.start()
    son.close()                #只用foo,把son给关了
    foo.send('hello')
    foo.send('hello')
    foo.send('hello')
    foo.send('heo')
    foo.close()

pipe实现消费者生产者模型：

　　

　　

from multiprocessing import Process,Pipe,Lock

def consumer(p,name,lock):
    produce, consume=p
    produce.close()
    while True:
        # lock.acquire()
        baozi=consume.recv()
        # lock.release()
        if baozi:
            print('%s 收到包子:%s' %(name,baozi))
        else:
            consume.close()
            break

def producer(p,n):
    produce, consume=p
    consume.close()
    for i in range(n):
        produce.send(i)
    produce.send(None)
    produce.send(None)
    produce.close()

if __name__ == '__main__':
    produce,consume=Pipe()
    lock = Lock()
    c1=Process(target=consumer,args=((produce,consume),'c1',lock))
    c2=Process(target=consumer,args=((produce,consume),'c2',lock))
    p1=Process(target=producer,args=((produce,consume),10))
    c1.start()
    c2.start()
    p1.start()

    produce.close()
    consume.close()

    c1.join()
    c2.join()
    p1.join()
    print('主进程')

 4、Manager：是一个类，提供了可以进行数据共享的一个机制，提供了很多数据类型dict 、list、pipe(这些并不提供数据安全的支持）

from multiprocessing import Process,Manager,Lock
def work(dict,lock):
    lock.acquire()
    dict['count']-=1
    lock.release()

if __name__ =='__main__':
    lock=Lock()
    m=Manager()        #要用Manager定义dict
    dict=m.dict({'count':100})
    l=[]
    for i in range(100):
        p=Process(target=work,args=(dict,lock))
        p.start()
        l.append(p)
    [p.join() for p in l]
    print(dict)

 5、进程池：类似与有一个池子（工作间），把进程（员工）放进去，进行工作。接下来是发任务，然后循环对任务进行工作。（进程池这个功能已经被实现了）。

import os
import time
import random
from multiprocessing import Pool
from multiprocessing import Process
def func(i):
    i+=1

if __name__=='__main__':
    # p=Pool(5)                #1、带#的是另一种方法：用进程池
    # start=time.time()
    # p.map(func,range(1000))   #相当于target=func,args=next(iterable)
    # p.close()                 #是不允许再向进程池中添加任务
    # p.join()
    # print(time.time()-start)
    p = Process()                 #2、起进程的办法，太占内存，费时间：起进程的时候要分配资源，分配内存，耗费时间
    start=time.time()
    l=[]
    for i in range(100):
        p=Process(target=func,args=(i,))
        p.start()
        l.append(p)
    [i.join() for i in l]
    print(time.time()-start)