python进程间的通信-队列/管道以及进程间的数据共享和进程池

1. 进程之间的通信
　　1)队列 *****
　　2)管道 ***
2  进程之间的数据共享 *
3  进程池 ***** 

进程间通信

IPC(Inter-Process Communication)

进程的概念:

创建共享的进程队列，Queue是多进程安全的队列，可以使用Queue实现多进程之间的数据传递。 

Queue([maxsize]) 
创建共享的进程队列。
参数 ：maxsize是队列中允许的最大项数。如果省略此参数，则无大小限制。
底层队列使用管道和锁定实现。

Queue([maxsize]) 
创建共享的进程队列。maxsize是队列中允许的最大项数。如果省略此参数，则无大小限制。底层队列使用管道和锁定实现。另外，还需要运行支持线程以便队列中的数据传输到底层管道中。 
Queue的实例q具有以下方法：

q.get( [ block [ ,timeout ] ] ) 
返回q中的一个项目。如果q为空，此方法将阻塞，直到队列中有项目可用为止。block用于控制阻塞行为，默认为True. 如果设置为False，将引发Queue.Empty异常（定义在Queue模块中）。timeout是可选超时时间，用在阻塞模式中。如果在制定的时间间隔内没有项目变为可用，将引发Queue.Empty异常。

q.get_nowait( ) 
同q.get(False)方法。

q.put(item [, block [,timeout ] ] ) 
将item放入队列。如果队列已满，此方法将阻塞至有空间可用为止。block控制阻塞行为，默认为True。如果设置为False，将引发Queue.Empty异常（定义在Queue库模块中）。timeout指定在阻塞模式中等待可用空间的时间长短。超时后将引发Queue.Full异常。

q.qsize() 
返回队列中目前项目的正确数量。此函数的结果并不可靠，因为在返回结果和在稍后程序中使用结果之间，队列中可能添加或删除了项目。在某些系统上，此方法可能引发NotImplementedError异常。


q.empty() 
如果调用此方法时 q为空，返回True。如果其他进程或线程正在往队列中添加项目，结果是不可靠的。也就是说，在返回和使用结果之间，队列中可能已经加入新的项目。

q.full() 
如果q已满，返回为True. 由于线程的存在，结果也可能是不可靠的（参考q.empty（）方法）。。

其他相关方法(不常用)

q.close() 
关闭队列，防止队列中加入更多数据。调用此方法时，后台线程将继续写入那些已入队列但尚未写入的数据，但将在此方法完成时马上关闭。如果q被垃圾收集，将自动调用此方法。关闭队列不会在队列使用者中生成任何类型的数据结束信号或异常。例如，如果某个使用者正被阻塞在get（）操作上，关闭生产者中的队列不会导致get（）方法返回错误。

q.cancel_join_thread() 
不会再进程退出时自动连接后台线程。这可以防止join_thread()方法阻塞。

q.join_thread() 
连接队列的后台线程。此方法用于在调用q.close()方法后，等待所有队列项被消耗。默认情况下，此方法由不是q的原始创建者的所有进程调用。调用q.cancel_join_thread()方法可以禁止这种行为。

"""
multiprocessing 模块支持进程间通信的两种主要形式:管道和队列
都是基于消息传递来实现的,都是队列接口


"""
from multiprocessing import Queue
q = Queue(3)

#put ,get ,put_nowait,get_nowait,full,empty
q.put(2)
q.put(3)
q.put(4)
q.put(5) #如果队列满了,程序就会停在这里,等待数据被别人取走,再将数据放入队列.
#另外如果队列中的数据一直不被取走,程序就会永远停在这里

try:
    q.put_nowait(3) #可以使用put_nowait(),如果队列满了不会阻塞,但是会因为队列满了而报错.
except:#因此我们可以用一个try语句来处理这个错误,这样程序不会一直阻塞下去,但是会丢掉这个信息.
    print("队列已经满了")
#因此,我们再放入数据之前,可以先看一下队列的状态,如果已经满了,就不继续put了.
print(q.full()) #满了
print(q.get())
print(q.get())
print(q.get())
# print(q.get()) # 同put方法一样，如果队列已经空了，那么继续取就会出现阻塞。
try:
    q.get_nowait(3) # 可以使用get_nowait，如果队列满了不会阻塞，但是会因为没取到值而报错。
except: # 因此我们可以用一个try语句来处理这个错误。这样程序不会一直阻塞下去。
    print('队列已经空了')

print(q.empty()) #空了

下面是一个简单多进程的示例:

from multiprocessing import Queue,Process


def fun(q):
    q.put(3)

if __name__ == '__main__':
    q = Queue()
    Process(target=fun,args=(q,)).start()
    print(q.get())

 队列的多进程的存和多进程的取:

import time,os
from multiprocessing import Process ,Queue,freeze_support

def q_input(q):
    info = str(os.getpid()) + "put" + str(time.asctime())
    q.put(info)
def q_output(q):
    ret = q.get()
    print(ret)
if __name__ == '__main__':
    freeze_support()
    q_lst1 = []
    q_lst2 = []
    q = Queue()
    for i in range(10):
        p = Process(target=q_input,args=(q,))
        q_lst1.append(p)
        p.start()
    for i in range(10):
        p = Process(target=q_output,args=(q,))
        q_lst2.append(p)
        p.start()
    [i.join() for i in q_lst1]
    [i.join() for i in q_lst2]
    print("my best")

生产者消费者模型

在并发编程中使用生产者和消费者模式能够解决绝大多数并发问题。该模式通过平衡生产线程和消费线程的工作能力来提高程序的整体处理数据的速度。

为什么要使用生产者和消费者模式

在线程世界里，生产者就是生产数据的线程，消费者就是消费数据的线程。在多线程开发当中，如果生产者处理速度很快，而消费者处理速度很慢，那么生产者就必须等待消费者处理完，才能继续生产数据。同样的道理，如果消费者的处理能力大于生产者，那么消费者就必须等待生产者。为了解决这个问题于是引入了生产者和消费者模式。

什么是生产者消费者模式

生产者消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。

基于队列实现生产者消费者模型

示例代码如下:

import time,random
from multiprocessing import Process,Queue
def consumer(i,q):
    while True:

        ret = q.get()
        if ret == "end":break
        print("%s吃了%s"%(i,ret))
        time.sleep(random.random())

def producer(name,q,num):
    for i in range(num):
        info =name +"的"+ "第"+str(i)+"个包子"
        q.put(info)
        time.sleep(random.randrange(0,3))
if __name__ == '__main__':
    q_lst = []
    q = Queue()
    for i in range(3):
        p1 = Process(target=producer,args=("leon"+str(i),q,10))
        p1.start()
        q_lst.append(p1)


    p = Process(target=consumer,args=("alex",q))
    p.start()

    [i.join() for i in q_lst]
    q.put("end")

JoinableQueue([maxsize]) 

创建可连接的共享进程队列。这就像是一个Queue对象，但队列允许项目的使用者通知生产者项目已经被成功处理。通知进程是使用共享的信号和条件变量来实现的。

JoinableQueue的实例p除了与Queue对象相同的方法之外，还具有以下方法：

q.task_done() 
使用者使用此方法发出信号，表示q.get()返回的项目已经被处理。如果调用此方法的次数大于从队列中删除的项目数量，将引发ValueError异常。

q.join() 
生产者将使用此方法进行阻塞，直到队列中所有项目均被处理。阻塞将持续到为队列中的每个项目均调用q.task_done()方法为止。 
下面的例子说明如何建立永远运行的进程，使用和处理队列上的项目。生产者将项目放入队列，并等待它们被处理。

示例代码:

from multiprocessing import JoinableQueue,Process
import time,random
def consumer(name,q):
    while 1:
        ret = q.get()
        print("%s正在吃%s"%(name,ret))
        time.sleep(random.random())
        q.task_done()

def producter(name,q,num):
    for i in range(num):
        info = name + "的" +"第"+str(i)+"个包子"
        q.put(info)
        time.sleep(random.randrange(0,3))
    q.join()
if __name__ == '__main__':
    q = JoinableQueue()
    q_lst =[]
    for i in range(3):
        p = Process(target=producter,args=("leon"+str(i),q,3))
        p.start()
        q_lst.append(p)
    p = Process(target=consumer,args=("alex",q))
    p.daemon = True
    p.start()
    [i.join() for i in q_lst]

管道

#创建管道的类：
Pipe([duplex]):在进程之间创建一条管道，并返回元组（conn1,conn2）,其中conn1，conn2表示管道两端的连接对象，强调一点：必须在产生Process对象之前产生管道
#参数介绍：
dumplex:默认管道是全双工的，如果将duplex射成False，conn1只能用于接收，conn2只能用于发送。
#主要方法：
    conn1.recv():接收conn2.send(obj)发送的对象。如果没有消息可接收，recv方法会一直阻塞。如果连接的另外一端已经关闭，那么recv方法会抛出EOFError。
    conn1.send(obj):通过连接发送对象。obj是与序列化兼容的任意对象
 #其他方法：
conn1.close():关闭连接。如果conn1被垃圾回收，将自动调用此方法
conn1.fileno():返回连接使用的整数文件描述符
conn1.poll([timeout]):如果连接上的数据可用，返回True。timeout指定等待的最长时限。如果省略此参数，方法将立即返回结果。如果将timeout射成None，操作将无限期地等待数据到达。
 
conn1.recv_bytes([maxlength]):接收c.send_bytes()方法发送的一条完整的字节消息。maxlength指定要接收的最大字节数。如果进入的消息，超过了这个最大值，将引发IOError异常，并且在连接上无法进行进一步读取。如果连接的另外一端已经关闭，再也不存在任何数据，将引发EOFError异常。
conn.send_bytes(buffer [, offset [, size]])：通过连接发送字节数据缓冲区，buffer是支持缓冲区接口的任意对象，offset是缓冲区中的字节偏移量，而size是要发送字节数。结果数据以单条消息的形式发出，然后调用c.recv_bytes()函数进行接收    
 
conn1.recv_bytes_into(buffer [, offset]):接收一条完整的字节消息，并把它保存在buffer对象中，该对象支持可写入的缓冲区接口（即bytearray对象或类似的对象）。offset指定缓冲区中放置消息处的字节位移。返回值是收到的字节数。如果消息长度大于可用的缓冲区空间，将引发BufferTooShort异常。

# 队列就是基于管道实现的
# 队列 数据安全的
# 管道 数据不安全的
# 队列 = 管道 + 锁

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

简单的管道示例:

from multiprocessing import Pipe,Process

def fun(left,right):
    left.close()
    while 1 :
        try:
            print(right.recv())
        except EOFError:
            break



if __name__ == '__main__':
    left,right = Pipe()
    p = Process(target=fun,args=(left,right))
    p.start()
    right.close()
    left.send("hao")
    left.close()

from multiprocessing import Pipe,Process

def fun(left,right):
    left.close()  #不写close将不会引发EOFError
    while 1 :
        try:
            print(right.recv())
        except EOFError:
            print(666)
            break



if __name__ == '__main__':
    left,right = Pipe()
    p = Process(target=fun,args=(left,right))
    p.start()
    right.close()
    left.send("hao")
    left.close()

 基于管道实现的生产者和消费者模型:

from multiprocessing import Pipe,Process
def consumer(p,name):
    left,right = p
    left.close()
    while 1 :
        try:
            ret = right.recv()

            print("%s吃了%s" % (name, ret))
        except EOFError:
            print(123)
            break


def producer(p,name,num):
    left,right= p

    for i in range(num):
        info = name + "的"+"第"+str(i)+"包子"
        left.send(info)
    left.close()
    right.close()


if __name__ == '__main__':
    left,right = Pipe()
    p = Process(target=consumer,args=((left,right),"alex"))
    p.start()
    p = Process(target=producer,args=((left,right),"leon",10))
    p.start()
    left.close()

进程之间的数据共享

展望未来，基于消息传递的并发编程是大势所趋

即便是使用线程，推荐做法也是将程序设计为大量独立的线程集合，通过消息队列交换数据。

这样极大地减少了对使用锁定和其他同步手段的需求，还可以扩展到分布式系统中。

但进程间应该尽量避免通信，即便需要通信，也应该选择进程安全的工具来避免加锁带来的问题。

以后我们会尝试使用数据库来解决现在进程之间的数据共享问题。

Manager模块介绍

进程间数据是独立的，可以借助于队列或管道实现通信，二者都是基于消息传递的
虽然进程间数据独立，但可以通过Manager实现数据共享，事实上Manager的功能远不止于此

A manager object returned by Manager() controls a server process which holds Python objects and allows other processes to manipulate them using proxies.

A manager returned by Manager() will support types list, dict, Namespace, Lock, RLock, Semaphore, BoundedSemaphore, Condition, Event, Barrier, Queue, Value and Array.

示例如下:

from multiprocessing import Manager,Process,Lock
def work(d,lock):
    with lock: #不加锁而操作共享的数据,肯定会出现数据错乱
        d['count']-=1

if __name__ == '__main__':
    lock=Lock()
    with Manager() as m:
        dic=m.dict({'count':100})
        p_l=[]
        for i in range(100):
            p=Process(target=work,args=(dic,lock))
            p_l.append(p)
            p.start()
        for p in p_l:
            p.join()
        print(dic)