并发编程之多进程

1、multiprocessing模块介绍

　　python中的多线程没法利用多核优点，若是想要充分地使用多核CPU的资源（os.cpu\_count\(\)查看），在python中大部分状况须要使用多进程。

　　Python提供了multiprocessing模块用来开启子进程，并在子进程中执行咱们定制的任务（好比函数），该模块与多线程模块threading的编程接口相似。multiprocessing模块的功能众多：支持子进程、通讯和共享数据、执行不一样形式的同步，提供了Process、Queue、Pipe、Lock等组件。

　　须要再次强调的一点是：与线程不一样，进程没有任何共享状态，进程修改的数据，改动仅限于该进程内。

2、Process类的介绍

一、建立进程的类

Process([group [, target [, name [, args [, kwargs]]]]])，由该类实例化获得的对象，可用来开启一个子进程

强调：
1. 须要使用关键字的方式来指定参数
2. args指定的为传给target函数的位置参数，是一个元组形式，必须有逗号

二、参数介绍

　　group参数未使用，值始终未None

　　target表示调用对象，即子进程要执行的任务

　　args表示调用对象的位置参数元组，args=(1, 2, 'egon')

　　kwargs表示调用对象的字典，kwargs={'name':'egon', 'age':18}

　　name为子进程的名称

三、方法介绍

p.start()：启动进程，并调用该子进程中的p.run() 
p.run():进程启动时运行的方法，正是它去调用target指定的函数，咱们自定义类的类中必定要实现该方法  

p.terminate():强制终止进程p，不会进行任何清理操做，若是p建立了子进程，该子进程就成了僵尸进程，使用该方法须要特别当心这种状况。若是p还保存了一个锁那么也将不会被释放，进而致使死锁
p.is_alive():若是p仍然运行，返回True

p.join([timeout]):主线程等待p终止（强调：是主线程处于等的状态，而p是处于运行的状态）。timeout是可选的超时时间。

四、属性介绍

p.daemon：默认值为False，若是设为True，表明p为后台运行的守护进程，当p的父进程终止时，p也随之终止，而且设定为True后，p不能建立本身的新进程，必须在p.start()以前设置

p.name:进程的名称

p.pid：进程的pid

p.exitcode:进程在运行时为None、若是为–N，表示被信号N结束(了解便可)

p.authkey:进程的身份验证键,默认是由os.urandom()随机生成的32字符的字符串。这个键的用途是为涉及网络链接的底层进程间通讯提供安全性，这类链接只有在具备相同的身份验证键时才能成功（了解便可

3、Process类的使用

　　注意：在windows中Process()必须放到# if __name__ == '__main__':下：

　　这是因为Windows没有fork，多处理模式启动一个新的Python进程并导入调用模块。若是在导入时调用Process()，那么这个将启动无限继承的新进程（或直到机器耗尽资源）。

　　这对隐藏对Process()内部调用的缘由，使用if __name__ == '__main__'，这个if语句中的语句将不会在导入时被调用。

一、建立并开启子进程的两种方式

from multiprocessing import Process
import time

def task(name):
    print("%s is running" % name)
    time.sleep(3)
    print("%s is done" % name)

if __name__ == '__main__':
    p = Process(target=task, args=('子进程1',))  # 获得对象
    # Process(target=task, kwargs={'name': "子进程1"})

    p.start()   # 给操做系统发送启动信号
    print("主")
"""
主
子进程1 is running
子进程1 is done
"""

方式一：经过multiprocessing模块开启子进程

　　上述方法中，是有父子关系，初始状态和父亲是同样的，可是运行状态彻底无关。

from multiprocessing import Process
import time
class MyProcess(Process):
    def __init__(self, name):
        super().__init__()
        self.name = name

    def run(self):  # run()是固定形式，p.start本质是调用的绑定的run方法
        print('%s is running'%self.name)
        time.sleep(3)
        print("%s is done" % self.name)

if __name__ == '__main__':
    p = MyProcess('子进程')
    p.start()   # 给操做系统发送启动信号
    print('主')
"""
主
子进程 is running   # 间隔三秒
子进程 is done
"""

方式二：不用默认的multiprocessing模块，继承并订制本身的类

二、进程直接的内存空间是隔离的

from multiprocessing import Process
n=100 #在windows系统中应该把全局变量定义在if __name__ == '__main__'之上就能够了
def work():
    global n
    n=0
    print('子进程内: ',n)


if __name__ == '__main__':
    p=Process(target=work)
    p.start()
    print('主进程内: ',n)

三、练习题

　　一、思考开启进程的方式一和方式二各开启了几个进程？

答：两个方式都是开启了一个主进程和四个子进程。

　　二、进程之间的内存空间是共享的仍是隔离的？下述代码执行的结果？

答：进程之间的内存空间是隔离的，执行输出：“子进程内：0”

from multiprocessing import Process

n=100 #在windows系统中应该把全局变量定义在if __name__ == '__main__'之上就能够了

def work():
    global n
    n=0
    print('子进程内: ',n)


if __name__ == '__main__':
    p=Process(target=work)
    p.start()
"""输出：
子进程内:  0
"""

进程内存空间验证

　　三、基于多进程实现并发的套接字通讯？

# -*- coding:utf-8 -*-
__author__ = 'Qiushi Huang'

# 函数化改写
from socket import *
from multiprocessing import Process


def talk(conn):
    while True:
        try:
            data = conn.recv(1024)
            if not data:break
            conn.send(data.upper())
        except ConnectionResetError:
            break
    conn.close()


def server(ip, port):
    server = socket(AF_INET, SOCK_STREAM)
    server.setsockopt(SOL_SOCKET, SO_REUSEADDR,1)  # 连接循环
    server.bind((ip, port))
    server.listen(5)

    while True:
        conn, address = server.accept()   # 主进程一直建连接
        p = Process(target=talk, args=(conn,))  # 注意conn参数传入
        p.start()

    server.close()


if __name__ == '__main__':
    server('127.0.0.1', 9001)

服务端

# -*- coding:utf-8 -*-
__author__ = 'Qiushi Huang'

# 运行一次启动一个客户端进程
# 这种状况仍是存在问题：启动多个进程后，本机操做系统会崩溃掉
from socket import *


client = socket(AF_INET, SOCK_STREAM)
client.connect(('127.0.0.1', 9001))

while True:
    msg = input(">>: ").strip()
    if not msg:continue

    client.send(msg.encode('utf-8'))
    data = client.recv(1024)
    print(data.decode('utf-8'))

客户端

　　四、思考每来一个客户端，服务端就开启一个新的进程来服务它，这种实现方式有无问题？

答：以上是多进程实现并发套接字通讯的解决方案，这个解决方案是思路是：每来一个客户端，都在服务端开启一个进程，若是并发来一千一万个客户端，要开启成千上万个进程，而因为机器的硬件配置和性能限制是没法开启出那么多进程的。

 四、Process对象的join方法

from multiprocessing import Process
import time, os

def task():
    print('%s is running, parent id is <%s>' % (os.getpid(), os.getppid()))   # 进程和父进程查看方式
    time.sleep(3)
    print("%s is done, parent id is <%s>" % (os.getpid(), os.getppid()))

if __name__ == '__main__':
    p = Process(target=task, )
    p.start()

    p.join()   # 优先运行子进程，主进程卡在原地
    print('主进程', os.getpid(), 'pycharm ID', os.getppid())
    print(p.pid)  # 子进程运行完，变为僵尸进程，主进程仍可以查到子进程的pid，当主进程结束后，全部僵尸子进程将被丢掉。
"""
828 is running, parent id is <827>
828 is done, parent id is <827>
主进程 827 pycharm ID 504
828
"""

join方法：优先运行子进程，主进程卡在原地，子进程结束后，运行主进程后面的代码

from multiprocessing import Process
import time, os

def task(name ,n):
    print('%s is running' % name)
    time.sleep(n)

if __name__ == '__main__':
    start = time.time()
    p1 = Process(target=task, args=("子进程1",5))
    p2 = Process(target=task, args=("子进程2",3))
    p3 = Process(target=task, args=("子进程3",2))
    """
    进程开启顺序由操做系通通筹控制，顺序是不必定的
    主进程 1014 pycharm ID 504
    子进程2 is running
    子进程1 is running
    子进程3 is running
    """
    p1.start()
    p2.start()
    p3.start()
    # 再添加join函数前，主程序的执行输出次序是彻底随机的，须要加join()保证主程序等到在子进程以后执行完成
    p1.join()
    p2.join()
    p3.join()
    # 以上并不是串行执行，实际是并发执行，只是约束了主程序要等在子程序后结束
    # print('主进程', os.getpid(), 'pycharm ID', os.getppid())
    print("主进程", (time.time()-start))
"""
子进程1 is running
子进程2 is running
子进程3 is running
主进程 5.010260343551636   # 主程序只等了5秒，说明确实是并发执行
"""

并发执行，约束主程序要等在子程序后结束

from multiprocessing import Process
import time, os

def task(name ,n):
    print('%s is running' % name)
    time.sleep(n)

if __name__ == '__main__':
    start = time.time()
    p1 = Process(target=task, args=("子进程1",5))
    p2 = Process(target=task, args=("子进程2",3))
    p3 = Process(target=task, args=("子进程3",2))
    p_l = [p1, p2, p3]

    for p in p_l:
        p.start()

    for p in p_l:
        p.join()

    print("主进程", (time.time()-start))
"""
子进程1 is running
子进程2 is running
子进程3 is running
主进程 5.007940769195557
"""

上述并发执行简写

from multiprocessing import Process
import time, os

def task(name ,n):
    print('%s is running' % name)
    time.sleep(n)

if __name__ == '__main__':
    start = time.time()
    p1 = Process(target=task, args=("子进程1",5))
    p2 = Process(target=task, args=("子进程2",3))
    p3 = Process(target=task, args=("子进程3",2))
    # 串行执行
    p1.start()
    p1.join()
    p2.start()
    p2.join()
    p3.start()
    p3.join()

    print("主进程", (time.time()-start))
"""
子进程1 is running
子进程2 is running
子进程3 is running
主进程 10.019965887069702
"""

改写为串行执行

五、is_alive方法查看进程是否存活；terminate方法关闭进程

#进程对象的其余方法一:terminate,is_alive
from multiprocessing import Process
import time
import random

class Piao(Process):
    def __init__(self,name):
        self.name=name
        super().__init__()

    def run(self):
        print('%s is piaoing' %self.name)
        time.sleep(random.randrange(1,5))
        print('%s is piao end' %self.name)


p1=Piao('egon1')
p1.start()

p1.terminate()#关闭进程,不会当即关闭,因此is_alive马上查看的结果可能仍是存活
print(p1.is_alive()) #结果为True

print('开始')
print(p1.is_alive()) #结果为False

terminate与is_alive

六、查看进程对象属性pid及ppid

# -*- coding:utf-8 -*-
__author__ = 'Qiushi Huang'

from multiprocessing import Process
import time,os

def task():
    print('%s is running, parent id is <%s>' % (os.getpid(), os.getppid()))   # 进程和父进程查看方式
    time.sleep(3)
    print("%s is done, parent id is <%s>" % (os.getpid(), os.getppid()))

if __name__ == '__main__':
    # p = Process(target=task, args=('子进程1',))   # 报错提示去掉参数TypeError: task() takes 0 positional arguments but 1 was given
    p = Process(target=task, )
    p.start()   # 给操做系统发送一个信号
    print('主进程', os.getpid(), 'pycharm ID', os.getppid())
"""
主进程 713 pycharm ID 504
714 is running, parent id is <713>
714 is done, parent id is <713>
"""

进程和父进程查看方法

七、查看进程对象属性name

from multiprocessing import Process
import time
import random

def task(name):
    print('%s is piaoing' %name)
    time.sleep(random.randrange(1,5))
    print('%s is piao end' %name)

if __name__ == '__main__':
    p1=Process(target=task,args=('egon',),name='子进程1') #能够用关键参数来指定进程名
    p1.start()

    print(p1.name,p1.pid,)

name与pid

八、僵尸进程和孤儿进程

　　僵尸进程和孤儿进程详解：点击进入

　　参考博客：http://www.cnblogs.com/Anker/p/3271773.html

4、守护进程

主进程建立守护进程

　　其一：守护进程会在主进程代码执行结束后就终止

　　其二：守护进程内没法再开启子进程,不然抛出异常：AssertionError: daemonic processes are not allowed to have children

注意：进程之间是互相独立的，主进程代码运行结束，守护进程随即终止

from multiprocessing import Process
import time

def task(name):
    print("%s is running" % name)
    time.sleep(2)


if __name__ == '__main__':
    p = Process(target=task, args=('子进程', ))
    p.daemon=True    # 守护进程必定要在进程开启前设置
    p.start()

    print("主进程")
"""
主进程    ————》子进程还没开始就已经结束了
"""

守护进程必定要在进程开启前设置

# 验证守护进程内部能再开子进程——》守护进程再开子进程会形成问题：会形成一堆孤儿
from multiprocessing import Process
import time

def task(name):
    print("%s is running" % name)
    time.sleep(2)
    p = Process(target=time.sleep, args=(3, ))
    p.start()

if __name__ == '__main__':
    p = Process(target=task, args=('子进程', ))
    p.daemon=True    # 守护进程必定要在进程开启前设置
    p.start()

    p.join()   # 让主进程等待子进程结束

    print("主进程")
"""
AssertionError: daemonic processes are not allowed to have children
"""

验证守护进程内部不能再开子进程

练习：思考下列代码的执行结果有可能有哪些状况？为何？

from multiprocessing import Process

import time
def foo():
    print(123)
    time.sleep(1)
    print("end123")

def bar():
    print(456)
    time.sleep(3)
    print("end456")

if __name__ == '__main__':
    p1=Process(target=foo)
    p2=Process(target=bar)

    p1.daemon=True   # 主进程代码执行完毕后，守护进程死
    p1.start()
    p2.start()
    print("main-------")
"""
main-------
456
end456
"""
# 另外一种状况是机器性能特别强，在执行到main----以前，已经启动子进程p1了，会造成输出：
"""
123
main-------
456
end456
"""

比较主进程和守护进程执行顺序

5、互斥锁（进程同步）

　　进程之间数据不共享,可是共享同一套文件系统,因此访问同一个文件,或同一个打印终端,是没有问题的,而共享带来的是竞争，竞争带来的结果就是错乱，以下

part1：多个进程共享同一打印终端

from multiprocessing import Process
import time

def task(name):
    print('%s 第一次' % name)
    time.sleep(1)
    print('%s 第二次' % name)
    time.sleep(1)
    print('%s 第三次' % name)


if __name__ == '__main__':
    for i in range(3):
        p = Process(target=task, args=('进程%s' % i, ))
        p.start()
"""
进程0 第一次
进程1 第一次
进程2 第一次
进程0 第二次
进程1 第二次
进程2 第二次
进程0 第三次
进程1 第三次
进程2 第三次
"""

并发运行,效率高,但竞争同一打印终端，谁抢到了谁打印

from multiprocessing import Process, Lock
import time

def task(name, mutex):
    mutex.acquire()   # 上锁，哪一个进程抢到锁，就一直给他运行
    print('%s 第一次' % name)
    time.sleep(1)
    print('%s 第二次' % name)
    time.sleep(1)
    print('%s 第三次' % name)
    mutex.release()   # 解锁


if __name__ == '__main__':
    mutex = Lock()   # 只实例化一次，并传给子进程，要保证全部进程用同一把锁
    for i in range(3):
        p = Process(target=task, args=('进程%s' % i, mutex))  # 传递给子进程的锁
        p.start()
"""
进程0 第一次
进程0 第二次
进程0 第三次
进程1 第一次
进程1 第二次
进程1 第三次
进程2 第一次
进程2 第二次
进程2 第三次
"""

加锁：由并发变成了串行，下降效率保证数据安全不错乱

　　互斥锁的意思就是互相排斥，若是把多个进程比喻为多我的，互斥锁的工做原理就是多我的都要去争抢同一个资源：卫生间，一我的抢到卫生间后上一把锁，其余人都要等着，等到这个完成任务后释放锁，其余人才有可能有一个抢到......因此互斥锁的原理，就是把并发改为穿行，下降了效率，但保证了数据安全不错乱。

part2：多个进程共享同一文件

　　文件当数据库，模拟抢票，文件db的内容为：{"count": 1}

from multiprocessing import Process
import json, time

def search(name):
    """查票"""
    time.sleep(1)   # 模拟网络延迟,并发去看票数
    dic = json.load(open('db.txt', 'r', encoding='utf-8'))
    print('<%s> 查看到剩余票数[%s]' %(name, dic['count']))


def get(name):
    """买票"""
    time.sleep(1)   # 模拟网络延迟
    dic = json.load(open('db.txt', 'r', encoding='utf-8'))
    if dic['count'] > 0:    # 确认有票
        dic['count'] -= 1
        time.sleep(3)
        # 写入文件，即购票成功，这个必须是基于其余人购票的结果，由并发改成串行
        json.dump(dic, open('db.txt', 'w', encoding='utf-8'))
        print('<%s> 购票成功' % name)


def task(name):
    search(name)
    get(name)


if __name__ == '__main__':
    for i in range(10):
        p = Process(target=task, args=('路人%s' % i, ))
        p.start()
"""
<路人1> 查看到剩余票数[1]
<路人0> 查看到剩余票数[1]
<路人2> 查看到剩余票数[1]
<路人4> 查看到剩余票数[1]
<路人3> 查看到剩余票数[1]
<路人5> 查看到剩余票数[1]
<路人6> 查看到剩余票数[1]
<路人7> 查看到剩余票数[1]
<路人8> 查看到剩余票数[1]
<路人9> 查看到剩余票数[1]
<路人1> 购票成功
<路人0> 购票成功
<路人2> 购票成功
<路人3> 购票成功
<路人4> 购票成功
<路人5> 购票成功
<路人6> 购票成功
<路人8> 购票成功
<路人7> 购票成功
<路人9> 购票成功
"""

并发运行，效率高，竞争写入同一文件，数据写入错乱

　　db.txt里只有一张票，因为并发卖出了10次。须要把购票行为改成串行，只有第一我的能够买到票

from multiprocessing import Process, Lock
import json, time

def search(name):
    """查票"""
    time.sleep(1)   # 模拟网络延迟,并发去看票数
    dic = json.load(open('db.txt', 'r', encoding='utf-8'))
    print('<%s> 查看到剩余票数[%s]' %(name, dic['count']))


def get(name):
    """买票"""
    time.sleep(1)   # 模拟网络延迟
    dic = json.load(open('db.txt', 'r', encoding='utf-8'))
    if dic['count'] > 0:    # 确认有票
        dic['count'] -= 1
        time.sleep(3)
        # 写入文件，即购票成功，这个必须是基于其余人购票的结果，由并发改成串行
        json.dump(dic, open('db.txt', 'w', encoding='utf-8'))
        print('<%s> 购票成功' % name)


def task(name, mutex):
    search(name)   # 查票并发执行，人人均可以看到票
    mutex.acquire()    # 上锁
    get(name)      # 购票改成串行，其余人都必须等着
    mutex.release()    # 解锁


if __name__ == '__main__':
    mutex = Lock()
    for i in range(10):
        p = Process(target=task, args=('路人%s' % i, mutex))
        p.start()
"""
<路人0> 查看到剩余票数[1]
<路人1> 查看到剩余票数[1]
<路人2> 查看到剩余票数[1]
<路人3> 查看到剩余票数[1]
<路人4> 查看到剩余票数[1]
<路人5> 查看到剩余票数[1]
<路人6> 查看到剩余票数[1]
<路人7> 查看到剩余票数[1]
<路人8> 查看到剩余票数[1]
<路人9> 查看到剩余票数[1]
<路人0> 购票成功
"""

加锁：购票行为由并发变成了串行，牺牲运行效率，保证数据安全

总结：

　　加锁能够保证多个进程修改同一块数据时，同一时间只能有一个任务能够进行修改，即串行的修改，没错，速度是慢了，但牺牲了速度却保证了数据安全。

虽然能够用文件共享数据实现进程间通讯，但问题是：

　　1.效率低（共享数据基于文件，而文件是硬盘上的数据）

　　2.须要本身加锁处理

　　所以咱们最好找寻一种解决方案可以兼顾：一、效率高（多个进程共享一块内存的数据）二、帮咱们处理好锁问题。这就是mutiprocessing模块为咱们提供的基于消息的IPC通讯机制：队列和管道。

　　1 队列和管道都是将数据存放于内存中

　　2 队列又是基于（管道+锁）实现的，可让咱们从复杂的锁问题中解脱出来

　　咱们应该尽可能避免使用共享数据，尽量使用消息传递和队列，避免处理复杂的同步和锁问题，并且在进程数目增多时，每每能够得到更好的可获展性。

6、队列

　　进程彼此之间互相隔离，要实现进程间通讯（IPC），multiprocessing模块支持两种形式：队列和管道，这两种方式都是使用消息传递的。

一、建立队列的类（底层就是以管道和锁定的方式实现）

Queue([maxsize]):建立共享的进程队列，Queue是多进程安全的队列，可使用Queue实现多进程之间的数据传递。

二、参数介绍

　　maxsize是队列中容许最大项数，省略则无大小限制。

　　但须要明确：

　　　　一、队列内存放的是消息而非大数据；

　　　　二、队列占用的是内存空间，于是maxsize即便是无大小限制也受限于内存大小。

三、主要方法介绍 

q.put方法用以插入数据到队列中，put方法还有两个可选参数：blocked和timeout。若是blocked为True（默认值），而且timeout为正值，该方法会阻塞timeout指定的时间，直到该队列有剩余的空间。若是超时，会抛出Queue.Full异常。若是blocked为False，但该Queue已满，会当即抛出Queue.Full异常。
q.get方法能够从队列读取而且删除一个元素。一样，get方法有两个可选参数：blocked和timeout。若是blocked为True（默认值），而且timeout为正值，那么在等待时间内没有取到任何元素，会抛出Queue.Empty异常。若是blocked为False，有两种状况存在，若是Queue有一个值可用，则当即返回该值，不然，若是队列为空，则当即抛出Queue.Empty异常.
 
q.get_nowait():同q.get(False)
q.put_nowait():同q.put(False)

q.empty():调用此方法时q为空则返回True，该结果不可靠，好比在返回True的过程当中，若是队列中又加入了项目。
q.full()：调用此方法时q已满则返回True，该结果不可靠，好比在返回True的过程当中，若是队列中的项目被取走。
q.qsize():返回队列中目前项目的正确数量，结果也不可靠，理由同q.empty()和q.full()同样

四、其余方法

q.cancel_join_thread():不会在进程退出时自动链接后台线程。能够防止join_thread()方法阻塞

q.close():关闭队列，防止队列中加入更多数据。调用此方法，后台线程将继续写入那些已经入队列但还没有写入的数据，但将在此方法完成时立刻关闭。
若是q被垃圾收集，将调用此方法。关闭队列不会在队列使用者中产生任何类型的数据结束信号或异常。
例如，若是某个使用者正在被阻塞在get()操做上，关闭生产者中的队列不会致使get()方法返回错误。

q.join_thread()：链接队列的后台线程。此方法用于在调用q.close()方法以后，等待全部队列项被消耗。
默认状况下，此方法由不是q的原始建立者的全部进程调用。调用q.cancel_join_thread方法能够禁止这种行为

五、队列的使用

from multiprocessing import Queue

# 队列中应该放消息，不该该放大文件大数据
# 队列能够不设置长度，可是队列是受制于内存大小的，不可能无限存放
q = Queue(3)  # 指定队列大小
q.put('hello')
q.put({'a': 1})
q.put([3,3,3,])

print(q.full())   # 查看队列是否满了  # True
# q.put(123)    # 队列满了再往里面放时，被锁住，只能在原地卡着。

print(q.get())
print(q.get())
print(q.get())
print(q.empty())   # 判断队列是否所有清空  # True

# print(q.get())   # 因为已经空了，程序也卡在原处
"""
True
hello
{'a': 1}
[3, 3, 3]
True
"""

队列应用 

7、生产者消费者模型　

　　在并发编程中使用生产者和消费者模式可以解决绝大多数并发问题。该模式经过平衡生产线程和消费线程的工做能力来提升程序的总体处理数据的速度。

为何要使用生产者消费者模型？

　　生产者指的是生产数据的任务，消费者指的是处理数据的任务。
　　在并发编程中，若是生产者处理速度很快，而消费者处理速度很慢，那么生产者就必须等待消费者处理完，才能继续生产数据。一样的道理，若是消费者的处理能力大于生产者，那么消费者就必须等待生产者。为了解决这个问题因而引入了生产者和消费者模式

什么是生产者和消费者模式？

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

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

from multiprocessing import Process,Queue
import time,random,os
def consumer(q,name):
    while True:
        res=q.get()
        time.sleep(random.randint(1,3))
        print('\033[43m%s 吃 %s\033[0m' %(name,res))

def producer(q,name,food):
    for i in range(3):
        time.sleep(random.randint(1,3))
        res='%s%s' %(food,i)
        q.put(res)
        print('\033[45m%s 生产了 %s\033[0m' %(name,res))

if __name__ == '__main__':
    q=Queue()
    #生产者们:即厨师们
    p1=Process(target=producer,args=(q,'egon','包子'))

    #消费者们:即吃货们
    c1=Process(target=consumer,args=(q,'alex'))

    #开始
    p1.start()
    c1.start()
    print('主')
"""
执行结果
主
egon 生产了 包子0
egon 生产了 包子1
alex 吃 包子0
alex 吃 包子1
egon 生产了 包子2
alex 吃 包子2
"""

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

生产者消费者模型总结：

一、生产者消费者模型何时用？

　　程序中有两类角色：一类负责生产数据（生产者）；一类负责处理数据（消费者）。

二、怎么叫生产者消费者模型？

　　引入队列解决生产者和消费者之间的耦合，这个并不依赖进程，这实际是介绍了一种编程方式。

　　若是使用了Queue，说明生产者、消费者、queue都在一台机器，这属于集中式架构，严重影响性能和稳定性。

　　基于网络通讯的消息队列：Rabbitmq。

三、生产者消费者模型好处？

　　引入生产者消费者模型为了解决的问题是：

　　（1）平衡生产者与消费者之间的工做能力，从而提升程序总体处理数据的速度。

　　（2）生产者消费者模型实现类程序的解耦合。

四、如何实现生产者消费者模型

　　如何实现：生产者<--->队列<--->消费者

 

　　此时的问题是主进程永远不会结束，缘由是：生产者p在生产完后就结束了，可是消费者c在取空了q以后，则一直处于死循环中且卡在q.get()这一步。

　　解决方法：让生产者在生产完毕后，往队列中再发一个结束信号，这样消费者在接收到结束信号后就能够break出死循环。

from multiprocessing import Process, Queue
import time

def producer(q):
    for i in range(10):
        res = '包子%s' % i
        time.sleep(0.5)   # 模拟生产时间
        print('生产者生产了%s' % res)

        q.put(res)  # 放入队列中


def consumer(q):
    while True:    # 一直从队列取一旦取空了，会加一把锁，程序卡在这里
        res = q.get()   # 取队列中数据
        if res is None:break
        time.sleep(1)    # 模拟消费时间
        print('消费者消费了%s' % res)

if __name__ == '__main__':
    # 容器
    q = Queue()
    # 生产者
    p1 = Process(target=producer, args=(q, ))
    # 消费者
    c1 = Process(target=consumer, args=(q, ))

    p1.start()
    c1.start()

    p1.join()   # 保证生产者都执行完，主进程才执行完
    q.put(None)  # 往队列里放入None,给消费者判断
    print("主进程")
"""
生产者生产了包子0
生产者生产了包子1
生产者生产了包子2
消费者消费了包子0
生产者生产了包子3
消费者消费了包子1
生产者生产了包子4
生产者生产了包子5
消费者消费了包子2
生产者生产了包子6
生产者生产了包子7
消费者消费了包子3
生产者生产了包子8
生产者生产了包子9
主进程
消费者消费了包子4
消费者消费了包子5
消费者消费了包子6
消费者消费了包子7
消费者消费了包子8
消费者消费了包子9
"""

生产者在生产完毕后发送结束信号None

　　注意：结束信号None，不必定要由生产者发，主进程里一样能够发，但主进程须要等生产者结束后才应该发送该信号

　　有多个生产者和消费者时，解决方案是有几个消费者就发送几回结束信号：

from multiprocessing import Process, Queue
import time

def producer(q):
    for i in range(10):
        res = '包子%s' % i
        time.sleep(0.5)   # 模拟生产时间
        print('生产者生产了%s' % res)

        q.put(res)  # 放入队列中

        # q.put(None)  # 这种方式会将消费者提早中止


def consumer(q):
    while True:    # 一直从队列取一旦取空了，会加一把锁，程序卡在这里
        res = q.get()   # 取队列中数据
        if res is None:break
        time.sleep(1)    # 模拟消费时间
        print('消费者消费了%s' % res)

if __name__ == '__main__':
    # 容器
    q = Queue()
    # 生产者
    p1 = Process(target=producer, args=(q, ))
    p2 = Process(target=producer, args=(q,))
    p3 = Process(target=producer, args=(q,))
    # 消费者
    c1 = Process(target=consumer, args=(q, ))
    c2 = Process(target=consumer, args=(q,))

    p1.start()
    p2.start()
    p3.start()
    c1.start()
    c2.start()

    p1.join()   # 保证生产者都执行完，主进程才执行完
    p2.join()
    p3.join()
    # 跟在正常信号后面，必须保证全部的生产者都生产结束
    q.put(None)  # 往队列里放入None,给消费者判断
    q.put(None)  # 有几个消费者就须要几个结束信号
    print("主进程")

有几个消费者就须要发送几回结束信号

　　上面这种解决方案很是Low,可使用JoinableQueue来解决这个问题。

8、JoinableQueue

　　JoinableQueue([maxsize])：这就像是一个Queue对象，但队列容许项目的使用者通知生成者项目已经被成功处理。通知进程是使用共享的信号和条件变量来实现的。

参数介绍

　　maxsize是队列中容许最大项数，省略则无大小限制。

方法介绍

JoinableQueue的实例p除了与Queue对象相同的方法以外还具备：
q.task_done()：使用者使用此方法发出信号，表示q.get()的返回项目已经被处理。若是调用此方法的次数大于从队列中删除项目的数量，将引起ValueError异常
q.join():生产者调用此方法进行阻塞，直到队列中全部的项目均被处理。阻塞将持续到队列中的每一个项目均调用q.task_done（）方法为止

# -*- coding:utf-8 -*-
__author__ = 'Qiushi Huang'

# JoinableQueue的用法和queue相似，只是这个queue是能够被join的
from multiprocessing import Process, JoinableQueue
import time

def producer(q):
    for i in range(2):
        res = '包子%s' % i
        time.sleep(0.5)   # 模拟生产时间
        print('生产者生产了%s' % res)

        q.put(res)  # 放入队列中
    q.join()  # 生产者生产完等队列把数据都取完


def consumer(q):
    while True:
        res = q.get()   # 取队列中数据
        if res is None:break
        time.sleep(1)    # 模拟消费时间
        print('消费者消费了%s' % res)
        q.task_done()   # 提供的接口，消费者告诉生产者取走了一个数据


if __name__ == '__main__':
    # 容器
    q = JoinableQueue()
    q.join()    # 等队列执行完（等队列取完）

    # 生产者
    p1 = Process(target=producer, args=(q,))
    p2 = Process(target=producer, args=(q,))
    p3 = Process(target=producer, args=(q,))
    # 消费者
    c1 = Process(target=consumer, args=(q,))
    c2 = Process(target=consumer, args=(q,))
    # 主进程执行完以后，守护进程也终止，所以把消费者设置为守护进程
    c1.daemon=True
    c2.daemon=True

    p1.start()
    p2.start()
    p3.start()
    c1.start()
    c2.start()

    p1.join()
    p2.join()
    p3.join()

    print("主进程")
"""
生产者生产了包子0
生产者生产了包子0
生产者生产了包子0
生产者生产了包子1
生产者生产了包子1
生产者生产了包子1
消费者消费了包子0
消费者消费了包子0
消费者消费了包子0
消费者消费了包子1
消费者消费了包子1
消费者消费了包子1
主进程
"""

基于JoinableQueue实现生产者消费者模型

9、管道

　　管道是进程间通讯（IPC）的第二种方式。

#建立管道的类：
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异常。

介绍

from multiprocessing import Process,Pipe

import time,os
def consumer(p,name):
    left,right=p
    left.close()
    while True:
        try:
            baozi=right.recv()
            print('%s 收到包子:%s' %(name,baozi))
        except EOFError:
            right.close()
            break
def producer(seq,p):
    left,right=p
    right.close()
    for i in seq:
        left.send(i)
        # time.sleep(1)
    else:
        left.close()
if __name__ == '__main__':
    left,right=Pipe()

    c1=Process(target=consumer,args=((left,right),'c1'))
    c1.start()


    seq=(i for i in range(10))
    producer(seq,(left,right))

    right.close()
    left.close()

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

基于管道实现进程间通讯（与队列的方式是相似的，队列就是管道加锁实现的）

基于管道实现进程间通讯（与队列的方式是相似的，队列就是管道加锁实现的）

注意：生产者和消费者都没有使用管道的某个端点，就应该将其关闭，如在生产者中关闭管道的右端，在消费者中关闭管道的左端。若是忘记执行这些步骤，程序可能再消费者中的recv()操做上挂起。管道是由操做系统进行引用计数的,必须在全部进程中关闭管道后才能生产EOFError异常。所以在生产者中关闭管道不会有任何效果，付费消费者中也关闭了相同的管道端点。

from multiprocessing import Process,Pipe

import time,os
def adder(p,name):
    server,client=p
    client.close()
    while True:
        try:
            x,y=server.recv()
        except EOFError:
            server.close()
            break
        res=x+y
        server.send(res)
    print('server done')
if __name__ == '__main__':
    server,client=Pipe()

    c1=Process(target=adder,args=((server,client),'c1'))
    c1.start()

    server.close()

    client.send((10,20))
    print(client.recv())
    client.close()

    c1.join()
    print('主进程')
#注意：send()和recv()方法使用pickle模块对对象进行序列化。

管道能够用于双向通讯，利用一般在客户端/服务器中使用的请求／响应模型或远程过程调用，就可使用管道编写与进程交互的程序

管道能够用于双向通讯，利用一般在客户端/服务器中使用的请求／响应模型或远程过程调用，就可使用管道编写与进程交互的程序

10、共享数据

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

即使是使用线程，推荐作法也是将程序设计为大量独立的线程集合

经过消息队列交换数据。这样极大地减小了对使用锁定和其余同步手段的需求，

还能够扩展到分布式系统中

进程间通讯应该尽可能避免使用本节所讲的共享数据的方式

进程间数据时独立的，能够借助于队列或管道实现通讯，两者都是基于消息传递的。

虽然进程间数据独立，但能够经过Manager实现数据共享，事实上Manager的功能远不止于此。

from multiprocessing import Manager,Process,Lock
import os
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)
        #{'count': 94}

进程之间操做共享的数据

 

11、信号量

 

12、事件

 

十3、进程池