操作系统进程相关笔记
提示： 进程牵扯到大量操作系统原理知识，建议先学习《操作系统原理》，再来学习多进程相关内容。
1. 使用 os.fork() 创建多进程
说明：
此方法适用于 Linux 环境。调用 os.fork() 后，若返回值为 0，则证明当前为子进程；否则为父进程。
import os, time
result = os.fork()
print("process pid is:", result) # 如果输出 0 则证明是子进程
if result == 0:
print("子进程输出 process")
time.sleep(20)
print("子进程 pid is :", os.getpid())
print("子进程输出父进程 pid is ：", os.getppid())
else:
print("父进程输出 process")
time.sleep(20)
print("parent pid is:", os.getpid())
运行后的示例（使用 ps -ef |grep python3 查看进程）：
[root@redis-slave-k3s python-test]# ps -ef |grep python3
root 42017 1624 0 11:21 pts/0 00:00:00 python3 os_process.py
root 42018 42017 0 11:21 pts/0 00:00:00 python3 os_process.py
2. 僵尸进程
描述：
当子进程的 PCB（进程控制块）未被父进程回收时，即子进程结束后父进程未调用 wait 或 waitpid 获取状态，导致子进程的进程描述符仍保留在内存中，这种进程称为僵尸进程。
import os, time
result = os.fork()
print("process pid is:", result) # 如果输出 0 则证明是子进程
if result == 0:
print("子进程输出 process")
## time.sleep(20) 注释此处，让子进程比父进程提前退出
print("子进程 pid is :", os.getpid())
print("子进程输出父进程 pid is ：", os.getppid())
else:
print("父进程输出 process")
time.sleep(20)
print("parent pid is:", os.getpid())
运行后的示例（使用 ps -ef |grep python3 查看进程）：
[root@redis-slave-k3s python-test]# ps -ef |grep python3
root 41654 1624 0 11:20 pts/0 00:00:00 python3 os_process.py
root 41655 41654 0 11:20 pts/0 00:00:00 [python3] <defunct>
3. 孤儿进程
描述：
当父进程退出时，如果子进程仍在运行，此子进程会被 pid 为 1 的进程（通常为 init 或其替代进程）收养，称为孤儿进程。
import os, time
result = os.fork()
print("process pid is:", result) # 如果输出 0 则证明是子进程
if result == 0:
print("子进程输出 process")
time.sleep(20)
print("子进程 pid is :", os.getpid())
print("子进程输出父进程 pid is ：", os.getppid())
else:
print("父进程输出 process")
# time.sleep(20)
print("parent pid is:", os.getpid())
运行后的示例（使用 ps -ef |grep python3 查看进程）：
[root@redis-slave-k3s python-test]# ps -ef |grep python3
root 42993 1 0 11:23 pts/0 00:00:00 python3 os_process.py
4. 多进程（multiprocessing 模块）
4.1 基础示例
说明：
以下代码使用 multiprocessing 模块创建了多个进程，并使用 current_process 显示当前进程的信息。代码中将值追加到列表 lst，但注意：每个进程拥有独立内存，子进程中对 lst 的修改不会影响其他进程中的 lst。
#!python
# -*- coding: UTF-8 -*-
"""
@Project ：sc_learn
@File ：05多进程.py
@IDE ：PyCharm
@Author ：Cailuobozi
@Date ：2025/4/13 上午11:29
"""
from multiprocessing import Process, current_process
import time
lst = []
def task(i):
print(current_process().name, i, "start....")
time.sleep(1)
lst.append(i)
print(lst, id(lst))
print(current_process().name, i, "end...")
if __name__ == "__main__": # 程序入口文件，多进程必须放在入口中运行，否则会报错
for i in range(4):
p = Process(target=task, args=(i,))
p.start()
4.2 自定义进程类
说明：
下面示例展示了如何通过继承 Process 类来自定义进程。自定义类 MyProcess 中重写了 run 方法来定义进程任务。
#!python
# -*- coding: UTF-8 -*-
"""
@Project ：sc_learn
@File ：05多进程.py
@IDE ：PyCharm
@Author ：Cailuobozi
@Date ：2025/4/13 上午11:29
"""
from multiprocessing import Process, current_process
import time
class MyProcess(Process):
def __init__(self, name, delay):
super().__init__()
self.name = name
self.delay = delay
def run(self):
"""进程执行的任务"""
print(f"进程 {self.name} 开始运行")
count = 0
while count < 3:
time.sleep(self.delay)
print(f"{self.name}: 执行时间 {time.ctime(time.time())}")
count += 1
print(f"进程 {self.name} 结束")
if __name__ == "__main__": # 程序入口文件，多进程必须放在入口中运行，否则会报错
process1 = MyProcess(name="进程A", delay=1)
process2 = MyProcess(name="进程B", delay=2)
process1.start()
process2.start()
process1.join()
process2.join()
print("父进程结束")
进程间通信
消息队列方式
#!python
# -*- coding: UTF-8 -*-
#Queue
from multiprocessing import Process,Queue
def producer(q):
q.put("Data from producer")
def consumer(q):
print("Consumer received:",q.get())
if __name__ == '__main__':
q=Queue()
p1 = Process(target=producer, args=(q,))
p2 = Process(target=consumer, args=(q,))
p1.start()
p2.start()
p1.join()
p2.join()
print("end.....")
共享内存
#!python
# -*- coding: UTF-8 -*-
from multiprocessing import Process, Value, Lock
def increment(counter, lock):
with lock:
counter.value += 1
if __name__ == '__main__':
counter = Value("i", 0)
lock = Lock()
processes = [Process(target=increment, args=(counter, lock)) for _ in range(10)]
for p in processes:
p.start()
for p in processes:
p.join()
print("final counter:", counter.value)
进程池
#!python
# -*- coding: UTF-8 -*-
# 进程池 预创建
from multiprocessing import Pool, current_process
import time
lst = []
def task(i):
print(current_process().name, i, "start...")
time.sleep(1)
lst.append(i)
print(lst)
print(current_process().name, i, "end...")
if __name__ == '__main__':
# 创建进程池，建议设置的进程数和cpu核数一致
p = Pool(processes=4, maxtasksperchild=2)
for i in range(20):
# 进程池接收任务 异步执行单个任务
p.apply_async(func=task, args=(i,))
# 关闭进程池 不接受任务
p.close()
# 阻塞当前环境
p.join()
print("end....")