Python中的生产者消费者模型

Posted 2020-12-03 huyingsakai

---恢复内容开始---

了解知识点：

1、守护进程：

·什么是守护进程：

守护进程其实就是一个‘子进程’，守护即伴随，守护进程会伴随主进程的代码运行完毕后而死掉

·为何用守护进程：

当该子进程内的代码在父进程代码运行完毕后就没有存在的意义了，就应该将进程设置为守护进程，会在父进程代码结束后死掉

 

 1 from multiprocessing import Process
 2 
 3 import time,os
 4 
 5 def task(name):
 6     print(‘%s is running‘%name)
 7     time.sleep(3)
 8 
 9 if __name__ == ‘__main__‘:
10     p1=Process(target=task,args=(‘守护进程‘,))
11     p2=Process(target=task,args=(‘正常的子进程‘,))
12     p1.daemon=True  # 一定要放到p.start()之前
13     p1.start()
14     p2.start()
15     print(‘主‘)
16     

守护进程举例

 

以下是守护进程会迷惑人的范例：

 1 #主进程代码运行完毕,守护进程就会结束
 2 from multiprocessing import Process
 3 import time
 4 def foo():
 5     print(123)
 6     time.sleep(1)
 7     print("end123")
 8 
 9 def bar():
10     print(456)
11     time.sleep(3)
12     print("end456")
13 
14 if __name__ == ‘__main__‘:
15     p1=Process(target=foo)
16     p2=Process(target=bar)
17 
18     p1.daemon=True
19     p1.start()
20     p2.start()
21     print("main-------")
22 
23     ‘‘‘
24     main-------
25     456
26     enn456
27     ‘‘‘
28 
29 
30     ‘‘‘
31     main-------
32     123
33     456
34     enn456
35     ‘‘‘
36 
37     ‘‘‘
38     123
39     main-------
40     456
41     end456
42     ‘‘‘

View Code

2、互斥锁：

互斥锁：可以将要执行任务的部分代码（只涉及到修改共享数据的代码）变成串行

join：是要执行任务的所有代码整体串行

强调：必须是lock.acquire()一次，然后 lock.release()释放一次，才能继续lock.acquire()，不能连续的lock.acquire()。否者程序停在原地。

互斥锁vs join： 

大前提：二者的原理都是一样，都是将并发变成串行，从而保证有序（在多个程序共享一个资源时，为保证有序不乱，需将并发变成串行）

区别一：join是按照人为指定的顺序执行，而互斥锁是所以进程平等地竞争，谁先抢到谁执行

区别二：互斥锁可以让一部分代码（修改共享数据的代码）串行，而join只能将代码整体串行（详见抢票系统）

互斥锁

 1 from multiprocessing import Process,Lock
 2 import json
 3 import os
 4 import time
 5 import random
 6 
 7 def check():
 8     time.sleep(1) # 模拟网路延迟
 9     with open(‘db.txt‘,‘rt‘,encoding=‘utf-8‘) as f:
10         dic=json.load(f)
11     print(‘%s 查看到剩余票数 [%s]‘ %(os.getpid(),dic[‘count‘]))
12 
13 def get():
14     with open(‘db.txt‘,‘rt‘,encoding=‘utf-8‘) as f:
15         dic=json.load(f)
16     time.sleep(2)
17     if dic[‘count‘] > 0:
18         # 有票
19         dic[‘count‘]-=1
20         time.sleep(random.randint(1,3))
21         with open(‘db.txt‘,‘wt‘,encoding=‘utf-8‘) as f:
22             json.dump(dic,f)
23         print(‘%s 购票成功‘ %os.getpid())
24     else:
25         print(‘%s 没有余票‘ %os.getpid())
26 
27 
28 def task(mutex):
29     # 查票
30     check()
31 
32     #购票
33     mutex.acquire() # 互斥锁不能连续的acquire，必须是release以后才能重新acquire
34     get()
35     mutex.release()
36 
37 
38 
39     # with mutex:
40     #     get()
41 
42 if __name__ == ‘__main__‘:
43     mutex=Lock()
44     for i in  range(10):
45         p=Process(target=task,args=(mutex,))
46         p.start()
47         # p.join()

模拟抢票

3、IPC通信机制

进程之间通信必须找到一种介质，该介质必须满足

1、是所有进程共享的

2、必须是内存空间

附加：帮我们自动处理好锁的问题

a、   from multiprocessing import Manager（共享内存，但要自己解决锁的问题）

b、   IPC中的队列（Queue） 共享，内存，自动处理锁的问题（最常用）

c、   IPC中的管道（Pipe），共享，内存，需自己解决锁的问题

---恢复内容结束---