【2.0】知识点小结（进程进阶）#

【一】进程对象及其他方法#

【1】查看当前进程的进程号#

一台计算机上面运行着很多进程，那么计算机是如何区分并管理这些进程服务端呢？

计算机会给每一个运行的进程分配一个PID号

如何查看？

Windows系统

CMD 命令行 tasklist 即可查看

Mac系统

终端运行 ps aux 即可查看

如何根据指定进程号查看进程

Mac系统

终端运行 ps aux|grep PID 即可查看

Windows系统

CMD 命令行 tasklist |findstr PID 即可查看

（1）查看当前进程的进程号`current_process().pid` 方法#

1
# -*-coding: Utf-8 -*-
2
# @File : 01 进程对象及其他方法 .py
3
# author: Chimengmeng
4
# blog_url : https://www.cnblogs.com/dream-ze/
5
# Time：2023/6/24
6
from multiprocessing import Process, current_process
7
import os
8
import time
9

10

11
def task():
12
    # 查看当前进程的 进程（PID） 号
13
    print(f'当前程序:>>>>{current_process().pid} 正在运行')
14
    time.sleep(2)
15

16

17
if __name__ == '__main__':
18
    p = Process(target=task)
19
    p.start()
20

21
    print(f'这是主程序:>>>{current_process().pid}')
22

23
    # 这是主程序:>>>11168
24
    # 当前程序:>>>>10944 正在运行

（2）查看当前进程的进程号`os.getpid()` 方法#

1
# -*-coding: Utf-8 -*-
2
# @File : 01 进程对象及其他方法 .py
3
# author: Chimengmeng
4
# blog_url : https://www.cnblogs.com/dream-ze/
5
# Time：2023/6/24
6
from multiprocessing import Process, current_process
7
import os
8
import time
9

10

11
def task():
12
    # 查看当前进程的 进程（PID） 号
13
    print(f'当前程序:>>>>{os.getpid()} 正在运行')
14
    time.sleep(2)
15

16

17
if __name__ == '__main__':
18
    p = Process(target=task)
19
    p.start()
20

21
    print(f'这是主程序:>>>{os.getpid()}')
22

23
    # 这是主程序:>>>9928
24
    # 当前程序:>>>>3912 正在运行

（3）查看当前进程的父进程的进程号`os.getpid()` 方法#

1
# -*-coding: Utf-8 -*-
2
# @File : 01 进程对象及其他方法 .py
3
# author: Chimengmeng
4
# blog_url : https://www.cnblogs.com/dream-ze/
5
# Time：2023/6/24
6
from multiprocessing import Process, current_process
7
import os
8
import time
9

10

11
def task():
12
    # 查看当前进程的 进程（PID） 号
13
    print(f'当前程序:>>>>{os.getpid()} 正在运行')
14
    # 查看当前进程的 父进程（PID） 号
15
    print(f'当前程序的父进程:>>>>{os.getppid()} 正在运行')
16
    time.sleep(2)
17

18

19
if __name__ == '__main__':
20
    p = Process(target=task)
21
    p.start()
22

23
    print(f'这是主程序:>>>{os.getpid()}')
24
    print(f'这是主程序的父进程:>>>{os.getppid()}')
25

26
    # 这是主程序:>>>22236
27
    # 这是主程序的父进程:>>>17720
28
    # 当前程序:>>>>3756 正在运行
29
    # 当前程序的父进程:>>>>22236 正在运行

【2】杀死当前进程`p.terminate()`#

告诉操作系统帮我去杀死当前进程
但是需要一定的时间。
代码的运行速度极快

【3】判断当前进程是否存活`p.is_alive()`#

1
# -*-coding: Utf-8 -*-
2
# @File : 01 进程对象及其他方法 .py
3
# author: Chimengmeng
4
# blog_url : https://www.cnblogs.com/dream-ze/
5
# Time：2023/6/24
6
from multiprocessing import Process, current_process
7
import os
8
import time
9

10

11
def task():
12
    # 查看当前进程的 进程（PID） 号
13
    print(f'当前程序:>>>>{os.getpid()} 正在运行')
14
    print(f'当前程序的父进程:>>>>{os.getppid()} 正在运行')
15

16
    time.sleep(2)
17

18

19
if __name__ == '__main__':
20
    p = Process(target=task)
21
    p.start()
22

23
    # 杀死当前进程 - 需要给操作系统一点缓冲时间
24
    p.terminate()
25
    time.sleep(0.2)
26
    # 判断当前进程是否存活
27
    print(p.is_alive())
28

29
    print(f'这是主程序:>>>{os.getpid()}')
30
    print(f'这是主程序的父进程:>>>{os.getppid()}')
31

32
    # 主进程正常
33
    # 调用程序中的子进程被杀死
34
    # False
35
    # 这是主程序:>>>16796
36
    # 这是主程序的父进程:>>>17720

一般默认会将
存储布尔值的变量名
返回的结果是布尔值的方法名
起成is_ 开头的变量名

【二】僵尸进程与孤儿进程#

【1】僵尸进程#

僵尸进程是指完成了自己的任务,但父进程没有正确地释放它所占用的系统资源
导致它仍然存在于进程列表中
但已经停止了运行。
这些僵尸进程会占据一定的系统内存，并在一定程度上影响系统的性能。

当子进程开设后，该进程死后不会立刻释放占用的进程号
- 因为要让父进程能够查看到开设的子进程的一些基本信息
  - 占用的 PID 号，运行时间等
所有的进程都会步入僵尸进程
- 父进程不死并且在无限制的创建子进程并且子进程也不结束
如何回收子进程占用的 PID 号
- 父进程等待子进程运行结束
- 父进程调用 join 方法

1
# -*-coding: Utf-8 -*-
2
# @File : 02 僵尸进程 .py
3
# author: Chimengmeng
4
# blog_url : https://www.cnblogs.com/dream-ze/
5
# Time：2023/6/24
6

7
from multiprocessing import Process
8
import time
9

10

11
def run():
12
    print(f'the run is beginning:>>>>')
13
    time.sleep(1)
14
    print(f'the run is ending:>>>>')
15

16

17
if __name__ == '__main__':
18
    p = Process(target=run)
19

20
    p.start()
21

22
    print(f'这是主程序的执行方法:>>>>')
23
    # 这是主程序的执行方法:>>>>
24
    # the run is beginning:>>>>
25
    # the run is ending:>>>>

【2】孤儿进程#

孤儿进程则是指父进程在子进程终止之前就已经退出了
导致子进程失去了与父进程通信的能力。
这些孤儿进程将被init进程接管
init进程会等待它的状态信息并释放它的系统资源。

子进程存活，父进程意外死亡
- 没有父进程来帮助回收 PID 号
解决办法
- 操作系统会开设一个儿童福利院(init 进程)专门管理孤儿进程回收相关资源

【三】守护进程#

守护进程 (daemon) 是在计算机系统启动时就已经运行，并且一直在后台运行的一类特殊进程。
它们通常不与用户直接交互，也不接受标准输入和输出，而是在后台执行某种任务或提供某种服务。
守护进程往往是由系统管理员手动启动的，它们可以在系统启动时自动启动，一直运行在后台，直到系统关闭或被停止。
常见的守护进程包括网络服务 (如 web 服务器、邮件服务器、 ftp 服务器等)、日志记录系统 (如系统日志服务、应用程序日志服务等) 等。
守护进程通常在后台运行，不需要用户交互，并且有较高的权限，因此编写守护进程需要特别注意安全性和稳定性。

【1】引入 - 主进程死亡，子进程未死亡#

1
# -*-coding: Utf-8 -*-
2
# @File : 03 守护进程 .py
3
# author: Chimengmeng
4
# blog_url : https://www.cnblogs.com/dream-ze/
5
# Time：2023/6/24
6

7
from multiprocessing import Process
8
import time
9

10

11
def task(name):
12
    print(f'总管:>>{name}>>正常存活')
13
    time.sleep(2)
14
    print(f'总管:>>{name}>>正常死亡')
15

16

17
if __name__ == '__main__':
18
    print(f'皇帝 :>>> ChiMeng >>> 执掌江山')
19
    p = Process(target=task, args=('dream',))
20
    # p = Process(target=task, kwargs={"name":"dream"})
21

22
    p.start()
23

24
    print(f'皇帝 :>>> ChiMeng >>> 寿终正寝')
25

26
    # 皇帝 :>>> ChiMeng >>> 执掌江山
27
    # 皇帝 :>>> ChiMeng >>> 寿终正寝
28
    # 总管:>>dream>>正常存活
29
    # 总管:>>dream>>正常死亡

【2】使用 - 主进程死亡，子进程必死亡#

（1）错误使用演示#

1
# -*-coding: Utf-8 -*-
2
# @File : 03 守护进程 .py
3
# author: Chimengmeng
4
# blog_url : https://www.cnblogs.com/dream-ze/
5
# Time：2023/6/24
6

7
from multiprocessing import Process
8
import time
9

10

11
def task(name):
12
    print(f'总管:>>{name}>>正常存活')
13
    time.sleep(2)
14
    print(f'总管:>>{name}>>正常死亡')
15

16

17
if __name__ == '__main__':
18
    print(f'皇帝 :>>> ChiMeng >>> 执掌江山')
19
    p = Process(target=task, args=('dream',))
20
    # p = Process(target=task, kwargs={"name":"dream"})
21

22
    p.start()
23

24
    # 将进程 p 设置成守护进程
25
    p.daemon = True
26

27
    print(f'皇帝 :>>> ChiMeng >>> 寿终正寝')
28

29
    #     assert self._popen is None, 'process has already started'
30
    # AssertionError: process has already started
31
    # 报错原因：进程已经开始了无法再次重置进程状态
32

33
    # 总管:>>dream>>正常存活
34
    # 总管:>>dream>>正常死亡

（2）正确使用演示#

将设置主进程的方法在调用主进程的前面声明

1
# -*-coding: Utf-8 -*-
2
# @File : 03 守护进程 .py
3
# author: Chimengmeng
4
# blog_url : https://www.cnblogs.com/dream-ze/
5
# Time：2023/6/24
6

7
from multiprocessing import Process
8
import time
9

10

11
def task(name):
12
    print(f'总管:>>{name}>>正常存活')
13
    time.sleep(2)
14
    print(f'总管:>>{name}>>正常死亡')
15

16

17
if __name__ == '__main__':
18
    print(f'皇帝 :>>> ChiMeng >>> 执掌江山')
19
    p = Process(target=task, args=('dream',))
20
    # p = Process(target=task, kwargs={"name":"dream"})
21

22
    # 将进程 p 设置成守护进程
23
    p.daemon = True
24

25
    p.start()
26

27
    print(f'皇帝 :>>> ChiMeng >>> 寿终正寝')
28

29
    # 皇帝 :>>> ChiMeng >>> 执掌江山
30
    # 皇帝 :>>> ChiMeng >>> 寿终正寝

【四】互斥锁#

互斥锁（Mutex）是一种用于多线程编程中控制对共享资源访问的机制。
其作用是保证在同一时刻只有一个线程在访问共享资源，从而避免多个线程同时读写数据造成的问题。
互斥锁的基本原理是在对共享资源进行访问前加锁，使得其他线程无法访问该资源，当访问完成后再解锁，使得其他线程可以进行访问。
通过这种方式，可以保证同一时间只有一个线程在执行关键代码段，从而保证了数据的安全性。
需要注意的是，互斥锁会带来一些额外的开销，

【1】数据错乱问题#

多个进程操作同一份数据的时候，会出现数据错乱的问题

1
# -*-coding: Utf-8 -*-
2
# @File : 04 互斥锁 .py
3
# author: Chimengmeng
4
# blog_url : https://www.cnblogs.com/dream-ze/
5
# Time：2023/6/24
6

7
from multiprocessing import Process
8
import json
9
import time
10
import random
11

12

13
# 拿到票的数据
14
def get_ticket():
15
    with open('ticket_data', 'r', encoding='utf8') as f:
16
        ticket_dict = json.load(f)
17
    return ticket_dict
18

19

20
# 存储票的数据
21
def save_ticket(ticket_dict):
22
    with open('ticket_data', 'w', encoding='utf8') as f:
23
        json.dump(ticket_dict, f)
24

25

26
# （1）查票
27
def search_ticket(name):
28
    # 模拟从数据库读取票数
29
    # 打开文件读取票数
30
    ticket_dict = get_ticket()
31
    print(f'用户:>>>{name} 正在查询余票:>>>{ticket_dict.get("ticket_number")}')
32

33

34
# （2）买票
35
def buy_ticket(name):
36
    # （2.1）先查票
37
    ticket_dict = get_ticket()
38

39
    # （2.2）模拟网络延迟
40
    time.sleep(random.randint(1, 3))
41

42
    # （2.3） 判断当前是否有票
43
    if ticket_dict.get('ticket_number') > 0:
44
        # （2.5）有票再买票
45
        ticket_dict['ticket_number'] -= 1
46
        save_ticket(ticket_dict)
47
        print(f'用户:>>>{name} 买票成功!!')
48
    else:
49
        # （2.4） 没票则反馈信息
50
        print(f'当前无余票!!')
51

52

53
# 整合成一个人的功能
54
def main(name):
55
    search_ticket(name)
56
    buy_ticket(name)
57

58

59
if __name__ == '__main__':
60
    # 多进程演示 - 多个人购票
61
    for i in range(1, 5):
62
        p = Process(target=main, args=(i,))
63
        p.start()
64

65
    # 所有人都能查票成功，且能购票成功
66
    # 原因是所有人都查票的时候，数据发生了错乱
67
    # 用户:>>>1 正在查询余票:>>>2
68
    # 用户:>>>2 正在查询余票:>>>2
69
    # 用户:>>>3 正在查询余票:>>>2
70
    # 用户:>>>4 正在查询余票:>>>2
71
    # 用户:>>>2 买票成功!!
72
    # 用户:>>>4 买票成功!!
73
    # 用户:>>>1 买票成功!!
74
    # 用户:>>>3 买票成功!!

【2】加锁处理#

针对上述数据错乱问题，解决方式就是加锁处理

将并发变成串行牺牲效率，但是保证了数据的安全

1
# -*-coding: Utf-8 -*-
2
# @File : 04 互斥锁 .py
3
# author: Chimengmeng
4
# blog_url : https://www.cnblogs.com/dream-ze/
5
# Time：2023/6/24
6

7
from multiprocessing import Process, Lock
8
import json
9
import time
10
import random
11

12

13
# 拿到票的数据
14
def get_ticket():
15
    with open('ticket_data', 'r', encoding='utf8') as f:
16
        ticket_dict = json.load(f)
17
    return ticket_dict
18

19

20
# 存储票的数据
21
def save_ticket(ticket_dict):
22
    with open('ticket_data', 'w', encoding='utf8') as f:
23
        json.dump(ticket_dict, f)
24

25

26
# （1）查票
27
def search_ticket(name):
28
    # 模拟从数据库读取票数
29
    # 打开文件读取票数
30
    ticket_dict = get_ticket()
31
    print(f'用户:>>>{name} 正在查询余票:>>>{ticket_dict.get("ticket_number")}')
32

33

34
# （2）买票
35
def buy_ticket(name):
36
    # （2.1）先查票
37
    ticket_dict = get_ticket()
38

39
    # （2.2）模拟网络延迟
40
    time.sleep(random.randint(1, 3))
41

42
    # （2.3） 判断当前是否有票
43
    if ticket_dict.get('ticket_number') > 0:
44
        # （2.5）有票再买票
45
        ticket_dict['ticket_number'] -= 1
46
        save_ticket(ticket_dict)
47
        print(f'用户:>>>{name} 买票成功!!')
48
    else:
49
        # （2.4） 没票则反馈信息
50
        print(f'用户:>>>{name} 当前无余票!!')
51

52

53
# 整合成一个人的功能
54
def main(name, mutex):
55
    # 所有人都去查票
56
    search_ticket(name)
57

58
    # 给买票环节加锁处理
59
    # （1）先强锁
60
    mutex.acquire()
61
    buy_ticket(name)
62
    # （2）抢完后释放锁
63
    mutex.release()
64

65

66
if __name__ == '__main__':
67
    # 在主进程中生成一把锁,让所有的子进程去抢，谁抢到谁就先买票
68
    mutex = Lock()
69

70
    # 多进程演示 - 多个人购票
71
    for i in range(1, 5):
72
        p = Process(target=main, args=(i, mutex))
73
        p.start()
74

75
    # 所有人都能查票成功，但是只有前两名购票成功
76
    # 前两名是随机抢到的，不一定是按顺序（考虑到各种因素，如网络等）
77
    # 用户:>>>1 正在查询余票:>>>2
78
    # 用户:>>>2 正在查询余票:>>>2
79
    # 用户:>>>3 正在查询余票:>>>2
80
    # 用户:>>>4 正在查询余票:>>>2
81
    # 用户:>>>1 买票成功!!
82
    # 用户:>>>2 买票成功!!
83
    # 用户:>>>3 当前无余票!!
84
    # 用户:>>>4 当前无余票!!

【3】行锁与表锁#

行锁和表锁是数据库中常用的锁定机制。

（1）行锁#

行锁是在对数据库表中的某个数据行进行修改时
- 一次只允许一个用户操作该行
- 其他用户如果需要修改该行数据就必须等待。
通过行锁定可以避免多个用户同时修改同一行数据所导致的数据不一致问题。

（2）表锁#

表锁则是当一个用户操作某个数据库表时
- 会锁定整个表，其他用户同时不能操作该表。
这在一些特殊场景下比如表维护、备份等是非常有用的。

（3）小结#

总的来说
- 行锁定是比较细粒度的锁定
- 而表锁定则是更为粗粒度的锁定方法。

【4】注意#

锁不要轻易使用，容易造成死锁现象
锁只在处理数据的部分加，用来保证数据的安全（只在争抢数据的环节加锁）

【五】进程间通信#

借助于：消息队列

管道：subprocess
stdin stdout stderr
不同类型的信息分别存储在不同的管道内
当我们进行read方法后，其他进程无法再拿到指定管道内的其他消息

队列：管道 + 锁

队列：先进先出
堆栈：后进后出

存取数据：存是为了更好的取
千方百计的存
简单快捷的取

【1】队列模块：Queue#

（1）简单使用#

1
# -*-coding: Utf-8 -*-
2
# @File : 05 队列模块 .py
3
# author: Chimengmeng
4
# blog_url : https://www.cnblogs.com/dream-ze/
5
# Time：2023/6/24
6
import queue
7
from multiprocessing import Queue
8

9
# 创建队列
10
# 括号内可以传参数，表示生成的队列最大可以同时存放的数据量
11
# 默认 为 0
12
q = queue.Queue(5)
13

14
# 向队列中放数据
15
q.put(111)
16

17
# 在队列中取数据
18
res_get = q.get()
19
print(res_get)
20
# 111

当队列数据放满了以后，如果还有数据要放程序就会阻塞，直到有位置让出来（不会报错）
队列中如果已经没有数据的话，程序也会阻塞

（2）没有数据抛出异常`.get_nowait()`#

队列中没有数据直接抛出异常

1
# -*-coding: Utf-8 -*-
2
# @File : 05 队列模块 .py
3
# author: Chimengmeng
4
# blog_url : https://www.cnblogs.com/dream-ze/
5
# Time：2023/6/24
6
import queue
7
from multiprocessing import Queue
8

9
# 创建队列
10
# 括号内可以传参数，表示生成的队列最大可以同时存放的数据量
11
# 默认 为 0
12
q = queue.Queue(5)
13

14
# 向队列中放数据
15
q.put(111)
16
q.put(222)
17

18
# 在队列中取数据
19
res_get = q.get()
20
res_get1 = q.get()
21
res_get2 = q.get_nowait()
22

23
print(res_get)
24

25
# raise Empty
26
# _queue.Empty

（3）没有数据延迟抛出异常`.get(timeout=3)`#

队列中没有数据，延迟三秒抛出异常

1
# -*-coding: Utf-8 -*-
2
# @File : 05 队列模块 .py
3
# author: Chimengmeng
4
# blog_url : https://www.cnblogs.com/dream-ze/
5
# Time：2023/6/24
6
import queue
7
from multiprocessing import Queue
8

9
# 创建队列
10
# 括号内可以传参数，表示生成的队列最大可以同时存放的数据量
11
# 默认 为 0
12
q = queue.Queue(5)
13

14
# 向队列中放数据
15
q.put(111)
16
q.put(222)
17

18
# 在队列中取数据
19
res_get = q.get()
20
res_get1 = q.get()
21
res_get2 = q.get(timeout=3)
22

23
print(res_get)
24
# raise Empty
25
# _queue.Empty

（4）判断当前队列数据是否填满队列`.full()`#

1
# -*-coding: Utf-8 -*-
2
# @File : 05 队列模块 .py
3
# author: Chimengmeng
4
# blog_url : https://www.cnblogs.com/dream-ze/
5
# Time：2023/6/24
6
import queue
7
from multiprocessing import Queue
8

9
# 创建队列
10
# 括号内可以传参数，表示生成的队列最大可以同时存放的数据量
11
# 默认 为 0
12
q = queue.Queue(3)
13

14
# 向队列中放数据
15
q.put(111)
16
print(q.full())
17
# False
18
q.put(222)
19
print(q.full())
20
# False
21
q.put(333)
22
print(q.full())
23
# True
24

25

26
# 在队列中取数据
27
res_get = q.get()
28
res_get1 = q.get()

（5）判断当前队列是否为空`.empty()`#

1
# -*-coding: Utf-8 -*-
2
# @File : 05 队列模块 .py
3
# author: Chimengmeng
4
# blog_url : https://www.cnblogs.com/dream-ze/
5
# Time：2023/6/24
6
import queue
7
from multiprocessing import Queue
8

9
# 创建队列
10
# 括号内可以传参数，表示生成的队列最大可以同时存放的数据量
11
# 默认 为 0
12
q = queue.Queue(3)
13

14
# 向队列中放数据
15
q.put(111)
16

17
# 判断当前队列数据是否填满队列
18
print(q.full())
19
# False
20
q.put(222)
21
print(q.full())
22
# False
23
q.put(333)
24

25
# 判断当前队列数据是否填满队列
26
print(q.full())
27
# True
28

29

30
# 在队列中取数据
31
res_get = q.get()
32
res_get1 = q.get()
33

34
# 判断当前队列是够为空
35
print(q.empty())
36
# False
37

38
res_get3 = q.get()
39
# 判断当前队列是够为空
40
print(q.empty())
41
# True

（6）小结#

1
# 判断当前队列是否满了
2
q.full()
3

4
# 判断当前队列是够空了
5
q.empty()
6

7
# 判断当前队列是否为空
8
# 如果队列为空，该方法将引发 queue.Empty 异常
9
q.get_nowait()

以上方法在多进程的情况下不精确
多进程的队列切换时间是非常短的，存在可能判断失误

【2】IPC机制#

IPC机制指进程间通信机制（Inter-Process Communication），它是指在不同进程间传输数据或者信息的一种机制。
在多进程操作系统中，各个进程相互独立，不能直接访问对方的内存空间，所以必须通过特殊的通信方式实现进程之间的信息交换和协调。
常见的IPC机制包括管道、消息队列、共享内存和信号量等方式。
其中，管道、消息队列和共享内存都是用于进程之间的数据传输
而信号量则是用于进程之间的同步与互斥。
通过使用IPC机制，可以使得不同进程能够有效地协同工作，在复杂的程序设计中起到十分重要的作用。

（1）子进程与主进程之间通过队列进行通信#

1
# -*-coding: Utf-8 -*-
2
# @File : 06 IPC机制 .py
3
# author: Chimengmeng
4
# blog_url : https://www.cnblogs.com/dream-ze/
5
# Time：2023/6/24
6
from multiprocessing import Process, Queue
7

8

9
# 需求
10
# （1）主进程与字进程之间借助于队列进行通信
11
# （2）子进程与子进程之间借助于队列进行通信
12

13

14
def producer(que_ue):
15
    que_ue.put('this is a put message')
16
    print('this in a queue')
17

18

19
if __name__ == '__main__':
20
    que_ue = Queue()
21
    pro_cess = Process(target=producer, args=(que_ue,))
22

23
    pro_cess.start()
24
    msg_from_p = que_ue.get()
25

26
    print(msg_from_p)
27
    # this in a queue
28
    # this is a put message

（2）子进程与子进程之间借助队列进行通信#

1
# -*-coding: Utf-8 -*-
2
# @File : 06 IPC机制 .py
3
# author: Chimengmeng
4
# blog_url : https://www.cnblogs.com/dream-ze/
5
# Time：2023/6/24
6
from multiprocessing import Process, Queue
7

8

9
# 需求
10
# （1）主进程与字进程之间借助于队列进行通信
11
# （2）子进程与子进程之间借助于队列进行通信
12

13

14
def producer(que_ue):
15
    que_ue.put('this is a put message')
16
    print('this in a queue')
17

18

19
def customer(que_ue):
20
    msg_from_producer = que_ue.get()
21
    print(msg_from_producer)
22
    # this in a queue
23
    # this is a put message
24

25

26
if __name__ == '__main__':
27
    que_ue = Queue()
28
    p_producer = Process(target=producer, args=(que_ue,))
29
    p_customer = Process(target=customer, args=(que_ue,))
30

31
    p_producer.start()
32
    p_customer.start()

【六】生产者与消费者模型#

生产者模型和消费者模型是指通过利用队列解耦生产者和消费者的一种并发编程模型。
在生产者模型中，生产者负责将数据放入共享队列中，而消费者则从队列中取出数据进行处理。
生产者和消费者之间通过共享这个队列来进行信息的交流。
这种模型适用于生产者和消费者之间的处理速度不一致的情况，同时还能够保证数据传输的安全性和正确性。

在消费者模型中，消费者负责向队列中插入任务，而由线程池中的工作线程进行任务的处理。
消费者和工作线程之间通过共享线程池中的任务队列来完成任务分发和执行。
这种模型适用于任务处理需要一定时间的情况，能够充分利用多线程的优势提高系统的并发性能，提高效率。

生产者：生产/制造东西
消费者：消费/处理东西
该模型还需要一个媒介

比如做包子是先将包子做好后放在蒸笼（媒介）里面，买包子的去蒸笼里面拿
厨师做菜之后用盘子（媒介）装着给消费者端过去
生产者与消费者之间不是直接做交互的，而是借助于媒介
生产者（做包子的） + 媒介（蒸包子） + 消费者（吃包子的）

【1】版本1.0#

1
# -*-coding: Utf-8 -*-
2
# @File : 07 生产者模型与消费者模型 .py
3
# author: Chimengmeng
4
# blog_url : https://www.cnblogs.com/dream-ze/
5
# Time：2023/6/24
6

7
from multiprocessing import Process, Queue
8
import time
9
import random
10

11

12
# 需求
13
# （1）主进程与字进程之间借助于队列进行通信
14
# （2）子进程与子进程之间借助于队列进行通信
15

16

17
def producer(name, food, q):
18
    '''
19

20
    :param name: 生产者名字
21
    :param food: 生产的数据：食物
22
    :param q: 队列
23
    :return:
24
    '''
25
    for i in range(1, 4):
26
        data = f'当前大厨:>>> [{name}] 生产出:>>>[{i}道{food}]菜'
27
        # 模拟延迟
28
        time.sleep(random.randint(1, 3))
29
        q.put(data)
30

31

32
def customer(name, q):
33
    '''
34

35
    :param name: 消费者名字
36
    :param q: 队列
37
    :return:
38
    '''
39
    # 假定消费者消费能力很强
40
    while True:
41
        food = q.get()
42
        # 模拟延迟
43
        time.sleep(random.randint(1, 3))
44
        print(f'当前消费者:>>>[{name}] 消费了:>>>[{food}]')
45

46

47
def main():
48
    pass
49

50

51
if __name__ == '__main__':
52
    # （1）建立模型（媒介，蒸笼）
53
    que_ue = Queue()
54
    # （2） 建立模型（生产者，厨师）
55
    p_producer_dream = Process(target=producer, args=('dream', '鱼香肉丝', que_ue))
56
    p_producer_sprout = Process(target=producer, args=('sprout', '宫保鸡丁', que_ue))
57
    # （3） 建立模型（消费者，顾客）
58
    p_customer_smile = Process(target=customer, args=('smile', que_ue))
59
    p_customer_sad = Process(target=customer, args=('sad', que_ue))
60

61
    p_producer_dream.start()
62
    p_producer_sprout.start()
63
    p_customer_smile.start()
64
    p_customer_sad.start()
65

66
    '''
67
    当前消费者:>>>[smile] 消费了:>>>[当前大厨:>>> [dream] 生产出:>>>[1道鱼香肉丝]菜]
68
    当前消费者:>>>[sad] 消费了:>>>[当前大厨:>>> [dream] 生产出:>>>[2道鱼香肉丝]菜]
69
    当前消费者:>>>[smile] 消费了:>>>[当前大厨:>>> [sprout] 生产出:>>>[1道宫保鸡丁]菜]
70
    当前消费者:>>>[smile] 消费了:>>>[当前大厨:>>> [sprout] 生产出:>>>[2道宫保鸡丁]菜]
71
    当前消费者:>>>[smile] 消费了:>>>[当前大厨:>>> [sprout] 生产出:>>>[3道宫保鸡丁]菜]
72
    当前消费者:>>>[sad] 消费了:>>>[当前大厨:>>> [dream] 生产出:>>>[3道鱼香肉丝]菜]
73
    '''

问题：消费者没有数据被卡主

【2】版本2.0#

向队列的结尾添加结束标志的符号，作为结束的标志

1
# -*-coding: Utf-8 -*-
2
# @File : 07 生产者模型与消费者模型 .py
3
# author: Chimengmeng
4
# blog_url : https://www.cnblogs.com/dream-ze/
5
# Time：2023/6/24
6

7
from multiprocessing import Process, Queue,JoinableQueue
8
import time
9
import random
10

11

12
# 需求
13
# （1）主进程与字进程之间借助于队列进行通信
14
# （2）子进程与子进程之间借助于队列进行通信
15

16

17
def producer(name, food, q):
18
    '''
19

20
    :param name: 生产者名字
21
    :param food: 生产的数据：食物
22
    :param q: 队列
23
    :return:
24
    '''
25
    for i in range(1, 4):
26
        data = f'当前大厨:>>> [{name}] 生产出的:>>>[{i}道{food}]菜'
27
        # 模拟延迟
28
        time.sleep(random.randint(1, 3))
29
        q.put(data)
30

31

32
def customer(name, q):
33
    '''
34

35
    :param name: 消费者名字
36
    :param q: 队列
37
    :return:
38
    '''
39
    # 假定消费者消费能力很强
40
    while True:
41
        food = q.get()
42

43
        # 对收到的信息进行判断：是否有结束的标志
44
        if food is None:
45
            print('消费者消费完毕:>>>!!')
46
            break
47

48
        # 模拟延迟
49
        time.sleep(random.randint(1, 3))
50
        print(f'当前消费者:>>>[{name}] 消费了:>>>[{food}]')
51

52

53
def main():
54
    pass
55

56

57
if __name__ == '__main__':
58
    # （1）建立模型（媒介，蒸笼）
59
    que_ue = Queue()
60
    # （2） 建立模型（生产者，厨师）
61
    p_producer_dream = Process(target=producer, args=('dream', '鱼香肉丝', que_ue))
62
    p_producer_sprout = Process(target=producer, args=('sprout', '宫保鸡丁', que_ue))
63
    # （3） 建立模型（消费者，顾客）
64
    p_customer_smile = Process(target=customer, args=('smile', que_ue))
65
    p_customer_sad = Process(target=customer, args=('sad', que_ue))
66

67
    p_producer_dream.start()
68
    p_producer_sprout.start()
69
    p_customer_smile.start()
70
    p_customer_sad.start()
71

72
    p_producer_dream.join()
73
    p_producer_sprout.join()
74
    # 等待生产者消费完毕之后，向队列中添加特定的结束符号
75
    que_ue.put(None)  # 本条数据一定是生产者将数据全部生产完之后添加进去的
76
    que_ue.put(None)  # 本条数据一定是生产者将数据全部生产完之后添加进去的
77

78
    '''
79
    当前消费者:>>>[smile] 消费了:>>>[当前大厨:>>> [dream] 生产出的:>>>[1道鱼香肉丝]菜]
80
    当前消费者:>>>[sad] 消费了:>>>[当前大厨:>>> [sprout] 生产出的:>>>[1道宫保鸡丁]菜]
81
    当前消费者:>>>[sad] 消费了:>>>[当前大厨:>>> [dream] 生产出的:>>>[2道鱼香肉丝]菜]
82
    当前消费者:>>>[smile] 消费了:>>>[当前大厨:>>> [sprout] 生产出的:>>>[2道宫保鸡丁]菜]
83
    当前消费者:>>>[sad] 消费了:>>>[当前大厨:>>> [sprout] 生产出的:>>>[3道宫保鸡丁]菜]
84
    消费者消费完毕:>>>!!
85
    当前消费者:>>>[smile] 消费了:>>>[当前大厨:>>> [dream] 生产出的:>>>[3道鱼香肉丝]菜]
86
    消费者消费完毕:>>>!!
87
    '''

问题：消费者的个数决定了添加标志的个数

【3】版本3.0 `JoinableQueue`模块#

每当向队列对象中存入数据的时候，队列对象内部会有一个计数器 +1
每当调用一次 task_done() 方法的时候，队列对象内部的计数器 -1
join() 当计数器为 0 时，继续执行代码

1
# -*-coding: Utf-8 -*-
2
# @File : 07 生产者模型与消费者模型 .py
3
# author: Chimengmeng
4
# blog_url : https://www.cnblogs.com/dream-ze/
5
# Time：2023/6/24
6

7
from multiprocessing import Process, Queue, JoinableQueue
8
import time
9
import random
10

11

12
# 需求
13
# （1）主进程与字进程之间借助于队列进行通信
14
# （2）子进程与子进程之间借助于队列进行通信
15

16

17
def producer(name, food, q):
18
    '''
19

20
    :param name: 生产者名字
21
    :param food: 生产的数据：食物
22
    :param q: 队列
23
    :return:
24
    '''
25
    for i in range(1, 4):
26
        data = f'当前大厨:>>> [{name}] 生产出的:>>>[{i}道{food}]菜'
27
        # 模拟延迟
28
        time.sleep(random.randint(1, 3))
29
        q.put(data)
30

31

32
def customer(name, q):
33
    '''
34

35
    :param name: 消费者名字
36
    :param q: 队列
37
    :return:
38
    '''
39
    # 假定消费者消费能力很强
40
    while True:
41
        food = q.get()
42

43
        # 模拟延迟
44
        time.sleep(random.randint(1, 3))
45
        print(f'当前消费者:>>>[{name}] 消费了:>>>[{food}]')
46

47
        # 告诉队列已经从队列中已经取出一个数据了
48
        q.task_done()
49

50

51
def main():
52
    pass
53

54

55
if __name__ == '__main__':
56
    # （1）建立模型（媒介，蒸笼）
57
    que_ue = JoinableQueue()
58
    # （2） 建立模型（生产者，厨师）
59
    p_producer_dream = Process(target=producer, args=('dream', '鱼香肉丝', que_ue))
60
    p_producer_sprout = Process(target=producer, args=('sprout', '宫保鸡丁', que_ue))
61
    # （3） 建立模型（消费者，顾客）
62
    p_customer_smile = Process(target=customer, args=('smile', que_ue))
63
    p_customer_sad = Process(target=customer, args=('sad', que_ue))
64

65
    p_producer_dream.start()
66
    p_producer_sprout.start()
67

68
    # 将消费者设置成守护进程：主进程死亡，子进程跟着死亡
69
    p_customer_smile.daemon = True
70
    p_customer_sad.daemon = True
71

72
    p_customer_smile.start()
73
    p_customer_sad.start()
74

75
    p_producer_dream.join()
76
    p_producer_sprout.join()
77

78
    # 等待消息队列中的所有数据被取完再往下执行代码
79
    que_ue.join()
80
    # join() 当计数器为 0 时，消费者就没有存在的必要了
81

82
    '''
83
    当前消费者:>>>[smile] 消费了:>>>[当前大厨:>>> [dream] 生产出的:>>>[1道鱼香肉丝]菜]
84
    当前消费者:>>>[sad] 消费了:>>>[当前大厨:>>> [sprout] 生产出的:>>>[1道宫保鸡丁]菜]
85
    当前消费者:>>>[sad] 消费了:>>>[当前大厨:>>> [dream] 生产出的:>>>[2道鱼香肉丝]菜]
86
    当前消费者:>>>[smile] 消费了:>>>[当前大厨:>>> [sprout] 生产出的:>>>[2道宫保鸡丁]菜]
87
    当前消费者:>>>[sad] 消费了:>>>[当前大厨:>>> [sprout] 生产出的:>>>[3道宫保鸡丁]菜]
88
    消费者消费完毕:>>>!!
89
    当前消费者:>>>[smile] 消费了:>>>[当前大厨:>>> [dream] 生产出的:>>>[3道鱼香肉丝]菜]
90
    消费者消费完毕:>>>!!
91
    '''

音乐

进程进阶知识点小结

【2.0】知识点小结（进程进阶）#

【一】进程对象及其他方法#

【1】查看当前进程的进程号#

（1）查看当前进程的进程号current_process().pid 方法#

（2）查看当前进程的进程号os.getpid() 方法#

（3）查看当前进程的父进程的进程号os.getpid() 方法#

【2】杀死当前进程p.terminate()#

【3】判断当前进程是否存活p.is_alive()#

【二】僵尸进程与孤儿进程#

【1】僵尸进程#

【2】孤儿进程#

【三】守护进程#

【1】引入 - 主进程死亡，子进程未死亡#

【2】使用 - 主进程死亡，子进程必死亡#

（1）错误使用演示#

（2）正确使用演示#

【四】互斥锁#

【1】数据错乱问题#

【2】加锁处理#

【3】行锁与表锁#

（1）行锁#

（2）表锁#

（3）小结#

【4】注意#

【五】进程间通信#

【1】队列模块：Queue#

（1）简单使用#

（2）没有数据抛出异常.get_nowait()#

（3）没有数据延迟抛出异常.get(timeout=3)#

（4）判断当前队列数据是否填满队列.full()#

（5）判断当前队列是否为空.empty()#

（6）小结#

【2】IPC机制#

（1）子进程与主进程之间通过队列进行通信#

（2）子进程与子进程之间借助队列进行通信#

【六】生产者与消费者模型#

【1】版本1.0#

【2】版本2.0#

【3】版本3.0 JoinableQueue模块#

支持与分享

（1）查看当前进程的进程号`current_process().pid` 方法#

（2）查看当前进程的进程号`os.getpid()` 方法#

（3）查看当前进程的父进程的进程号`os.getpid()` 方法#

【2】杀死当前进程`p.terminate()`#

【3】判断当前进程是否存活`p.is_alive()`#

（2）没有数据抛出异常`.get_nowait()`#

（3）没有数据延迟抛出异常`.get(timeout=3)`#

（4）判断当前队列数据是否填满队列`.full()`#

（5）判断当前队列是否为空`.empty()`#

【3】版本3.0 `JoinableQueue`模块#