python学习Day37--生产者消费者模型

一、回顾

1、并发与并行：

　　并发：在同一个时间短内多个任务同时进行。

　　并行：在用一个时间点上多个任务同时进行。

2、进程的三大基本状态：

　　就绪状态：所有进程需要的资源都获取到了，除了CPU。

　　执行状态：获取到了所有资源包括CPU，进程处于运行状态。

　　阻塞状态：进程停止不再运行，放弃了CPU，进程此时处于内存中。

3、什么叫进程？

　　正在运行的程序；由代码段，数据段，PCB（进程控制块）组成；

　　进程是资源分配的基本单位；

4、进程之间能不能直接通信？

　　正常情况下，多进程之间是无法进行通信的。因为每一个进程都有自己独立的内存空间。

5、锁机制

　　为了多进程通信时，保护数据的安全性。

　　一把锁醅一把钥匙。

　　l = Lock( )

　　l.acquire( )

　　l.release( )

6、信号量

　　一把锁配多把钥匙。

　　sem = Semaphore(num)

　　num代表的是几把钥匙。

7、事件

　　e = Event( )

　　e.is_set( )　　返回一个bool值。

　　e.wait( )　　   阻塞和非阻塞

　　e.set( )　　     把is_set的bool值变为True

　　e.clear( )　　  把is_set的bool变为False

二、生产者消费者模型

　　主要是为了解耦

　　借助队列来实现生产者消费者模型。

1、栈与队列介绍

　　栈：先进后出（First In Last Out 简称FILO）

　　队列：先进先出（First In First Out 简称FIFO），队列是安全的。

import queue # 不能进程多进程之前的数据传输（不用）
from multiprocessing import Queue # 用这个模块

（1）队列的主要内容

q = Queue(num)
# num：队列的最大长度
q.get() # 阻塞等待获取数据，如果有数据直接获取，如果没有数据，阻塞等待
q.put() # 阻塞，如果可以继续往队列中放数据，就直接放，不能放就阻塞等待

q.get_nowait() # 不阻塞，如果有数据直接获取，没有数据就报错
q.put_nowait() # 不阻塞，如果可以继续网队列中放数据，就直接放，不能就报错

（2）初始队列

# *******************练习**********************开始
from multiprocessing import Queue

q = Queue(3) # 长度为3

q.put(1)
q.put('abc')
q.put([4,5,6])
print(123)
# q.put('娃哈哈') # 阻塞，不报错
try:
    q.put_nowait('娃哈哈') # 阻塞，并报错
except:
    print("队列满了")
print(111)

print(q.get())
print(q.get())
print(q.get())
# print(q.get()) # 阻塞，等待
try:
    print(q.get_nowait())
except:
    print("队列已经空了")

# *******************练习**********************结束

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

# *******************队列实现生产者消费者模型**********************开始
from multiprocessing import Queue,Process
import time

def consumer(q,name):
    while 1:
        info = q.get_nowait()
        if info:
            print('%s拿走了%s' % (name,info))
        else:
            print("没有娃娃了，等下一批吧")
            break


# 消费者如何判断，生产者是没来得及生产，还是生产者不在生产了？

def producer(q,product):
    for i in range(20):
        info = product+'娃娃%s号' % str(i)
        q.put(info)
    q.put(None)

if __name__ == '__main__':
    q = Queue(10)
    p_pro = Process(target=producer,args=(q,'mini版本'))
    p_con = Process(target=consumer,args=(q,'xxx'))
    p_pro.start()
    p_con.start()
# *******************队列实现生产者消费者模型**********************结束

3、进阶生产者消费者模型

（1）进阶：将生产者生产结束的标识，放到父进程中。

# *******************进阶生产者消费者模型**********************开始
from multiprocessing import Queue,Process
import time

def consumer(q,name,color):
    while 1:
        info = q.get()
        if info:
            print('%s%s拿走了%s33[0m' % (color,name,info))
        else:
            print("没有娃娃了，等下一批吧")
            break

# 消费者如何判断，生产者是没来得及生产，还是生产者不在生产了？

def producer(q,product):
    for i in range(20):
        info = product+'娃娃%s号' % str(i)
        q.put(info)

if __name__ == '__main__':
    q = Queue(10)
    p_pro1 = Process(target=producer,args=(q,'mini版本'))
    p_pro2 = Process(target=producer,args=(q,'maxi版本'))
    p_pro3 = Process(target=producer,args=(q,'super版本'))
    p_con1 = Process(target=consumer,args=(q,'夏明','33[31m'))
    p_con2 = Process(target=consumer,args=(q,'小华','33[36m'))
    p_1 = [p_con1,p_con2,p_pro1,p_pro2,p_pro3]
    [i.start() for i in p_1]

    # 父进程如何感知生产者子进p_pro1程不在生产数据了？
    # 通过加入join来解决
    p_pro1.join()
    p_pro2.join()
    p_pro3.join()
    q.put(None) # 几个消费者就接收几个结束标识
    q.put(None)
# *******************进阶生产者消费者模型**********************结束

（2）新模块JoinableQueue（可连接队列）——改进生产者消费者模型

　　继承Queue，所有可以使用Queue里面的方法。

# *******************新模块joinableQueue**********************开始
from multiprocessing import Process,JoinableQueue

# q = JoinableQueue()

# q.join() # 用于生产者。等待q.task_done返回结果，通过返回结果，生产和就能获得消费者当前消费了多少个数据
# q.task_done() # 用于消费者，是指每消费队列中一个数据，就给join返回一个标识

from multiprocessing import Queue,Process
import time

def consumer(q,name,color):
    while 1:
        info = q.get()
        print('%s%s拿走了%s33[0m' % (color,name,info))
        q.task_done()

# 消费者如何判断，生产者是没来得及生产，还是生产者不在生产了？

def producer(q,product):
    for i in range(20):
        info = product+'娃娃%s号' % str(i)
        q.put(info)
    q.join() #记录了生产20个数据在队列中，此时会阻塞等待消费者消费完队列中所有数据

if __name__ == '__main__':
    q = JoinableQueue(10)
    p_pro1 = Process(target=producer,args=(q,'mini版本'))
    # p_pro2 = Process(target=producer,args=(q,'maxi版本'))
    # p_pro3 = Process(target=producer,args=(q,'super版本'))
    p_con1 = Process(target=consumer,args=(q,'夏明','33[31m'))
    # p_con2 = Process(target=consumer,args=(q,'小华','33[36m'))
    p_con1.daemon = True # 把消费者进程设为守护进程（这样做是：最后才能结束消费者进程，主进程结束，守护进程也就结束了）
    p_pro1.start()
    p_con1.start()
    p_pro1.join() # 主进程等待生产者进程结束

'''
    程序有3个进程，主进程和生产者进程和消费者进程。 当主进程执行到p_pro1.join()代码时，主进程会等待生产进程结束
    而生产进程中会等待消费者进程把所有数据消费完，生产者进程才结束。
    现在的状态就是  主进程等待生产者进程结束，生产者进程等待消费者消费完所有数据
    所以，把消费者设置为守护进程。  当主进程执行完，就代表生产进程已经结束，也就代表消费者进程已经把队列中数据消费完
    此时，主进程一旦结束，守护进程也就是消费者进程也就跟着结束。    整个程序也就能正常结束了。
'''
# *******************新模块joinableQueue**********************结束

4、管道

　　管道是不安全的。

（1）管道介绍、

# *******************管道**********************开始
from multiprocessing import Pipe

con1,con2 = Pipe()

con1.send('abc')
print(con2.recv())
con2.send('456')
print(con1.recv())
# *******************管道**********************结束

（2）多进程下的管道

# *******************多进程下的管道**********************开始
from multiprocessing import Pipe,Process

def func(con):
    con1,con2 = con
    con1.close()
    while 1:
        try:
            print(con2.recv())
        except EOFError:
            con2.close()
            break

if __name__ == '__main__':
    con1,con2 = Pipe()
    p = Process(target=func,args=((con1,con2),))
    p.start()
    con2.close()
    for i in range(10):
        con1.send('abc')
    con1.close()
# *******************多进程下的管道**********************结束

【注意】一个小问题：你知道的IPC有哪些？

　　管道、队列、（锁，信号量，事件）

5、进程之间的共享内存Manager

# *******************多进程之前的共享内存**********************开始
from multiprocessing import Process,Manager

def func(num):
    num[0] -= 1
    print("子进程中的num的值是",num) # [0, 2, 3]


if __name__ == '__main__':
    m = Manager()
    num = m.list([1,2,3])
    p = Process(target=func,args=(num,))
    p.start()
    p.join()
    print("父进程中的num的值是",num) # [0, 2, 3]
# *******************多进程之前的共享内存**********************结束

6、进程池

 　   进程池：一个池子，里边有固定数量的进程。这些进程抑制处于待命状态，一旦有任务来，马上就有进程去处理。

　　因为在实际业务中，任务量是有多又有少的，如果任务量特别多，不可能要开对应那么多的进程数；

　　开启那么多进程首先就需要消耗大量的时间让操作系统来管理它，其次还需要消耗大量时间让CPU帮你调度它。

　　进程池还会帮程序员去管理池中的进程。

进程池有三个方法

（1）map(func, iterable)

　　func：进程池中的进程执行的任务函数

　　iterable：可迭代对象，是把可迭代对象中的每个元素依次传给任务函数的参数

（2）apply(func, args=( )) ——同步的效率，也就是说池中的进程一个一个的去执行任务

　　func：进程池中的进程执行的任务函数

　　args：可迭代对象型的参数，是传给任务函数的参数

　　同步处理任务时，不需要close和join

（3）apply_async(func, args=( ),callback=None) ——异步的效率，也就是说池中的进程一次性都去执行任务

　　func：进程池中的进程执行的任务函数

　　args：可迭代对象型的参数，是传给任务函数的参数

　　callback：回调函数，就是说每当进程池中有进程处理完任务了，返回的结果可以交给回调函数，由回调函数进行进一步的处理，回调函数只有异步才有，同步没有的。

　　异步处理任务时，必须要加上close和join

# *******************进程池**********************开始
from multiprocessing import Pool,Process
import os
import time

def func(num):
    num += 1
    print(num)

if __name__ == '__main__':
    # p = Pool(os.cpu_count())
    # start = time.time()
    # p.map(func,[i for i in range(100)])
    # p.close() # 指不允许再向进程池中添加任务
    # p.join() # 等待进程池中所有进程执行完所有任务
    # print("进程池做任务的效率", time.time() - start)

    # p = Pool(os.cpu_count())
    # # start = time.time()
    # for i in range(100):
    #     res = p.apply(func,args=(i,)) # （进程池的同步处理）是指让进程池中的进程，同步的帮你做任务
    # print("进程池做任务的效率", time.time() - start)

    p = Pool(os.cpu_count())
    start = time.time()
    for i in range(100):
        p.apply_async(func, args=(i,)) # （进程池的异步处理）是指让进程池中的进程，异步的帮你做任务
    print("进程池做任务的效率", time.time() - start)

    start = time.time()
    p_1 = []
    for i in range(100):
        p1 = Process(target=func,args=(i,)) # （多线程的处理）
        p1.start()
        p_1.append(p1)
    [p1.join() for p1 in p_1]
    print("多进程做任务的效率",time.time()-start)
# *******************进程池**********************结束

回调函数的使用：

　　进程的任务函数的返回值，被当成回调函数的形参接收到，以此进行进一步的操作。

　　回调函数是由主进程调用的，而不是子进程，子进程只负责把结果传递给回调函数。

from multiprocessing import Pool
import requests
import time,os

def func(url):
    res = requests.get(url)
    print('子进程的pid:%s,父进程的pid：%s'%(os.getpid(),os.getppid()))
    # print(res.text)
    if res.status_code == 200:
        return url,res.text

def cal_back(sta):
    url,text = sta
    print('回调函数的pid', os.getpid())
    with open('a.txt','a',encoding='utf-8') as f:
        f.write(url + text)
    print('回调函数中！',url)

if __name__ == '__main__':
    p = Pool(5)
    l = ['https://www.baidu.com',
         'http://www.jd.com',
         'http://www.taobao.com',
         'http://www.mi.com',
         'http://www.cnblogs.com',
         'https://www.bilibili.com',
         ]
    print('主进程的pid',os.getpid())
    for i in l:
        p.apply_async(func, args=(i,),callback=cal_back) # 回调函数是由主进程调用的
        # 异步执行任务func，每有一个进程执行完任务后，在func中return一个结果，结果会自动的被callback指定的函数，当成形式参数来接收到
    p.close()
    p.join()