Manager模块 队列 管道 进程池

Manager模块

作用:  多进程共享变量.

Manager的字典类型：

• 如果value是简单类型，比如int，可以直接赋值给共享变量，并可以后续直接修改
• 如果value是复杂类型 ，比如list，dict，则必须先用临时变量做完所有修改后，最后一次性赋值给共享变量。

共享变量的另一个方法:Value

Manager简单用法:

from multiprocessing import Process,Manager


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

if __name__ == '__main__':
    m = Manager() 
    num = m.list([1,2,3]) #  将列表设置为共享变量.
    p = Process(target=func,args=(num,))　　# 实例化一个进程
    p.start()　　# 开启进程
    p.join()　　# 阻塞, 必须执行完上面的子进程才能继续执行主进程
    print('父进程中的num的值是',num) # 此处打印的num和子进程中的num是一样的.

---

队列(主要用于消费者生产者模型的解耦问题)

生产者消费者模型主要是为了解耦问题, 可借助队列来实现生产者消费者模型的解耦问题.

队列的特点:  先进先出(First In First Out    简称FIFO)

栈的特点(区别于队列): 先进后出(First in Lost Out    简称FILO)

两个进程间的数据是独立的，要进行数据传递的话可通过几个方法

Queue: 通过队列来进行进程间数据的传递

Pipe:  通过管道来进行进程间数据的传递

本次重点讲:

Queue: 通过队列来进行进程间数据的传递

队列是安全的, 可借助Queue解决生产者消费者模型.

Queue中的一些用法:

　　q = Queue(num)   实例化一个队列

　　q.get()  #  阻塞等待获取数据, 如果有数据直接获取, 如果没有数据, 阻塞等待.

　　q.put()  #  阻塞, 如果可以继续在队列中添加数据就继续添加, 如果不能继续添加, 就阻塞等待.

　　q.get_nowait()  # 不阻塞, 如果有数据就直接获取, 没有就报错.

　　qput_nowait()  #  不阻塞, 如果可以继续在队列中添加数据, 就直接添加, 不能添加就报错.

实例: 

实例一.

import multiprocessing import Queue


def get_1(q, name): # 消费者函数
    while 1:
        info = q.get()  # 获得生产者端传过来的标识.
        if info:
            print('%s拿了%s' % (name, info))
        else:
            break  # 当接收到None 指令后, 结束消费者模块.


def put_1(q):
    for i in range(10):  #  定义生产10个娃娃
        doll = '第%s个娃娃' % str(i)
        q.put() # 将产生的数据添加到队列中传给消费者.
    q.put(None) # 当玩具生产完玩具后, 传给消费者一个None, 使消费者进程结束.


if __name__ == '__main__':
    q = Queue()  # 初始化队列
    p = Process(target = get_1, args = (q, '消费者01')) # 实例化消费者进程
    p1 = Process(target = put_1, args = (q)) # 实例化生产者进程
    p1.join() # 阻塞, 等待p1进程结束后,再结束主进程.
    p.start()  #  开启消费者进程.
    p1.start() #  开启生产者进程.

实例二:

from multiprocessing import Process Queue

def get_1(q, name):  # 定义一个消费者模块
　　info = q.get() # 获得生产者产生的数据.
　　if info:
　　　　print('%s拿了%s' %(name, info))
　　else:
　　　　break
def put_1(q):  # 定义一个生产者模块
　　for i in range(20):  # 限定生产数据的数量
　　　　info = '第%s个娃娃' %str(i)
　　　　q.put(info)
if __name__ = '__main__':
　　q = Queue(4) # 实例化队列, 最多允许每次产生5个数据.
　　p = Process(target = get_1, args = (q, '大白'))  # 实例化消费者进程
　　p1 = Process(target = put_1, args = (q))  # 实例化生产者进程
　　p.start() # 开启消费者进程
　　p1.start()  # 开启生产者进程
　　p1.join() # 阻塞, 等待生产进程全部完以后, 主进程就能获得生产结束的标识
　　q.put(None) # 给消费者传输一个空的数据, 使消费者模块运行结束.

 实例三:(使用JoinableQueue实现队列)

PS:  JoinableQueue是继承的Queue, 所以可以受用Queue模块的方法.

　　 在JoinableQueue模块中, 

　　　　1, q.join()  用于生产者,   是等待q.task_done()的返回结果.通过返回结果,

　　　　    生产者就能获得消费者当前消费了多少个数据.

　　　　2, q.task_done  用于消费者, 是指消费者每次消费队列中的一个数据, 就给join()返回一个标识.

from multiprocessing import Process JoinableQueue

def gett(q, name): # 创建一个消费者模块
    while 1: # 无限循环接收娃娃
        q.get() # 接收来自生产者队列中的数据.
        print('%s拿到了%s' %(name, q))
        q.task_done() # 将每一次拿到娃娃的数据反馈给生产者

def putt(q):  # 创建一个生产者模块
    for i in range(20):  #  定义生产20个娃娃.
        f = '第%s号娃娃' % str(i) # 实例化娃娃, 将娃娃赋值给f
        q.put(f) # 调用JoinableQueue中的pur()函数, 将生产的娃娃添加到队列中
    q.join() # 调用JoinableQueue中的join()函数, 接收到来自消费者每次消费反馈的数据.

if __name__ == "__main__":
    q = JoinableQueue(4) # 初始化JoinableQueue, 使其一次最多生产5个娃娃
    p = Process(target=gett, args=(q, '（●─●）大白')) # 实例化一个消费者进程
    p1 = Process(target=putt, args=(q,)) # 实例化一个生产者进程
    p.daemon = Ture # 将消费者进程转变成守护进程, 使其随着主进程的结束而结束.
    p.start() # 开启消费者进程.
    p1.start() # 开启生产者进程.
    p1.join() # 主进程等待生产者进程执行结束后继续执行.


注释:
    # 此函数 有三个进程, 主进程和生产者进程,消费者进程.当主进程执行到p1.join()的时候,
    # 就会等待消费者进程运行结束后主进程再结束.

    # 当生产这进程运行到q.join()的时候, 会等待消费者把所有的消费标识都返回给生产者的join以后再结束.
    # 而消费者是主函数的守护进程, 所以当生产者接收到消费者消费完所有数据的标识后, 结束, 从而主函数结束.
    
    # 因为消费者进程为守护进程, 当主进程运行结束的时候, 消费者进程(守护进程)也会被强迫结束.
    # 因此, 三个进程全部结束.
    

管道:(了解)

    管道是不安全的.

from multiprocessing import Pipe
con1,con2 = Pipe()
管道是不安全的。
管道是用于多进程之间通信的一种方式。
如果在单进程中使用管道，那么就是con1收数据，就是con2发数据。
如果是con1发数据，就是con2收数据

如果在多进程中使用管道，那么就必须是父进程使用con1收，子进程就必须使用con2发
父进程使用con1发，子进程就必须使用con2收
父进程使用con2收，子进程就必须使用con1发
父进程使用con2发，子进程就必须使用con1收
在管道中有一个著名的错误叫做EOFError。是指，父进程中如果关闭了发送端，子进程还继续接收数据，那么就会引发EOFError。

单进程下实例:

from multiprocessing import Pipe

# con1,con2 = Pipe()
#
# con1.send('abc')
# print(con2.recv())
# con2.send(123)
# print(con1.recv())

多进程下实例:

from multiprocessing import Pipe,Process

def func(con):
    con1,con2 = con
    con1.close()# 子进程使用con2和父进程通信，所以
    while 1:
        try:
            print(con2.recv())#当主进程的con1发数据时，子进程要死循环的去接收。
        except EOFError:# 如果主进程的con1发完数据并关闭con1，子进程的con2继续接收时，就会报错，使用try的方式，获取错误
            con2.close()# 获取到错误，就是指子进程已经把管道中所有数据都接收完了，所以用这种方式去关闭管道
            break

if __name__ == '__main__':
    con1,con2 = Pipe()
    p = Process(target=func,args=((con1,con2),))
    p.start()
    con2.close()# 在父进程中，使用con1去和子进程通信，所以不需要con2，就提前关闭
    for i in range(10):# 生产数据
        con1.send(i)# 给子进程的con2发送数据
    con1.close()# 生产完数据，关闭父进程这一端的管道

4 进程池
进程池：一个池子，里边有固定数量的进程。这些进程一直处于待命状态，一旦有任务来，马上就有进程去处理。
因为在实际业务中，任务量是有多有少的，如果任务量特别的多，不可能要开对应那么多的进程数
开启那么多进程首先就需要消耗大量的时间让操作系统来为你管理它。其次还需要消耗大量时间让
cpu帮你调度它。
进程池还会帮程序员去管理池中的进程。
from multiprocessing import Pool
p = Pool(os.cpu_count() + 1)

进程池有三个方法： 

map(func,iterable) func：进程池中的进程执行的任务函数 iterable: 可迭代对象，是把可迭代对象中的每个元素依次传给任务函数当参数

apply(func,args=())： 同步的效率，也就是说池中的进程一个一个的去执行任务
func：进程池中的进程执行的任务函数
args: 可迭代对象型的参数，是传给任务函数的参数
同步处理任务时，不需要close和join
同步处理任务时，进程池中的所有进程是普通进程（主进程需要等待其执行结束）

apply_async(func,args=(),callback=None)： 异步的效率，也就是说池中的进程一次性都去执行任务
func：进程池中的进程执行的任务函数
args: 可迭代对象型的参数，是传给任务函数的参数
callback: 回调函数，就是说每当进程池中有进程处理完任务了，返回的结果可以交给回调函数，由回调函数进行进一步的处理,回调函数只有异步才有，同步是没有的
异步处理任务时，进程池中的所有进程是守护进程（主进程代码执行完毕守护进程就结束）
异步处理任务时，必须要加上close和join

回调函数的使用：
进程的任务函数的返回值，被当成回调函数的形参接收到，以此进行进一步的处理操作
回调函数是由主进程调用的，而不是子进程，子进程只负责把结果传递给回调函数

例子一(map)

# 注意:  map 处理的时候是异步处理

from multiprocessing import Pool
# 导入Pool模块

def func(num):  # 定义一个任务函数, 接收主进程传递过来的数据, 其参数名可以命名(尽量与传入时保持一致)
    num += 1  # 将的到的数据进行累加
    print(num)  # 将每次运行的结果打印在屏幕上.
    return num  # 将每次执行完的结果当做一个返回值.

if __name__ == '__main__':
    p = Pool(5) # 初始化一个进程池.  一次最多执行5个进程
    res = p.map(func,[i for i in range(100)]) # 将i 传递给func任务函数, 然后获得func执行完毕的返回值.
    p.close() # 关闭进程池, 防止其他数据进入使函数混乱.
    p.join()  # 等待进程池中的进程执行完毕 之后再继续执行主进程.
    print('主进程中map的返回值',res)  # 将map的到的返回值打印在屏幕上.

例二: 

进程池中的同步执行方法[apply(func,args=())]:

apply(func,args=())： 同步的效率，也就是说池中的进程一个一个的去执行任务
func：进程池中的进程执行的任务函数
args: 可迭代对象型的参数，是传给任务函数的参数
同步处理任务时，不需要close和join
同步处理任务时，进程池中的所有进程是普通进程（主进程需要等待其执行结束）

from multiprocessing import Pool

def func(i): # 定义一个任务函数, 必须要有参数
    i += 1
    return i

if __name__ == '__main__':
    q = Pool(5) # 实例化进程池, 由于此处为同步(单进程)执行, 所以定义再多的进程都智慧一个一个去执行.
    for i in range(100): # 开启10个进程, (一个一个开启)
        apply_1 = q.apply(func,args=(i,))  # 将每个i 传给func任务函数. 并接收func的返回值
        print(apply_1) # 打印返回的结果.

# 注意: 再同步处理时, 进程池中的所有进程都是普通进程.(主进程需要等待其执行结束) 不需要close, join.

实例三:

进程池中的异步:

apply_async(func,args=(),callback=None)： 异步的效率高于同步的效率，也就是说池中的进程一次性都去执行任务
func：进程池中的进程执行的任务函数
args: 可迭代对象型的参数，是传给任务函数的参数
callback: 回调函数，就是说每当进程池中有进程处理完任务了，返回的结果可以交给回调函数，由回调函数进行进一步的处理,回调函数只有异步才有，同步是没有的
异步处理任务时，进程池中的所有进程是守护进程（主进程代码执行完毕守护进程就结束）
异步处理任务时，必须要加上close和join

from multiprocessing import Pool

def func(i): # 定义的任务函数, 必须要有形参.
    i += 1 # 进行累加
    return i  # 返回i

if __name__ == '__main__':
    q = Pool(5) # 实例一个进程池, 最多一次开启5个进程.
    l = [] # 付那个一一个空列表, 方便后面打印结果使用(不会讲异步处理改成同步处理. 如果不打印的话就不需要这一步) 或者将打印的步骤放到任务函数中去.
    for i in range(100):  # 开启100个进程
        f = q.apply_async(func, args=(i,)) # 将i 传给func任务函数, 并接受返回结果
        l.append(f) # 将结果添加到l列表中
    print([i.get() for i in l]) # 打印结果到屏幕上  实际编程中看情况选择.
    q.close() # 关闭进程池
    q.join()  # 等待进程池中的进程执行结束.
# 在异步处理的时候, 任务函数会随着主进程的执行结束而结束, 从而, 所有的子进程都是守护进程.

# 注意:   在进程池中, 异步处理的效率远远大于同步处理的效率.

同步, 异步的效率对比:

from multiprocessing import Pool
import time

def func(i):
    i += 1
    return i

if __name__ == '__main__':
    q = Pool(5)
    time_1 = time.time()
    for i in range(100):
        f = q.apply(func, args=(i,))

    print('同步消耗的时间',time.time() - time_1)
    time_2 = time.time()
    for i in range(100):
        f1 = q.apply_async(func, args=(i,))
    q.close()
    q.join()
    print('异步消耗的时间',time.time() - time_2)

# 其一般情况下同步所消耗的时间要大于异步所消耗的时间.

进程池中, 回调函数的使用:

回调函数的使用：
进程的任务函数的返回值，被当成回调函数的形参接收到，以此进行进一步的处理操作
回调函数是由主进程调用的，而不是子进程，子进程只负责把结果传递给回调函数

from multiprocessing import Pool
import requests
import time, os

def func(i): # 定义一个获取网页的任务函数
    res = requests.get(i)  # 获取网页内容
    print('子进程的pid:%s,父进程的pid：%s'%(os.getpid(),os.getppid()))  # 将识别码打印到屏幕上
    if res.status_code == 200: # 当状态码等于200时, 表示链接网页成功
        return i, res.text  # 返回的内容以texe的格式返回.

def cal_back(a): # 使用回调函数的返回值进行操作的人物函数.
    i,text = a  # 解构网址  和 源码
    # print(a)  # 将返回的内容打印到屏幕上
    print('回调函数的pid', os.getpid()) # 将获得的pid(进程识别码) 打印到屏幕上
    with open('网站源码.txt', mode='a', encoding='utf-8') as f1:  #
        f1.write(i + text) # 将 网址 和 获得的源码 写入到 '网站源码.txt' 文件中
        f1.flush() # 刷新文件
        f1.close() # 关闭文件

if __name__ == '__main__':
    q = Pool(5) # 实例化进程池, 并且最多同时开启5个进程, 每当有几个进程结束,会再添加几个进程
    time_1 = time.time()
    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',
]   # l 中是要爬去源码的网站网址,

    print('父进程的pid：%s' % (os.getpid()))
    for i in l: # 遍历出每一个网址, 赋值给i
        f = q.apply_async(func, args=(i,), callback=cal_back)
        # 异步执行将每个网站网址i 传给 func {此函数中最多同时传递5个} 任务函数爬取网站源码,
        # 每有一个进程执行完任务后，在func中return一个结果，
        # 然后回调函数 callback 会自动将返回值传给cal_back 任务函数,当成形式参数来接收到

    q.close() # 关闭进程池
    q.join() # 等待进程池中的所有进程全部结束后再结束.
    print(time.time() - time_1) # 计算一共花费的时间.