day35 守护进程、互斥锁、IPC
1、守护进程
# 守护进程：当父进程执行完毕后，设置的守护进程也会跟着结束
# 当一个进程被设置为守护进程后，其不能再产生子进程
from multiprocessing import Process import time def test(): print("妃子入宫了") time.sleep(5) print("子进程挂了") if __name__ == '__main__': p = Process(target=test) p.daemon = True p.start() time.sleep(3) print("父进程挂了") # 妃子入宫了 # 父进程挂了
2、互斥锁
# 在进行打印操作时，如果三个全部进行打印操作就会互相影响打印结果
# 所以此时需要将打印操作设置为串行
#使用join方法可以将优先级进行提升，但是这样执行顺序就固定了，就不能使每个子进程公平竞争CPU
# 如果按照下方的程序进行处理，那么这样就没有开启子进程的意义了
def test(): print("i am li") time.sleep(random.random()) print("my age is 15") time.sleep(random.random()) print("i am a boy") def test2(): print("i am lee") time.sleep(random.random()) print("my age is 20") time.sleep(random.random()) print("i am a girl") def test3(): print("i am wu") time.sleep(random.random()) print("my age is 25") time.sleep(random.random()) print("i am a man") if __name__ == '__main__': p = multiprocessing.Process(target=test) p1 = multiprocessing.Process(target=test3) p2 = multiprocessing.Process(target=test2) p.start() p.join() p1.start() p1.join() p2.start() p2.join() # i am li # my age is 15 # i am a boy # i am lee # my age is 20 # i am a girl # i am wu # my age is 25 # i am a man
# 使用互斥锁
# lock = Lock()
# 注意：互斥锁不能在全部变量中进行设置，这是因为子进程在进行设置时，会导入父进程的代码，
# 只有放在if __name__ == "__main__"中确保使用的是同一把锁
# 使用join与互斥锁的区别：
# 1、join会人为的打乱子进程的优先级，从而使某个子进程首先执行完，打乱了公平竞争
# 2、join只能将整个函数的优先级进行提高，不能对其中的某一行进行操作
# 互斥锁可以直接对函数中的某一行代码进行操作，这样在进行程序的执行时，可以提高程序的执行效率
# 这种又称之为粒度，粒度越小，执行效率越高，粒度越大执行效率越低
# 实际上，互斥锁不是锁住的文件或者共享资源，
# 而是在共享内存中增加一个标志，当碰到Lock对象时就会去全局变量中访问这个标志，如果这个标志为True就会执行acquire方法，将共享空间中的标志改为False
# 如果标志为False就不能执行这段代码，直到等到更改这个变量的对象再次访问全局变量，并使用release方法将这个标志重新改为True,其余进程才能访问并更改这个标志
def test(lock): lock.acquire() print("i am li") time.sleep(random.random()) print("my age is 15") time.sleep(random.random()) print("i am a boy") lock.release() def test2(lock): lock.acquire() print("i am lee") time.sleep(random.random()) print("my age is 20") time.sleep(random.random()) print("i am a girl") lock.release() def test3(lock): lock.acquire() print("i am wu") time.sleep(random.random()) print("my age is 25") time.sleep(random.random()) print("i am a man") lock.release() if __name__ == '__main__': lock = Lock() p = multiprocessing.Process(target=test,args=(lock,)) p1 = multiprocessing.Process(target=test3,args=(lock,)) p2 = multiprocessing.Process(target=test2,args=(lock,)) p.start() p1.start() p2.start() # i am li # my age is 15 # i am a boy # i am wu # my age is 25 # i am a man # i am lee # my age is 20 # i am a girl
3、互斥锁的应用
import multiprocessing from multiprocessing import Lock import time import json import random def show_ticket(i): time.sleep(random.random()) with open("aa.json","rt",encoding="utf-8") as f: num_dic = json.load(f) time.sleep(random.random()) print("%s正在查看余票，余票数量为%s"%(i,num_dic["num"])) return num_dic["num"] def buy_ticket(i): time.sleep(random.random()) with open("aa.json","rt",encoding="utf-8") as f: num_dic = json.load(f) ticket_num = num_dic["num"] if ticket_num > 0: time.sleep(random.random()) ticket_num -= 1 num_dic["num"] = ticket_num print("%s抢票成功"%i) with open("aa.json","wt",encoding="utf-8") as f: print(num_dic) json.dump(num_dic,f) else: print("%s余票数量为%s,抢票失败"%(i,ticket_num)) def run(i,lock): show_ticket(i) lock.acquire() buy_ticket(i) lock.release() if __name__ == '__main__': lock = Lock() for i in range(10): p = multiprocessing.Process(target=run,args=[i,lock]) p.start() # 执行结果为： # 1正在查看余票，余票数量为1 # 0正在查看余票，余票数量为1 # 4正在查看余票，余票数量为1 # 3正在查看余票，余票数量为1 # 6正在查看余票，余票数量为1 # 5正在查看余票，余票数量为1 # 8正在查看余票，余票数量为1 # 2正在查看余票，余票数量为1 # 7正在查看余票，余票数量为1 # 1抢票成功 # {'num': 0} # 0余票数量为0,抢票失败 # 9正在查看余票，余票数量为1 # 4余票数量为0,抢票失败 # 3余票数量为0,抢票失败 # 6余票数量为0,抢票失败 # 5余票数量为0,抢票失败 # 8余票数量为0,抢票失败 # 2余票数量为0,抢票失败 # 7余票数量为0,抢票失败 # 9余票数量为0,抢票失败
4、IPC
# IPC （inter-process communication）:进程间通信
# 在内存中，进程间的内存是相互独立的，如果想要使用其他进程中的数据此时需要进程间的通信
# 实现进程间通信的方式：
# 1、使用共享文件
# 2、使用共享内存
# 3、使用管道
# 4、使用队列
# 使用共享文件进行进程间通信：由于文件存在于硬盘中，使用该方法进行通信效率较低，较耗时
# 使用subprocess管道进行进程间通信：管道只能进行单向的进程通信
# 使用共享内存进行进程间通信
# 使用共享内存的方法进行进程间通信，由于进程访问的是同一个数据，所以需要加互斥锁，如果不加将会影响数据的安全性
# 在父进程中设置共享内存的内容，如果父进程在子进程还执行时结束，那么就会报错，此时必须设置join方法使父进程在子进程结束后再结束
# 由于使用互斥锁存在死锁的问题，所有尽量少使用这种方式实现进程间通信
from multiprocessing import Manager,Lock import multiprocessing import time import random def fn(dic,lock): time.sleep(random.random()) # lock.acquire() a = dic["a"] a = a-1 time.sleep(random.random()) dic["a"] = a # lock.release() print(dic["a"]) if __name__ == '__main__': dic = {"a":10} lock = Lock() manage = Manager() data = manage.dict({"a":10}) for i in range(10): p = multiprocessing.Process(target=fn,args=[data,lock]) p.start() time.sleep(8) print(data["a"]) # 使用队列(Queue)实现进程间通信 # 在multiprocessing 中封装了Queue类来通过队列实现进程间的通讯 from multiprocessing import Queue q = Queue(4) q.put(1) q.put(2) q.put(3) q.put(4) q.put(5) # 当在队列创建时设置了队列的最大值，那么当队列中存在的值超过队列的最大值时就会处于阻塞状态，直到队列中的值被取出后剩下的数量少于设置的最大值 print(q.get()) print(q.get()) print(q.get()) print(q.get()) # 当管道中的数据被取出后，进程就会处于阻塞态，等待着队列中添加数据
# q.get及q.put中的其它参数
# 1、block(阻塞)：默认为True,如果设置为False,那么在超出队列最大值，或者队列中的值被取完后还在取值时报错
# 2、timeout ： 只有在block 设置为True时才有效，当设置该值后，在进入阻塞态一段时间后才会报错
5、生产者消费者模型
# 生产者消费者模型
import time from multiprocessing import Queue,Process import random def maker(q): for i in range(10): time.sleep(random.random()) print("第%s份食物做好啦..."%i) q.put(i) def eater(q): for t in range(10): i = q.get() time.sleep(random.random()) print(" 第%s份食物吃完啦..."%i) # 在不使用多进程之前，相当于只有一份餐具，我们只能创建一份吃一份食物 # for i in range(10): # maker(i) # eater(i) # 在多进程中，创建食物与吃食物可以一起进行，相当于有好多餐具 if __name__ == '__main__': q = Queue() mp = Process(target=maker,args=[q,]) ep = Process(target=eater,args=[q,]) mp.start() ep.start()