5.进程 -锁

锁 - 事件 -IPC进程通信-生产者消费者

1.进程锁

(1) Lock : 同一时间,允许一个进程上一把锁

# 1.锁 lock
"""
同一时间,允许一个进程上一把锁,就是lock
创建锁对象 : lock = Lock()
上锁和解锁是成对出现:
    1.上锁 : lock.acquire()
    2.解锁 : lock.release()
"""
from multiprocessing import Lock,Process
import json,random,time

###12306抢票
# 1.读写数据票数
def wr_info(sign,dic=None):
    if sign == "r": #读取票数
        with open("data.txt",mode = "r",encoding="utf-8") as fp:
            dic = json.load(fp)
            return dic
    elif sign == "w": #写入票数
        with open("data.txt",mode="w",encoding="utf-8") as fp:
            json.dump(dic,fp)

# 2.执行抢票方案
def get_ticket(person):
    dic = wr_info("r") #获取数据
    time.sleep(random.uniform(0,1)) #模拟网络延迟
    if dic["amount"] > 0: 
        print("恭喜~{}抢票成功".format(person))
        dic["amount"] -= 1 #抢票成功后,总票数减一
        wr_info("w",dic) #更新票数
    else:
        print("{}抢票失败~".format(person))

def main(person,lock):
    dic = wr_info("r")
    print("{}查看剩余票数为{}张".format(person,dic["amount"]))

    lock.acquire() #上锁 确保同一时间只有一个进程进来抢票
    get_ticket(person)
    lock.release() #解锁 别的进程才有机会进来抢票

if __name__ == "__main__":
    lock = Lock() #创建锁对象
    lst = ["贾英贺","熊大","熊二","光头强","吉吉","涂涂"]
    for i in lst:
        p = Process(target=main,args=(i,lock))
        p.start()

"""
创建进程时,是异步程序
当上锁是,变成了同步程序

(2) 信号量Semaphore : 同一时间允许多个程序上n把锁

# 信号量Semaphore
"""
sem = Semaphore(n)
同一时间允许多个程序上n把锁
sem.acquire() #上锁
sem.release() # 解锁
"""
from multiprocessing import Semaphore,Process
import time,random

# 去歌房唱歌
def sing(person,sem):
    sem.acquire()
    print("{}进入歌房唱歌~".format(person))
    time.sleep(random.randrange(2,8))
    print("{}离开了~~".format(person))
    sem.release()

if __name__ == "__main__":
    sem = Semaphore(3) #同一时间3个进程上锁
    lst = [11,22,33,44,55,66,77,88,99]
    for i in lst:
        p = Process(target=sing,args=(i,sem))
        p.start()

2.事件 Event

# 事件 Event
"""
阻塞事件:
    e = Event() 生成事件对象
    e.wait() 动态给程序加阻塞,程序中是否加阻塞完全取决于该对象中的is_set()
        如果是True 不加阻塞
        如果是False 加阻塞
控制这个属性的值:
    set() 将属性值改为True
    clear() 将属性值改为False
    is_set() 判断当前属性是否为True (默认是False)
"""
from multiprocessing import Process,Event
import time,random

# 1.语法一
"""
e = Event() #生成事件对象
print(e.is_set())
e.wait()
print("代码执行了~") #阻塞,不能打印
"""
# 2.语法二
e = Event()
e.set() #将属性值改为True
e.wait()
print("代码执行了1~") #不阻塞,可以打印

# 3.语法三 阻塞几秒后放行
e = Event()
e.wait(1) #阻塞1秒后执行下面代码
print("代码执行了2~")

# 4.模拟红绿灯效果
#模拟红绿灯
def traffic_light(e): 
    print("红灯亮")
    while True:
        if e.is_set(): #True绿灯过来
            time.sleep(1)
            print("红灯亮")
            e.clear() #改成False 变红灯
        else:       #红灯过来
            time.sleep(1)
            print("绿灯亮")
            e.set() #改成True 变绿灯
            
# 模拟小车
def car(e,i):
    if not e.is_set(): #当红灯时,小车阻塞
        print("当前小车{}需等待~".format(i))
        e.wait() #小车阻塞
    # 当绿灯时小车放行    
    print("当前小车{}放行了~".format(i))

# 全国交通灯
"""
if __name__ == "__main__":
    e = Event()
    p1 = Process(target=traffic_light,args=(e,)) #创建交通灯进程
    p1.start()

    for i in range(20):
        time.sleep(random.randrange(3))
        p2 = Process(target=car,args=(e,i))
        p2.start()
"""
# 包头交通灯(车跑完后,交通灯关闭,省电)
if __name__ == "__main__":
    e = Event()
    p1 = Process(target=traffic_light,args=(e,))
    p1.daemon = True
    p1.start()

    lst = []
    for i in range(1,21):
        time.sleep(random.randrange(2))
        p2 = Process(target=car,args=(e,i))
        lst.append(p2)
        p2.start()
    for i in lst:
        i.join()
    print("关闭交通灯")

3.进程队列 IPC进程通信

(1) Queue

# 进程间通信IPC (Inter-Process Communication)
"""
实现进程间通信两种机制:
    1.管道 Pipe
    2.队列 Queue
q.put() 存放数据
q.get() 获取数据 
get_nowait() 拿不到报异常
put_nowait() 数据存满再存报错
q.empty()  检测是否为空
q.full()  检测是否存满
"""

# 进程队列
from multiprocessing import Queue,Process
"""
先进先出,后进后出
q = Queue()  生成队列对象
q = Queue(n) 生成可以存n个数据队列对象
"""

# 1.基本语法
q = Queue() #生成队列对象
q.put(1) #存放数据
q.put(2)
print(q.get()) # 获取数据 1
print(q.get()) # 2
# print(q.get()) #获取不到数据阻塞

# 2.进程间IPC通信
def func(q):
    res = q.get() #2.子进程获取主进程存放的数据
    print(res,"<子进程>")
    q.put("贾英贺") #3.子进程存放数据

if __name__ == "__main__":
    q = Queue()
    p = Process(target=func,args=(q,))
    p.start()
    q.put("光头强") #1.主进程存储数据
    p.join() #为了让主进程获取数据,必须等待子进程执行完毕后,在向下执行
    res = q.get()
    print(res,"<主进程>")
"""
光头强 <子进程>
贾英贺 <主进程>
"""

(2) Queue 生产消费模型

# 生产者消费者模型
"""
从程序上看:
    生产者负责储存数据(put)
    消费者负责获取数据(get)
理想的生产者消费者模型:
    1.生产多少,消费多少
    2.生产数据的速度与消费数据的速度相对一致
"""
from multiprocessing import Queue,Process
import time,random

# 1.基础版
def producer(q,name,food): #生产者
    for i in range(1,5):
        time.sleep(random.random())
        res = "第{}碗{}".format(i,food)
        print("{}生产了{}".format(name,res))
        q.put(res) #储存数据

def consumer(q,name): #消费者
    while True:
        time.sleep(random.random())
        res = q.get() #获取数据
        print("{}吃了{}".format(name,res))

if __name__ == "__main__":
    q = Queue()
    p1 = Process(target=producer,args=(q,"贾英贺","粥"))
    p2 = Process(target=consumer,args=(q,"光头强"))
    p1.start()
    p2.start()
缺点 : 循环没断

(3) JoinableQueue 生产消费模型

# JoinableQueue 队列
"""
put 存放一次数据 计数器属性值 +1
get 获取 
task_done 一次数据, 计数器属性值 -1
当计数器值=0时
    队列.join() 放行
当计数器值非0时
    队列.join() 阻塞
"""
# 对生产者消费者模型进行改造
from multiprocessing import JoinableQueue,Process
import time,random

# 生产者
def producer(q,name,food):
    for i in range(1,5):
        time.sleep(random.random())
        res = "第{}碗{}".format(i,food)
        print("{}生产{}".format(name,res))
        q.put(res) #存储数据,内置计数器+1

# 消费者
def consumer(q,name):
    while True:
        time.sleep(random.random())
        food = q.get() #获取数据
        print("{}吃了{}".format(name,food))
        q.task_done() #内置计数器-1

if __name__ == "__main__":
    jq = JoinableQueue()
    p1 = Process(target = producer, args=(jq,"光拖欠","小米粥"))
    p2 = Process(target= consumer, args = (jq,"熊大"))
    p2.daemon = True #守护消费者进程
    p1.start()
    p2.start()
    p1.join() #必须等到生产者全部生产完毕,在放行
    jq.join() #必须让消费者全部吃完
    print("主进程结束")

4.Maneger (list,dict) 多进程之间共享列表,字典

# Manager(list,dict)
from multiprocessing import Process,Manager,Lock

def work(data,lock):
    with lock: # 上锁解锁自动完成
        data[0] -=1
        
if __name__=="__main__":
    lst = []
    lock = Lock()
    m = Manager()
    # 创建一个多进程之间共享数据的字典
	# data = m.dict( {"count":0} )
    # 创建一个多进程之间共享数据的列表
    data = m.list([111,222])
    for i in range(10):
        p = Process(target  = work ,args = (data,lock))
        p.start()
        lst.append(p)
    for i in lst:
        i.join()
    print(data)#[101, 222]