守护进程、互斥锁、队列、生产者消费者模型

守护进程

关于守护进程：
其一：守护进程会在主进程代码执行结束后就终止
其二：守护进程内无法再开启子进程,否则抛出异常：AssertionError: daemonic processes are not allowed to have children

#!/usr/bin/env python3
# -*- coding:utf-8 -*-

from multiprocessing import Process
import time

def task(name):
    print('%s is runing'%name)
    time.sleep(3)
    print('%s is done'%name)

if __name__ == '__main__':
    p1 = Process(target=task,args=('子进程1',))
    p2 = Process(target=task,args=('子进程2',))
    p2.daemon = True # 设置守护进程
    p1.start()
    p2.start()
    print('主')
'''
输出结果：
主
子进程1 is runing
子进程1 is don
'''

互斥锁（Lock)

互斥锁：互斥锁的意思就是互相排斥，如果把多个进程比喻为多个人，互斥锁的工作原理就是多个人都要去争抢同一个资源：卫生间，一个人抢到卫生间后上一把锁，其他人都要等着，等到这个完成任务后释放锁，其他人才有可能有一个抢到......所以互斥锁的原理，就是把并发改成穿行，降低了效率，但保证了数据安全不错乱
例子：

#由并发变成了串行,牺牲了运行效率,但避免了竞争
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()

模拟抢票练习

#!/usr/bin/env python3
# -*- coding:utf-8 -*-

import json
from multiprocessing import Process,Lock
import time
def search(name):
    time.sleep(1)
    data = json.load(open('a.tx','r',encoding='utf-8'))
    print('%s<查看剩余票数>%s'%(name,data['count']))

def get(name):
    time.sleep(2)
    data = json.load(open('a.tx','r',encoding='utf-8'))
    if data['count'] > 0:
        data['count'] -= 1
        time.sleep(3)
        json.dump(data,open('a.tx','w',encoding='utf-8'))
        print('%s购票成功'%name)
    else:
        print('%s购票失败'%name)
def run(name,mutex):
    search(name)  # 并发执行
    mutex.acquire()  # 加锁
    get(name) # 串行执行
    mutex.release() # 释放锁

if __name__ == '__main__':
    mutex = Lock()
    for i in range(10): # 并发10个用户
        p1 = Process(target=run,args=('路人%s'%i,mutex))
        p1.start()

join与互斥锁的区别

join是将一个任务整体串行，而互斥锁的好处则是可以将一个任务中的某一段代码串行，比如只让run函数中的get任务串行

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

队列

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

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

参数介绍：

maxsize是队列中允许最大项数，省略则无大小限制。
但需要明确：
    1、队列内存放的是消息而非大数据
    2、队列占用的是内存空间，因而maxsize即便是无大小限制也受限于内存大小

主要方法介绍：

q.put方法用以插入数据到队列中。
q.get方法可以从队列读取并且删除一个元素。

队列的使用

from multiprocessing import Process,Queue
q=Queue(3)
#put ,get ,put_nowait,get_nowait,full,empty
q.put(1)
q.put(2)
q.put(3)
print(q.full()) #满了
# q.put(4) #再放就阻塞住了

print(q.get())
print(q.get())
print(q.get())
print(q.empty()) #空了
# print(q.get()) #再取就阻塞住了

生产者消费者模型

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

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

什么是生产者和消费者模式

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

这个阻塞队列就是用来给生产者和消费者解耦的

生产者消费者模型实现

#!/usr/bin/env python3
# -*- coding:utf-8 -*-

from  multiprocessing import Process,Queue
import time

def produce(q,name):
    for i in range(3):
        res = '%s is 包子'%i
        time.sleep(0.5)
        print('%s生产了%s'%(name,res))
        q.put(res)

def consumer(q,name):
    while True:
        res = q.get()
        time.sleep(1)
        print('%s消费了%s'%(name,res))

if __name__ == '__main__':
    q = Queue()
    # 生产者
    p1 = Process(target=produce,args=(q,'路人a',))
    p2 = Process(target=produce,args=(q,'路人b',))
    # 消费者
    c1 = Process(target=consumer,args=(q,'路人c'))
    c2 = Process(target=consumer,args=(q,'路人d',))

    p1.start()
    p2.start()
    c1.start()
    c2.start()
    p1.join()
    p2.join()
    q.put(None)
    q.put(None)
    print('主')
'''
输出结果：
路人a生产了0 is 包子
路人b生产了0 is 包子
路人a生产了1 is 包子
路人b生产了1 is 包子
路人a生产了2 is 包子
路人c消费了0 is 包子
路人b生产了2 is 包子
路人d消费了0 is 包子
主
路人c消费了1 is 包子
路人d消费了1 is 包子
路人c消费了2 is 包子
路人d消费了2 is 包子
路人c消费了None
路人d消费了None
'''

JoinableQueue 的使用

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

参数介绍

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

方法介绍

JoinableQueue的实例p除了与Queue对象相同的方法之外还具有：
q.task_done()：使用者使用此方法发出信号，表示q.get()的返回项目已经被处理。如果调用此方法的次数大于从队列中删除项目的数量，将引发ValueError异常
q.join():生产者调用此方法进行阻塞，直到队列中所有的项目均被处理。阻塞将持续到队列中的每个项目均调用q.task_done（）方法为止

基于JoinableQueue实现生产者消费者模型

#!/usr/bin/env python3
# -*- coding:utf-8 -*-
from  multiprocessing import Process,JoinableQueue
import time

def produce(q,name):
    for i in range(10):
        res = '%s is 包子'%i
        time.sleep(0.5)
        print('%s生产了%s'%(name,res))
        q.put(res)

def consumer(q,name):
    while True:
        res = q.get()
        time.sleep(1)
        print('%s消费了%s'%(name,res))

if __name__ == '__main__':
    q = JoinableQueue()
    # 生产者
    p1 = Process(target=produce,args=(q,'路人甲',))
    p2 = Process(target=produce,args=(q,'路人甲',))
    # 消费者
    c1 = Process(target=consumer,args=(q,'路人乙'))
    c2 = Process(target=consumer,args=(q,'路人乙',))
    c1.daemon = True 
    c2.daemon = True
    p1.start()
    p2.start()
    c1.start()
    c2.start()
    p1.join()
    p2.join()
    print('主')
#1、主进程等生产者p1、p2结束
#2、而p1、p2是在消费者把所有数据都取干净之后才会结束
 #3、所以一旦p1、p2结束了，证明消费者也没必要存在了，应该随着主进程一块死掉，因而需要将生产者们设置成守护进程

生产者消费者模型总结

1、程序中有两类角色

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

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

平衡生产者与消费者之间的速度差
程序解开耦合

3、如何实现生产者消费者模型

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