python3全栈开发-多进程的守护进程、进程同步、生产者消费者模式（重点）

2023-01-06 22:44:16

一、守护进程

主进程创建守护进程

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

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

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

from multiprocessing import Process，Lock

import time

mutex=Lock()

def task(name):

    print("%s is running"%name)

    time.sleep(3)

if __name__=="__main":

    p=Process(target=task,args=("duoduo",))

    p.daemon=True #一定要在p.start()前设置,设置p为守护进程,禁止p创建子进程,并且父进程代码执行结束,p即终止运行

    p.start()

    print("----------------->")

守护进程运用

二、进程同步(锁)

进程之间数据不共享,但是共享同一套文件系统,所以访问同一个文件,或同一个打印终端,是没有问题的,

而共享带来的是竞争，竞争带来的结果就是错乱，如何控制，就是加锁处理

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

#并发运行,效率高,但竞争同一打印终端,带来了打印错乱

from multiprocessing import Process

import os,time

def work():

    print('%s is running' %os.getpid())

    time.sleep(2)

    print('%s is done' %os.getpid())

if __name__ == '__main__':

    for i in range(3):

        p=Process(target=work)

        p.start()

共享同一打印终端

#由并发变成了串行,牺牲了运行效率,但避免了竞争

from multiprocessing import Process,Lock

import os,time

def work(lock):

    lock.acquire()

    print('%s is running' %os.getpid())

    time.sleep(2)

    print('%s is done' %os.getpid())

    lock.release()

if __name__ == '__main__':

    lock=Lock()

    for i in range(3):

        p=Process(target=work,args=(lock,))

        p.start()

加锁状态

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

文件当数据库，模拟抢票

#文件db的内容为：{"count":1}

#注意一定要用双引号，不然json无法识别

from multiprocessing import Process

import time,json,random

def search():

    dic=json.load(open('db.txt'))

    print('\033[43m剩余票数%s\033[0m' %dic['count'])

def get():

    dic=json.load(open('db.txt'))

    time.sleep(0.1) #模拟读数据的网络延迟

    if dic['count'] >0:

        dic['count']-=1

        time.sleep(0.2) #模拟写数据的网络延迟

        json.dump(dic,open('db.txt','w'))

        print('\033[43m购票成功\033[0m')

def task():

    search()

    get()

if __name__ == '__main__':

    for i in range(100): #模拟并发100个客户端抢票

        p=Process(target=task)

        p.start()

模拟抢票

#文件db的内容为：{"count":1}

#注意一定要用双引号，不然json无法识别

from multiprocessing import Process,Lock

import time,json,random

def search():

    dic=json.load(open('db.txt'))

    print('\033[43m剩余票数%s\033[0m' %dic['count'])

def get():

    dic=json.load(open('db.txt'))

    time.sleep(0.1) #模拟读数据的网络延迟

    if dic['count'] >0:

        dic['count']-=1

        time.sleep(0.2) #模拟写数据的网络延迟

        json.dump(dic,open('db.txt','w'))

        print('\033[43m购票成功\033[0m')

def task(lock):

    search()

    lock.acquire()

    get()

    lock.release()

if __name__ == '__main__':

    lock=Lock()

    for i in range(100): #模拟并发100个客户端抢票

        p=Process(target=task,args=(lock,))

        p.start()

加锁状态抢票

总结：

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

虽然可以用文件共享数据实现进程间通信，但问题是：

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

2.需要自己加锁处理

#因此我们最好找寻一种解决方案能够兼顾：

1、效率高（多个进程共享一块内存的数据）

2、帮我们处理好锁问题。这就是mutiprocessing模块为我们提供的基于消息的IPC通信机制：队列和管道。

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

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

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

小结、引子

三、队列（推荐使用）

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

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

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

参数介绍：

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方法可以禁止这种行为

其他方法（了解）

　应用：

'''

multiprocessing模块支持进程间通信的两种主要形式:管道和队列

都是基于消息传递实现的,但是队列接口

'''

from multiprocessing import Process,Queue

import time

q=Queue(3)

#put ,get ,put_nowait,get_nowait,full,empty

q.put(3)

q.put(3)

q.put(3)

print(q.full()) #满了

print(q.get())

print(q.get())

print(q.get())

print(q.empty()) #空了

Queue

四、生产者消费者模型

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

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

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

什么是生产者消费者模式

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

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

from multiprocessing import Process,Queue

import time,random,os

def consumer(q):

    while True:

        res=q.get()

        time.sleep(random.randint(1,3))

        print('\033[45m%s 吃 %s\033[0m' %(os.getpid(),res))

def producer(q):

    for i in range(10):

        time.sleep(random.randint(1,3))

        res='包子%s' %i

        q.put(res)

        print('\033[44m%s 生产了 %s\033[0m' %(os.getpid(),res))

if __name__ == '__main__':

    q=Queue()

    #生产者们:即厨师们

    p1=Process(target=producer,args=(q,))

    #消费者们:即吃货们

    c1=Process(target=consumer,args=(q,))

    #开始

    p1.start()

    c1.start()

生产包子吃包子

#生产者消费者模型总结

    #程序中有两类角色

        一类负责生产数据（生产者）

        一类负责处理数据（消费者）

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

        平衡生产者与消费者之间的工作能力，从而提高程序整体处理数据的速度

    #如何实现：

        生产者<-->队列<——>消费者

    #生产者消费者模型实现类程序的解耦和

总结

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

解决方式无非是让生产者在生产完毕后，往队列中再发一个结束信号，这样消费者在接收到结束信号后就可以break出死循环

from multiprocessing import Process,Queue

import time,random,os

def consumer(q):

    while True:

        res=q.get()

        if res is None:break #收到结束信号则结束

        time.sleep(random.randint(1,3))

        print('\033[45m%s 吃 %s\033[0m' %(os.getpid(),res))

def producer(q):

    for i in range(10):

        time.sleep(random.randint(1,3))

        res='包子%s' %i

        q.put(res)

        print('\033[44m%s 生产了 %s\033[0m' %(os.getpid(),res))

    q.put(None) #发送结束信号

if __name__ == '__main__':

    q=Queue()

    #生产者们:即厨师们

    p1=Process(target=producer,args=(q,))

    #消费者们:即吃货们

    c1=Process(target=consumer,args=(q,))

    #开始

    p1.start()

    c1.start()

    print('主')

解决思路1

注意：结束信号None，不一定要由生产者发，主进程里同样可以发，但主进程需要等生产者结束后才应该发送该信号

from multiprocessing import Process,Queue

import time,random,os

def consumer(q):

    while True:

        res=q.get()

        if res is None:break #收到结束信号则结束

        time.sleep(random.randint(1,3))

        print('\033[45m%s 吃 %s\033[0m' %(os.getpid(),res))

def producer(q):

    for i in range(2):

        time.sleep(random.randint(1,3))

        res='包子%s' %i

        q.put(res)

        print('\033[44m%s 生产了 %s\033[0m' %(os.getpid(),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) #发送结束信号

    print('主')

解决思路2

但上述解决方式，在有多个生产者和多个消费者时，我们则需要用一个很low的方式去解决

from multiprocessing import Process,Queue

import time,random,os

def consumer(q):

    while True:

        res=q.get()

        if res is None:break #收到结束信号则结束

        time.sleep(random.randint(1,3))

        print('\033[45m%s 吃 %s\033[0m' %(os.getpid(),res))

def producer(name,q):

    for i in range(2):

        time.sleep(random.randint(1,3))

        res='%s%s' %(name,i)

        q.put(res)

        print('\033[44m%s 生产了 %s\033[0m' %(os.getpid(),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()

    p1.join() #必须保证生产者全部生产完毕,才应该发送结束信号

    p2.join()

    p3.join()

    q.put(None) #有几个消费者就应该发送几次结束信号None

    q.put(None) #发送结束信号

    print('主')

解决思路3

其实我们的思路无非是发送结束信号而已，有另外一种队列提供了这种机制

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

   #参数介绍：

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

　 #方法介绍：

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

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

    q.join():生产者调用此方法进行阻塞，直到队列中所有的项目均被处理。阻塞将持续到队列中的每个项目均调用q.task_done（）方法为止

JoinableQueue模块

from multiprocessing import Process,JoinableQueue

import time,random,os

def consumer(q):

    while True:

        res=q.get()

        time.sleep(random.randint(1,3))

        print('\033[45m%s 吃 %s\033[0m' %(os.getpid(),res))

        q.task_done() #向q.join()发送一次信号,证明一个数据已经被取走了

def producer(name,q):

    for i in range(10):

        time.sleep(random.randint(1,3))

        res='%s%s' %(name,i)

        q.put(res)

        print('\033[44m%s 生产了 %s\033[0m' %(os.getpid(),res))

    q.join()

if __name__ == '__main__':

    q=JoinableQueue()

    #生产者们:即厨师们

    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

    #开始

    p_l=[p1,p2,p3,c1,c2]

    for p in p_l:

        p.start()

    p1.join()

    p2.join()

    p3.join()

    print('主') 

    #主进程等--->p1,p2,p3等---->c1,c2

    #p1,p2,p3结束了,证明c1,c2肯定全都收完了p1,p2,p3发到队列的数据

    #因而c1,c2也没有存在的价值了,应该随着主进程的结束而结束,所以设置成守护进程

终极版本

五、共享数据

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

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

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

还可以扩展到分布式系统中

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

#进程间数据是独立的，可以借助于队列或管道实现通信，二者都是基于消息传递的

#虽然进程间数据独立，但可以通过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}

例子

码农公寓

一、守护进程

二、进程同步(锁)

三、 队列（推荐使用）

四、生产者消费者模型

五、共享数据

相关文章

三、队列（推荐使用）