Python_守护进程、锁、信号量、事件、队列

1、创建进程

　　守护进程（*****）

　　　　_.daemon = True　　# _进程成为守护进程

　　　　守护进程也是一个子进程。

　　　　主进程的<代码>执行结束之后守护进程自动结束.

 1 import time
 2 from multiprocessing import Process
 3 
 4 def func():
 5     while True:
 6         print('is alive')
 7         time.sleep(0.5)
 8 
 9 def wahaha():
10     i = 0
11     while i < 5:
12         print(f'第{i}秒')
13         i += 1
14 
15 if __name__ == '__main__':
16     Process(target=wahaha).start()
17     p = Process(target=func)
18     p.daemon = True     # 这个p进程就成为守护进程了
19     p.start()
20     time.sleep(3)
21     print('主进程')

守护进程示例

 1 D:Python36python.exe E:/Python/草稿纸.py
 2 第0秒
 3 第1秒
 4 第2秒
 5 第3秒
 6 第4秒
 7 is alive
 8 is alive
 9 is alive
10 is alive
11 is alive
12 is alive
13 主进程
14 
15 Process finished with exit code 0

结果

 1 import time
 2 from multiprocessing import Process
 3 
 4 def func():
 5     while True:
 6         print('is alive')
 7         time.sleep(0.5)
 8 
 9 def wahaha():
10     i = 0
11     while i < 5:
12         print(f'第{i}秒')
13         time.sleep(1)
14         i += 1
15 
16 if __name__ == '__main__':
17     p2 = Process(target=wahaha)
18     p2.start()
19     p = Process(target=func)
20     p.daemon = True     # 这个p进程就成为守护进程了
21     p.start()
22     time.sleep(3)
23     print('主进程')
24     p2.join()       # 守护进程会等待p2子进程结束之后才结束守护

守护进程(join)

D:Python36python.exe E:/Python/草稿纸.py
is alive
第0秒
is alive
is alive
第1秒
is alive
is alive
第2秒
is alive
主进程
is alive
第3秒
is alive
is alive
第4秒
is alive
is alive

Process finished with exit code 0

结果

　　_.is_alive()　　# 判断一个子进程是否活着

　　_.terminate()　　# 结束一个子进程

 1 import time
 2 from multiprocessing import Process
 3 
 4 def wahaha():
 5     i = 0
 6     while i < 5:
 7         print(f'第{i}秒')
 8         time.sleep(1)
 9         i += 1
10 
11 if __name__ == '__main__':
12     p = Process(target=wahaha)
13     p.start()
14     time.sleep(3)
15     print('主进程')
16     print(p.is_alive())     # 判断一个子进程是否活着
17     p.terminate()       # 结束一个子进程
18     print(p.is_alive())     # 判断一个子进程是否活着
19     time.sleep(0.1)
20     print(p.is_alive())

结束子进程与判断子进程活着示例

D:Python36python.exe E:/Python/草稿纸.py
第0秒
第1秒
第2秒
主进程
True
True
False

Process finished with exit code 0

结果

　　在执行结束子进程后,后面的判断子进程是否活着输出True是因为结束子进程需要时间,在结束函数还没执行完成时就执行了下一行判断,所以输出True,在沉睡0.1秒后结束进程已经执行结束,子进程结束,输出False.

2、进程同步部分

　　（1）锁（*****）

　　　　为了避免同一段代码被多个进程同时执行.

　　　　维护数据的安全.

　　　　降低了程序的效率.

　　　　所有的效率都是建立在数据安全的角度上的.

　　　　但凡涉及到并发编程都要考虑数据啥的安全性.

　　　　我们需要在并发部分对数据修改的操作格外小心,如果涉及到数据的不安全,就需要进行加锁控制.

　　　　lock / acquire / release的另外一种用法:

　　　　　　lock 内部实现了进程之间的通信,使得谁acquire了谁release了能够在多个拥有lock参数的子程序中透明.

1 from multiprocessing import Lock
2 
3 lock = Lock()
4 lock.acquire()  # 想拿钥匙
5 print(1)
6 lock.release()  # 还钥匙
7 lock.acquire()  # 想拿钥匙
8 print('拿到钥匙了')
9 lock.release()  # 还钥匙

锁_示例

D:Python36python.exe E:/Python/草稿纸.py
1
拿到钥匙了

Process finished with exit code 0

结果

　　查票的例子

 1 import json
 2 import time
 3 from multiprocessing import Process, Lock
 4 
 5 def search(i):
 6     with open('db') as f:
 7         ticket_dic = json.load(f)
 8     print(f'{i}正在查票,剩余票数:{ticket_dic["count"]}')
 9 
10 def buy(i):
11     with open('db') as f:
12         ticket_dic = json.load(f)
13     time.sleep(0.2)     # 模拟请求数据库的网络延时
14     if ticket_dic['count'] > 0:
15         ticket_dic['count'] -= 1
16         print(f'{i}买到票了')
17         time.sleep(0.2)     # 模拟网数据库写消息的网络延迟
18         with open('db', 'w') as f:
19             json.dump(ticket_dic, f)
20 
21 def get_ticket(i, lock):
22     search(i)       # 查询余票,可以多个用户同时查询
23     with lock:
24         buy(i)      # 买票的操作涉及到数据修改,需要加锁来保证数据的安全
25 
26 if __name__ == '__main__':
27     lock = Lock()
28     for i in range(10):
29         p = Process(target=get_ticket, args=(i, lock))
30         p.start()

文件管理版

 1 D:Python36python.exe E:/Python/草稿纸.py
 2 1正在查票,剩余票数:3
 3 2正在查票,剩余票数:3
 4 3正在查票,剩余票数:3
 5 0正在查票,剩余票数:3
 6 4正在查票,剩余票数:3
 7 1买到票了
 8 6正在查票,剩余票数:3
 9 5正在查票,剩余票数:3
10 9正在查票,剩余票数:3
11 7正在查票,剩余票数:3
12 8正在查票,剩余票数:3
13 2买到票了
14 3买到票了
15 
16 Process finished with exit code 0

结果

 1 import json
 2 import time
 3 from multiprocessing import Process, Lock
 4 
 5 def search(i):
 6     with open('db') as f:
 7         ticket_dic = json.load(f)
 8         print(f'{i}正在查票,剩余票数:{ticket_dic["count"]}')
 9 
10 def buy(i):
11     with open('db') as f:
12         tick_dic = json.load(f)
13     time.sleep(0.2)     # 模拟请求数据库的网络延时
14     if tick_dic['count'] > 0:
15         tick_dic['count'] -= 1
16         print(f'{i}买到票了')
17         time.sleep(0.2)     # 模拟往数据库写消息的网络延时
18         with open('db', 'w') as f:
19             json.dump(tick_dic, f)
20 
21 def get_ticket(i, lock):
22     search(i)       # 查询余票,可以多个用户同时查询
23     lock.acquire()
24     buy(i)      # 买票的操作涉及到数据修改,需要加锁来保证数据的安全
25     lock.release()
26 
27 if __name__ == '__main__':
28     lock = Lock()
29     for i in range(10):
30         p = Process(target=get_ticket, args=(i, lock))
31         p.start()

正常版

D:Python36python.exe E:/Python/草稿纸.py
0正在查票,剩余票数:3
1正在查票,剩余票数:3
3正在查票,剩余票数:3
2正在查票,剩余票数:3
0买到票了
4正在查票,剩余票数:3
6正在查票,剩余票数:3
5正在查票,剩余票数:3
7正在查票,剩余票数:3
8正在查票,剩余票数:3
9正在查票,剩余票数:3
1买到票了
3买到票了

Process finished with exit code 0

结果

　　（2）信号量（***）

　　信号量是由 lock 加计数器组成的.

 1 from multiprocessing import Semaphore
 2 
 3 sem = Semaphore(4)
 4 sem.acquire()
 5 print(1)
 6 sem.acquire()
 7 print(2)
 8 sem.acquire()
 9 print(3)
10 sem.acquire()
11 print(4)
12 sem.release()
13 print(5)

信号量

D:Python36python.exe E:/Python/草稿纸.py
1
2
3
4
5

Process finished with exit code 0

结果

　　商场KTV小例:

 1 import time
 2 import random
 3 from multiprocessing import Process, Semaphore
 4 
 5 def ktv(sem, i):
 6     sem.acquire()
 7     print(f'{i}走进KTV')
 8     time.sleep(random.randint(1, 5))
 9     print(f'{i}走出了KTV')
10     sem.release()
11 
12 if __name__ == "__main__":
13     sem = Semaphore(4)
14     for i in range(10):
15         p = Process(target=ktv, args=(sem, i))
16         p.start()

正常版

D:Python36python.exe E:/Python/草稿纸.py
1走进KTV
3走进KTV
0走进KTV
2走进KTV
1走出了KTV
7走进KTV
2走出了KTV
5走进KTV
0走出了KTV
4走进KTV
7走出了KTV
9走进KTV
3走出了KTV
6走进KTV
6走出了KTV
8走进KTV
5走出了KTV
4走出了KTV
9走出了KTV
8走出了KTV

Process finished with exit code 0

结果

 1 import time
 2 import random
 3 from multiprocessing import Process, Semaphore
 4 
 5 def ktv(sem, i):
 6     with sem:
 7         print(f'{i}走进了KTV')
 8         time.sleep(random.randint(1, 5))
 9         print(f'{i}走出了KTV')
10 
11 if __name__ == "__main__":
12     sem = Semaphore(4)
13     for i in range(10):
14         p = Process(target=ktv, args=(sem, i))
15         p.start()

文件管理版

D:Python36python.exe E:/Python/草稿纸.py
0走进了KTV
2走进了KTV
1走进了KTV
3走进了KTV
0走出了KTV
4走进了KTV
2走出了KTV
6走进了KTV
3走出了KTV
5走进了KTV
4走出了KTV
8走进了KTV
1走出了KTV
9走进了KTV
8走出了KTV
7走进了KTV
9走出了KTV
7走出了KTV
6走出了KTV
5走出了KTV

Process finished with exit code 0

结果

　　（3）事件（**）

　　　　 from multiprocessing import Event

　　　　控制子进程执行还是一个阻塞机制 —— 事件

　　　　wait 方法：

　　　　　　在事件中还有一个标志，

　　　　　　如果这个标志是True，wait方法的执行效果就是pass，

　　　　　　如果这个标志是False，wait方法的效果就是阻塞，直到这个标志变成True。

　　　　控制标志：

　　　　　　判断标志的状态 is_set，

　　　　　　set方法将标志设置为True，

　　　　　　clear方法将标志设置为False。

　　　　事件创建之初标志的状态是False。

1 from multiprocessing import Event
2 e = Event()
3 print(e.is_set())
4 e.wait()

事件默认标志的状态为False

D:Python36python.exe E:/Python/草稿纸.py
False

结果（状态阻塞）

1 from multiprocessing import Event
2 e = Event()
3 e.set()
4 print(e.is_set())
5 e.wait()

改变标志位为True

D:Python36python.exe E:/Python/草稿纸.py
True

Process finished with exit code 0

结果

　　wait（）设置等待时间。但是不改变标志状态。

 1 import time
 2 from multiprocessing import Event, Process
 3 
 4 def func1(e):
 5     print('start func1')
 6     e.wait(1)   # wait()可以设置等待的时间,当等待时间结束,程序继续往下运行,标志维持原状态
 7     print(e.is_set())
 8     print('end func1')
 9 
10 if __name__ == '__main__':
11     e = Event()
12     Process(target=func1, args=(e,)).start()
13     time.sleep(5)
14     e.set()

wait（）设置等待时间

D:Python36python.exe E:/Python/草稿纸.py
start func1
False
end func1

Process finished with exit code 0

结果

　　　　红绿灯

 1 import time
 2 import random
 3 from multiprocessing import Process, Event
 4 
 5 def traffic_light(e):
 6     print('33[1;31m红灯亮33[0m')
 7     while True:
 8         time.sleep(2)
 9         if e.is_set():
10             print('33[1;31m红灯亮33[0m')
11             e.clear()
12         else:
13             print('33[1;32m绿灯亮33[0m')
14             e.set()
15 
16 def car(i, e):
17     if not e.is_set():
18         print(f'car{i}正在等待通过')
19         e.wait()
20     print(f'car{i}通过')
21 
22 if __name__ == '__main__':
23     e = Event()
24     light = Process(target=traffic_light, args=(e,))
25     light.daemon = True
26     light.start()
27     car_list = []
28     for i in range(20):
29         p = Process(target=car, args=(i,e))
30         p.start()
31         time.sleep(random.randint(0,3))
32         car_list.append(p)
33 
34     for car in car_list:
35         car.join()

示例

D:Python36python.exe E:/Python/草稿纸.py
红灯亮
car0正在等待通过
绿灯亮
car0通过
car1通过
红灯亮
car2正在等待通过
绿灯亮
car2通过
红灯亮
car3正在等待通过
car4正在等待通过
car5正在等待通过
绿灯亮
car3通过
car4通过
car5通过
car6通过
car7通过
红灯亮
car8正在等待通过
绿灯亮
car8通过
红灯亮
car9正在等待通过
绿灯亮
car9通过
car10通过
红灯亮
car11正在等待通过
绿灯亮
car11通过
car12通过
car13通过
红灯亮
绿灯亮
car14通过
红灯亮
car15正在等待通过
绿灯亮
car15通过
car16通过
红灯亮
car17正在等待通过
绿灯亮
car17通过
car18通过
car19通过
红灯亮
绿灯亮

Process finished with exit code 0

结果

3、进程之间数据共享（部分）

　　队列（*****）

　　IPC 进程之间的通信

 1 from multiprocessing import Queue
 2 
 3 q = Queue()
 4 q.put(1)
 5 q.put('aaa')
 6 q.put([1, 2, 3])
 7 q.put({'k': 'v'})
 8 q.put({'k', 'v'})
 9 print(q.get())
10 print(q.get())
11 print(q.get())
12 print(q.get())
13 print(q.get())

队列可以存放的数据类型

　　队列可以设置大小，当队列满了以后就会阻塞。

 1 from multiprocessing import Queue
 2 
 3 q = Queue(5)
 4 q.put(1)
 5 q.put('aaa')
 6 q.put([1, 2, 3])
 7 q.put({'k': 'v'})
 8 q.put({'k', 'v'})
 9 print(q.empty())
10 print(q.full())
11 q.put((1, 2, 3))

设置队列大小

D:Python36python.exe E:/Python/草稿纸.py
False
True

结果（阻塞）

　　当队列中没有值时，get（）取不到值，就会阻塞，直到队列里取到值。

1 from multiprocessing import Queue
2 q = Queue(3)
3 q.get()

示例

D:Python36python.exe E:/Python/草稿纸.py

结果（进入阻塞）

　　当队列中没有值时，使用get_nowait()取不到值就会报错。

1 from multiprocessing import Queue
2 q = Queue(3)
3 try:
4     q.get_nowait()
5 except:
6     print('队列中没有值')

get_nowait()

1 D:Python36python.exe E:/Python/草稿纸.py
2 队列中没有值
3 
4 Process finished with exit code 0

结果

　　当队列满了以后在put（）数据进去就会进入阻塞。

1 from multiprocessing import Queue
2 
3 q = Queue(3)
4 q.put(1)
5 q.put('aaa')
6 q.put([1, 2, 3])
7 q.put('alex')

队列满进入阻塞

D:Python36python.exe E:/Python/草稿纸.py

结果（进入阻塞）

　　当队列满了后使用put_nowait() 不会报错，但是put_nowait()的数据会丢失。

 1 from multiprocessing import Queue
 2 
 3 q = Queue(3)
 4 q.put(1)
 5 q.put('aaa')
 6 q.put([1, 2, 3])
 7 try:
 8     q.put_nowait('alex')
 9 except:
10     print('丢了一个数据')

示例

D:Python36python.exe E:/Python/草稿纸.py
丢了一个数据

Process finished with exit code 0

结果

　　多进程的情况下不准：

　　　　empty

　　　　full

　　会引起程序的阻塞：

　　　　get

　　　　put

　　不会引起程序的阻塞，但是程序会报错：

　　　　get_nowait

　　　　put_nowait　　# 容易丢失数据.

　　获得10的10次方:

from multiprocessing import Process, Queue

def  func(num, q):
    q.put({num:num**num})

if __name__ == '__main__':
    q = Queue()
    p = Process(target=func, args=(10, q))
    p.start()
    print(q.get())

10**10

D:Python36python.exe E:/Python/草稿纸.py
{10: 10000000000}

Process finished with exit code 0

结果

　　计算 0 到 9 的对应次方

 1 from multiprocessing import Process, Queue
 2 
 3 def  func(num, q):
 4     q.put({num:num**num})
 5 
 6 if __name__ == '__main__':
 7     q = Queue()
 8     for i in range(10):
 9         p = Process(target=func, args=(i, q))
10         p.start()
11     for i in range(10):
12         print(q.get())

0到9的对应次方

D:Python36python.exe E:/Python/草稿纸.py
{0: 1}
{2: 4}
{3: 27}
{1: 1}
{4: 256}
{6: 46656}
{7: 823543}
{8: 16777216}
{5: 3125}
{9: 387420489}

Process finished with exit code 0

结果