python并发编程之守护进程、互斥锁以及生产者和消费者模型

一、守护进程

主进程创建守护进程

守护进程其实就是'子进程'

一、守护进程内无法在开启子进程，否则会报错
二、进程之间代码是相互独立的，主进程代码运行完毕，守护进程也会随机结束

守护进程简单实例：

from multiprocessing import Process
import time
def task(name):                       #此时的task为守护进程
    print('%s is running' % name)     #该行并不会被打印，因为主进程结束，守护进程会随之结束
    time.sleep(3)

if __name__ == '__main__':
    obj = Process(target=task, args=('egon',))
    obj.daemon=True           #一定要在obj.start()开启之前，将obj设置为守护进程，禁止obj创建子进程，否则就会报错
    obj.start()               # 发送信号给操作系统开启一个进程（守护进程）
    print('主')               #并且主进程结束，守护进程也会随之结束

为什么要用守护进程：

两方面理解：

一、进程：为了让父进程的任务能够并发的执行，需要将该任务放到子进程中去

二、守护：是因为子进程中执行的任务，在父进程运行完就没有存在的意义了，就设置成守护进程，会在父进程结束后，守护进程也随之结束

由于计算机的性能的原因，进程打印出的结果可能会有不同的结果：

from multiprocessing import Process
import time
def foo():                      #主进程结束，守护进程就会结束
    print(123)
    time.sleep(1)
    print("end123")

def bar():                      #正常的子进程，父进程会等正常的子进程结束才会结束(父进程可以看到子进程的pid)
    print(456)
    time.sleep(3)
    print("end456")

if __name__ == '__main__':

    p1=Process(target=foo)                #把p1做成守护进程
    p2=Process(target=bar)                #正常的子进程

    p1.daemon=True                #一定要在开启子进程之前，否则会报错，守护进程会伴随主进程的代码的运行完毕而死掉
    p1.start()
    p2.start()
    print("main-------")         #主进程结束，守护进程也随之结束

#正常机器
'''
main-------
456
end456
'''

#快一点机器
'''
main-------     #申请一个内存空间，然后调用操作系统，将其打印到终端
123             #机器性能好的话，可能在打印的时候，p1进程已经起来了，即向操作系统发送请求后，很快就做出回应开启了子进程
456
end456
'''

# 机器更快
'''
123          #当机器的性能在好一点，有可能会出现，p1向操作系统发起开启子进程的请求，迅速做出回应，然后开启子进程，那么123就会优先main先被打印出来
main-------
456
end456
'''

二、互斥锁

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

然而共享带来的是竞争，竞争带来的就是错乱无序，所以我们要加锁进行处理

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

'''能不自己处理锁，就尽量不用锁，'''
#进程间通信，一个进程修改后，其他进程能看到修改后的结果
'''互斥所就是把并发变成串行，保证了数据安全，但是牺牲了效率'''
'''锁同一时间锁只能被一个子进程抢，用完了再释放掉，才能被其他进程抢到再用'''
"""互斥锁可以将要执行代码的（部分共享的数据）变成串行，而join是将要执行代码所有部分变成串行"""

# 互斥锁：
#强调：必须是lock.acquire()一次，然后 lock.release()释放一次，才能继续lock.acquire()，不能连续的lock.acquire()

# 互斥锁vs join的区别一：
# 大前提：二者的原理都是一样，都是将并发变成串行，从而保证有序
# 区别：join是按照人为指定的顺序执行，而互斥锁是所以进程平等地竞争，谁先抢到谁执行

# 没加锁之前，打印会变的错乱
# 并发运行,效率高,但竞争同一打印终端,带来了打印错乱
from multiprocessing import Process
import time,random

def task1():               #为每一个进程加上一把锁，这样就会有和join一样的效果，变成了串行
    print('task1:名字是egon')
    time.sleep(random.randint(1,3))
    print('task1:性别是male')
    time.sleep(random.randint(1,3))
    print('task1:年龄是18')

def task2():
    print('task2:名字是alex')
    time.sleep(random.randint(1,3))
    print('task2:性别是male')
    time.sleep(random.randint(1,3))
    print('task2:年龄是78')

def task3():
    print('task3:名字是lxx')
    time.sleep(random.randint(1,3))
    print('task3:性别是female')
    time.sleep(random.randint(1,3))
    print('task3:年龄是30')

if __name__ == '__main__':
    p1=Process(target=task1)
    p2=Process(target=task2)
    p3=Process(target=task3)

    p1.start()
    p2.start()
    p3.start()

'''
运行结果：
task1:名字是egon
task3:名字是lxx
task2:名字是alex
task1:性别是male
task3:性别是female
task2:性别是male
task1:年龄是18
task3:年龄是30
task2:年龄是78
'''

from multiprocessing import Process,Lock
import time,random
mutex=Lock()
# 为每一个进程加锁，就可以让其变为串行，牺牲了效率，但保证了数据安全
def task1(lock):               #为每一个进程加上一把锁，这样就会有和join一样的效果，变成了串行
    lock.acquire() #
    print('task1:名字是egon')
    time.sleep(random.randint(1,3))
    print('task1:性别是male')
    time.sleep(random.randint(1,3))
    print('task1:年龄是18')
    lock.release()

def task2(lock):
    lock.acquire()
    print('task2:名字是alex')
    time.sleep(random.randint(1,3))
    print('task2:性别是male')
    time.sleep(random.randint(1,3))
    print('task2:年龄是78')
    lock.release()


def task3(lock):
    lock.acquire()
    print('task3:名字是lxx')
    time.sleep(random.randint(1,3))
    print('task3:性别是female')
    time.sleep(random.randint(1,3))
    print('task3:年龄是30')
    lock.release()


if __name__ == '__main__':
    p1=Process(target=task1,args=(mutex,))
    p2=Process(target=task2,args=(mutex,))
    p3=Process(target=task3,args=(mutex,))

    # p1.start()
    # p1.join()
    # p2.start()
    # p2.join()
    # p3.start()
    # p3.join()

    p1.start()
    p2.start()
    p3.start()

多个进程共享同一份文件--------模拟抢票

通过文件：db.json来模拟数据库

文件内容：

{"count":1}

# 共享一份数据
# 文件db.json内容：
# {'count':1}
import json
import time
import random
import os
from multiprocessing import Process
'''大家抢的是同一份数据，先有查票，然后购票，先在客户端减1，在发送到服务端减1'''
# 查票应该并发
def search():
    time.sleep(random.randint(1,3))        #模拟网络延时
    with open('db.json','r',encoding='utf-8') as f:
        dic=json.load(f)
        print('%s 剩余票数:%s' %(os.getpid(),dic['count']))  #用pi进行标识

# 购票应该一个一个来，才不至于购票时所有人都减1
def get():
    with open('db.json','r',encoding='utf-8') as f:   #之前查看的可能不准确，所以再打开一次
        dic=json.load(f)
    time.sleep(random.randint(1, 3))
    if dic['count'] > 0:
        dic['count']-=1       #在客户端的内存里减1，但要把他刷新到服务端的文件中去
        time.sleep(random.randint(1,3))        #往文件中刷也要有一个网络延迟
        with open('db.json','w',encoding='utf-8') as f:
            json.dump(dic,f)
        print('%s 购票成功' %os.getpid())

def task():
    search()             #先查票，
    get()                #在购票

if __name__ == '__main__':
    for i in range(10):                  #网络延迟已经够把10个进程都创建完
        p=Process(target=task)
        p.start()
        # p.join()                       #加上join后变为串行，但是此时的问题是大家要排队来查票，保证了数据安全，但是效率变低了，因为查票和购票都变成了串行


# 打印结果：
'''
12888 剩余票数:1
10352 剩余票数:1
10440 剩余票数:1
228 剩余票数:1
10736 剩余票数:1
7032 剩余票数:1
8580 剩余票数:1
2712 剩余票数:1
4248 剩余票数:1
2332 剩余票数:1
12888 购票成功
10736 购票成功
228 购票成功
10352 购票成功
10440 购票成功
4248 购票成功
8580 购票成功
2332 购票成功
7032 购票成功
2712 购票成功

'''

"""
抢票的正确思路是：
查看票数应该是并发进行，提升效率
购票时应加锁，让其串行，虽然降低了效率，但是保证了数据安全即修改时不会导致错乱
"""
import json
import time
import random
import os
from multiprocessing import Process,Lock
'''大家抢的是同一份数据，先有查票，然后购票，先在客户端减1，在发送到服务端减1'''
mutex=Lock()
# 互斥锁vs join的区别一：
# 互斥锁可以让一部分代码（只修改共享数据的代码）变成串行，而join只能将代码整体串行
#查票
def search():
    time.sleep(random.randint(1,3))        #模拟网络延时
    with open('db.json','r',encoding='utf-8') as f:
        dic=json.load(f)
        print('%s 剩余票数:%s' %(os.getpid(),dic['count']))  #用pi进行标识
#购票
def get():
    with open('db.json','r',encoding='utf-8') as f:   #之前查看的可能不准确，所以再打开一次
        dic=json.load(f)
    time.sleep(random.randint(1, 3))
    if dic['count'] > 0:
        dic['count']-=1       #在客户端的内存里减1，但要把他刷新到服务端的文件中去
        time.sleep(random.randint(1,3))        #往文件中刷也要有一个网络延迟
        with open('db.json','w',encoding='utf-8') as f:
            json.dump(dic,f)
        print('%s 购票成功' %os.getpid())

def task(lock):
    search()             #先查票，(是一个单独的行为，等你查完票在取购票的时候，可能票已经被抢完了)

    # 购票变成串行，对get即购票进行加锁处理
    lock.acquire()       #mutex=Lock().acquire()---------互斥所不能连续的acquire，必须释放掉才能在acquire，(有种独享的感觉)
    get()                #在购票
    lock.release()       #mutex=Lock().release()--------(释放掉锁，这把锁可以在被其他人抢到)

if __name__ == '__main__':
    for i in range(10):                  #网络延迟已经够把10个进程都创建完
        p=Process(target=task,args=(mutex,))
        p.start()
        # p.join()

 总结：

#加锁可以保证多个进程修改同一块数据时，同一时间只能有一个任务可以进行修改，即串行的修改，没错，速度是慢了，但牺牲了速度却保证了数据安全。
虽然可以用文件共享数据实现进程间通信，但问题是：
1.效率低（共享数据基于文件，而文件是硬盘上的数据）
2.需要自己加锁处理



#因此我们最好找寻一种解决方案能够兼顾：1、效率高（多个进程共享一块内存的数据）2、帮我们处理好锁问题。这就是mutiprocessing模块为我们提供的基于消息的IPC通信机制：队列和管道。
队列和管道都是将数据存放于内存中
队列又是基于（管道+锁）实现的，可以让我们从复杂的锁问题中解脱出来，
我们应该尽量避免使用共享数据，尽可能使用消息传递和队列，避免处理复杂的同步和锁问题，而且在进程数目增多时，往往可以获得更好的可获展性。

三、IPC通信机制

进程间的通信就是通过队列或管道来实现的，而队列本质就是管道+锁，所以在这我们只研究队列

进程之间通信必须找到一种介质，该介质必须满足
1、是所有进程共享的
2、必须是内存空间
附加：帮我们自动处理好锁的问题

"""
对列：是管道+锁实现的（管道本身就是一个共享的数据，但是管道没有加锁的处理，)----为了实现进程间通信的
1、队列是共享的空间
2、占用的是内存空间
3、自动帮我们处理好锁定问题
"""
from multiprocessing import Queue
'''参数是3表明队列最多只能放三次数据。我们通过put放数据，通过get取数据'''
q=Queue(3) #队列占用的是内存空间。-1代表队列没有限制，但是不能放大数据，应该往队列中放数据量比较小的消息数据
q.put('first')          #firs是放的对象
q.put({'second':None})
q.put('三')
#
# # q.put(4) #阻塞，------放满了就不能再放了，再放就会出现阻塞状态，这是因为有自动加锁的处理机制导致阻塞的，要想在放需取出数据才可继续再放
print(q.get())         #取数据，先进先出，先放进去的先被取出来
print(q.get())
print(q.get())
print(q.get())        #第四次在取会处于阻塞状态，这是因为队列自带锁处理机制，要想再取数据，需要向队列中放数据，否则一直处于阻塞状态

队列主要方法：

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

# 先进先出
from multiprocessing import Queue
q=Queue(5)
q.put('first',block=False)     #等价于：q.put_nowait('first)---------block=False，加超时时间没有意义，以为满了或者去空就会报错，此时和超时时间已经没有关系，无论超时时间等于多少，都会立刻报错，没有等的过程
q.put('second',block=False)
q.put('third',block=False)
q.put('fourth',block=False)    #不加锁，第四次队列放满了，不会被阻塞，而是直接抛出异常

q.put('first',block=True)      #默认锁，满了就不让往里面放了
q.put('second',block=True)     #超时时间只在block=True才有意义
q.put('third',block=True)
q.put('fourth',block=True,timeout=3) #加锁，timeout超时时间，第四次放已经满了，在放3秒后将抛出异常，如果没有放满，3秒后不会抛出异常


print(q.get(block=False))            #等价于：q.get_nowait()
print(q.get(block=False))
print(q.get(block=False))
print(q.get(block=False))            #不加锁，第四次队列取空了，不会被阻塞，而是直接抛出异常

print(q.get(block=True))
print(q.get(block=True))
print(q.get(block=True))              #超时时间只在block=True才有意义
print(q.get(block=True,timeout=2))    #加锁，第四次取已经去空，2秒后将抛出异常，如果没有取空，2秒后不会抛出异常

四、生产者和消费者模型

# # 终极版：
import time
import random
from multiprocessing import Process,JoinableQueue   #(队列空了就就执行join操作)

def consumer(name,q):           #
    while True:                 #源源不断的取是一个循环的过程
        res=q.get()
        time.sleep(random.randint(1,3))   #模拟经过一段时间处理完数据
        print('33[46m消费者===》%s 吃了 %s33[0m' %(name,res))
        q.task_done()

def producer(name,q,food):
    for i in range(5):
        time.sleep(random.randint(1,2))      #模拟生产者生产数据的时间
        res='%s%s' %(food,i)
        q.put(res)          #数据生产好了，不断的往队列里面放，供消费者消费
        print('33[45m生产者者===》%s 生产了 %s33[0m' %(name,res))


if __name__ == '__main__':
    #1、共享的盆
    q=JoinableQueue()         #中间的介质，不知道放多少个，所以不用写限制

    '''生产者和消费者都可以有多个'''
    #2、生产者们
    p1=Process(target=producer,args=('egon',q,'包子'))
    p2=Process(target=producer,args=('刘清政',q,'泔水'))
    p3=Process(target=producer,args=('杨军',q,'米饭'))

    #3、消费者们,消费者还没结束，，设置为守护进程
    c1=Process(target=consumer,args=('alex',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()

    q.join()    #等待队列被取干净
    #主进程代码运行完毕-----队列中的数据被取干净----消费者没有存在的意义了----设置为守护进程
    print('主')