37 生产者消费者模型（JoinableQueue）、管道(Pipe了解)、进程之间的共享内存（Manager模块）、进程池

2023-12-20 23:19:58

课程回顾：
    并发：在同一个时间段内多个任务同时进行
    并行：在同一个事件点上多个任务同时进行
    进程的三大基本状态：
      就绪状态：所有进程需要的资源都获取到了，除了CPU
      执行状态：获取到了所有资源包括CPU，进程处于运行状态
      阻塞状态：进程停滞不再运行，放弃了CPU，进程此时处于内存里
    什么叫进程？
      正在运行的程序。
      由代码段，数据段，PCB（进程控制块）
    进程是资源分配的基本单位。

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


    锁。为了多进程通信时，保护数据的安全性
       一把锁配一把钥匙
       l = Lock（）
       l.acquire()
       l.release()
    信号量。
       一把锁配多把钥匙
       sem = Semaphore（num）
       num代表的是几把钥匙
    事件。
       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  # 不能进行多进程之间的数据传输
    （1）from multiprocessing import Queue   借助Queue解决生产者消费者模型
       队列是安全的。

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

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

放：

37 生产者消费者模型（JoinableQueue）、管道(Pipe了解)、进程之间的共享内存（Manager模块）、进程池

from multiprocessing import Queue

q = Queue(3)

q.put(1)
q.put(‘abc‘)
q.put([4, 5, 6])
print(123)
q.put(‘汉子‘)      # 队列满了，无法放入，阻塞

print(456)

简单例子

上面的例子阻塞了，如果用的是put_nowait 直接报错，加个try:

from multiprocessing import Queue

q = Queue(3)

q.put(1)
q.put(‘abc‘)
q.put([4, 5, 6])
print(123)
try:
    q.put_nowait(‘汉子‘)
except:
    print(‘队列满了‘)
print(456)

加个try

取：

from multiprocessing import Queue

q = Queue(3)

q.put(1)
q.put(‘abc‘)
q.put([4, 5, 6])

print(q.get())
print(q.get())
print(q.get())
print(q.get())    # 阻塞等待

无数据的时候，再取数据就阻塞等待了

我们如果不想被阻塞了话。更改最后一行

try:
    print(q.get_nowait())
except:
    print(‘队列空了‘)

还是用try ,非阻塞取值

# 消费者如何判断，生产者是没来得及生产数据，还是生产者不再生产数据了？
# 如果你尝试用get_nowait() + try 的方式去尝试获得生产者不再生产数据，此时是有问题的。
用一个数据尾部加一个标识

 1 from multiprocessing import Queue,Process
 2 import time
 3 
 4 def consumer(q,name):
 5     while 1:
 6         info = q.get()
 7         if info:
 8             print(‘%s 拿走了%s‘%(name,info))
 9         else:    # 当消费者获得队列中数据时，如果获得的是None，就是获得到了生产者不再生产数据的标识
10             break# 此时消费者结束即可
11 
12 # 消费者如何判断，生产者是没来得及生产数据，还是生产者不再生产数据了？
13 # 如果你尝试用get_nowait() + try 的方式去尝试获得生产者不再生产数据，此时是有问题的。
14 
15 def producer(q,product):
16     for i in range(20):
17         info = product + ‘的娃娃%s号‘%str(i)
18         q.put(info)
19     q.put(None)    # 让生产者生产完数据后，给消费者一个不再生产数据的标识
20 
21 if __name__ == ‘__main__‘:
22     q = Queue(10)
23     p_pro = Process(target=producer,args=(q,‘岛国米饭保你爱‘))
24     p_con = Process(target=consumer,args=(q,‘alex‘))
25     p_pro.start()
26     p_con.start()
27

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

把标识放在主程序中

from multiprocessing import Queue,Process
import time

def consumer(q,name,color):
    while 1:
        info = q.get()
        if info:
            print(‘%s %s 拿走了%s \033[0m‘%(color,name,info))
        else:# 当消费者获得队列中数据时，如果获得的是None，就是获得到了生产者不再生产数据的标识
            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,‘岛国米饭保你爱‘))
    p_pro2 = Process(target=producer,args=(q,‘苍老师版‘))
    p_pro3 = Process(target=producer,args=(q,‘波多多版‘))
    p_con1 = Process(target=consumer,args=(q,‘alex‘,‘\033[31m‘))
    p_con2 = Process(target=consumer,args=(q,‘wusir‘,‘\033[32m‘))
    p_l = [p_con1,p_con2,p_pro1,p_pro2,p_pro3]
    [i.start() for i in p_l]
    # 父进程如何感知到生产者子进程不再生产数据了？
    p_pro1.join()
    p_pro2.join()
    p_pro3.join()
    q.put(None)         # 几个消费者就要接受几个结束标识
    q.put(None)

View Code

那我不想自己添加标识了，这时候引入模块JoinableQueue  再结合守护进程



    （2）from multiprocessing import JoinableQueue   #可连接的队列
       JoinableQueue是继承Queue，所以可以使用Queue中的方法
       并且JoinableQueue又多了两个方法
       q.join()　　　　　　　　　　# 用于生产者。等待 q.task_done的返回结果，通过返回结果，生产者就能获得消费者当前消费了多少个数据
       q.task_done() 　　　　　　# 用于消费者，是指每消费队列中一个数据，就给join返回一个标识。

from multiprocessing import Process,JoinableQueue

q = JoinableQueue()

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

# 假设生产者生产了100个数据，join就能记录下100这个数字。每次消费者消费一个数据，就必须要task_done返回一个标识，当生产者（join）接收到100个消费者返回来的标识的时候，生产者就能知道消费者已经把所有数据都消费完了。

from multiprocessing import Queue,Process
import time

def consumer(q,name,color):
    while 1:
        info = q.get()
        print(‘%s %s 拿走了%s \033[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,‘岛国米饭保你爱‘))
    p_con1 = Process(target=consumer,args=(q,‘alex‘,‘\033[31m‘))
    p_con1.daemon = True    # 把消费者进程设为守护进程
    p_con1.start()
    p_pro1.start()
    p_pro1.join()   # 主进程等待生产者进程结束
    # 程序有3个进程，主进程和生产者进程和消费者进程。  当主进程执行到35行代码时，主进程会等待生产进程结束
    # 而生产进程中（第26行）会等待消费者进程把所有数据消费完，生产者进程才结束。
    # 现在的状态就是  主进程等待生产者进程结束，生产者进程等待消费者消费完所有数据
    # 所以，把消费者设置为守护进程。  当主进程执行完，就代表生产进程已经结束，也就代表消费者进程已经把队列中数据消费完
    # 此时，主进程一旦结束，守护进程也就是消费者进程也就跟着结束。    整个程序也就能正常结束了。

joinablequeue



     2 管道(了解)
         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())   # c1发 c2收
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()  # 生产完数据，关闭父进程这一端的管道

多进程的管道


     3 进程之间的共享内存
       from multiprocessing import Manager,Value
       m = Manager()
       num = m.dict({键 : 值})
       num = m.list([1,2,3])

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)

manager简单使用



     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

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

import os
from multiprocessing import Pool, Process
import time


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


if __name__ == ‘__main__‘:
    p = Pool(os.cpu_count() + 1)
    start = time.time()
    p.map(func, [i for i in range(100)])
    p.close()  # 指不允许在向进程池中添加任务
    p.join()  # 等待进程池中所有进程执行完所有任务
    print(‘进程池做任务的效率：‘, time.time() - start)
    start = time.time()
    p_l = []
    for i in range(100):
        p1 = Process(target=func, args=(i,))
        p1.start()
        p_l.append(p1)
    [p1.join() for p1 in p_l]
    print(‘多进程直接做任务的效率‘, time.time() - start)
    # p.apply()是指让进程池中的进程，同步的帮你做任务
    # p.apply_async()# 是指让进程池中的进程，异步的帮你做任务

多进程和进程池效率对比

p.map(func,iterable)的返回值为 func的ruturn返回的值放在列表中　　

from multiprocessing import Pool
import time




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

if __name__ == ‘__main__‘:
    p = Pool(5)
    start = time.time()
    l = []
    for i in range(10000):
        res = p.apply_async(func,args=(i,))# 异步处理这100个任务，异步是指，进程中有5个进程，一下就处理5个任务，接下来哪个进程处理完任务了，就马上去接收下一个任务
        l.append(res)   # 异步的话，res为对象
    p.close()
    p.join()
    print(time.time() - start)
    [print(i.get()) for i in l]    # 因为l为对象，所以i.get() 取值

进程池异步处理任务

from multiprocessing import Pool
import time




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

if __name__ == ‘__main__‘:
    p = Pool(5)
    start = time.time()
    for i in range(10000):
        res = p.apply(func,args=(i,))# 同步处理这100个任务，同步是指，哪怕我进程中有5个进程，也依旧是1个进程1个进程的去执行任务
        # time.sleep(0.5)
        print(res)
    print(time.time() - start)

进程池同步

from multiprocessing import Pool
import requests
import time


def func(url):
    res = requests.get(url)
    print(res.text)
    if res.status_code == 200:
        return ‘ok‘


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‘,
         ]
    start = time.time()
    for i in l:
        p.apply(func, args=(i,))

    apply_ = time.time() - start

    start = time.time()
    for i in l:
        p.apply_async(func, args=(i,))
    p.close()
    p.join()
    print(‘同步的时间是%s，异步的时间是%s‘ % (apply_, time.time() - start))

同步和异步的效率对比


异步执行任务func，每有一个进程执行完任务后，在func中return一个结果，结果会自动的被callback指定的函数，当成形式参数来接收到

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()

回调函数的使用



作业：
    0 整理博客
    1 必须会一个生产者消费者模型的编码
    2 使用Manager实现一下银行的存取款问题
    3 使用进程池去实现一下socket的tcp协议编码。允许一个服务器同时接受多个客户端的请求，
      客户端连接成功后，客户端发送一个数据，服务器把该数据转成大写发送回去就行。










总结|：
    你知道的IPC都有哪些？
       管道，队列，（锁，信号量，事件）

37 生产者消费者模型（JoinableQueue）、管道(Pipe了解)、进程之间的共享内存（Manager模块）、进程池

码农公寓

相关文章