python---进程

文章目录

• 1、进程介绍
• 2、进程的特征
• 2、进程的组成
• • 3.1、 进程控制块（PCB）
  • 3.2 进程的内存布局
• 4、 创建进程
• • 4.1、创建线程代码
  • 4.2、线程池
  • • 三种执行方式详解
    • • 方式1：同步执行（apply）
      • 方式2：异步执行（apply_async）
      • 方式3：批量映射（map）
    • 三种方式性能对比
    • 使用 with 语句管理资源
    • 代码1：
• 5、进程间通信（IPC）
• • 5.1、 使用队列（Queue）
  • 5.2、 使用管道（Pipe）
• 6、进程的生命周期
• • 6.1、进程状态转换
• 7、进程间的关系
• • 7.1、 父子进程关系
  • 7.2 进程组和会话

1、进程介绍

Python中的多线程无法利用多核优势（参考文章python—线程），如果想要充分地使用多核CPU的资源，需要使用多进程。 进程是操作系统进行资源分配和调度的基本单位。每个进程都有自己独立的内存空间、代码段、数据段和堆栈段。进程的启动和销毁的代价是比较大。进程间的任务切换也是开销最大的。 一个进程至少要包含一个线程，每个进程在启动的时候就会自动的启动一个线程，进程里面的第一个线程就是主线程，每次在进程内创建的子线程都是由主线程进程创建和销毁，子线程也可以由主线程（即其他子线程）创建出来的线程创建和销毁线程。

2、进程的特征

1、独立性：进程之间相互隔离，一个进程崩溃不会影响其他进程 2、拥有资源：每个进程都有独立的地址空间和系统资源 3、并发性：多个进程可以同时执行（在多核CPU上真正并行） 4、动态性：进程有创建、执行、暂停、终止等生命周期

2、进程的组成

3.1、 进程控制块（PCB）

# 模拟进程控制块的概念
class ProcessControlBlock:
def __init__(self, pid, name, state, priority):
self.pid = pid # 进程ID
self.name = name # 进程名称
self.state = state # 状态：就绪、运行、阻塞
self.priority = priority # 优先级
self.program_counter = 0 # 程序计数器
self.memory_info = {} # 内存信息
self.io_status = {} # I/O状态信息

3.2 进程的内存布局

进程地址空间：
┌─────────────────┐ 高地址
│ 栈(stack) │ ← 局部变量、函数调用
├─────────────────┤
│ ↓ │ 向下增长
│ 空 │
├─────────────────┤
│ ↑ │ 向上增长
│ 堆(heap) │ ← 动态分配的内存
├─────────────────┤
│ 数据段(data) │ ← 全局变量、静态变量
├─────────────────┤
│ 代码段(text) │ ← 程序代码
└─────────────────┘ 低地址

4、 创建进程

创建进程使用Process类。 Process(group = None,target =None,name=None, args=[ ], kwargs={ }) 参数说明：

1 group参数未使用，值始终为None（可忽略）
2 target表示调用对象，即子进程要执行的回调函数
3 name为子进程的名称
4 args表示调用对象的位置参数元组，args=(1,2,'anne',)
5 kwargs表示调用对象的字典,kwargs={'name':'anne','age':18}

process类的属性&方法：

authkey:进程的身份验证密钥
daemon:同thread的setDaemon，守护进程
exitcode:进程运行时为None，若为—N，则表示被信号N结束
pid:进程号
name:进程名
is_alive():返回进程是否正在运行
join([timeout]):阻塞到线程结束或到timeout值
start():进程准备就绪，等待CPU调度
run():start()调用run方法，如果实例进程时未制定传入target，start执行默认run()方法。
terminate():不管任务是否完成，立即停止工作进程

4.1、创建线程代码

注意python文件名称不要和模块名称一样，否则会报错。 代码一：

import multiprocessing
import os
import time

def worker(name, delay):
print(f"进程 {name} (PID: {os.getpid()}) 开始执行")
time.sleep(delay)
print(f"进程 {name} 结束")
return f"进程 {name} 的结果"

if __name__ == "__main__":
# 创建进程
p1 = multiprocessing.Process(target=worker, args=("A", 2))
p2 = multiprocessing.Process(target=worker, args=("B", 1))

# 启动进程
p1.start()
p2.start()

# 等待进程结束
p1.join()
p2.join()

print("所有进程执行完毕")

代码2：

from multiprocessing import Process # 多进程的类
import time
import random

def test_function(name):
time.sleep(random.randrange(1, 5))
print(f"我是{name}子进程！")

# 进程一定要写在“if __name__ == '__main__':”下面
if __name__ == '__main__':
process_list = []
for i in range(3):
# 进程中的参数args表示调用对象的位置参数元组.注意：元组中只有一个元素时结尾要加","逗号
p = Process(target=test_function, args=(f"sonThread{i+1}",))
p.start()
process_list.append(p)
for i in process_list:
i.join()
print("主进程结束！")

4.2、线程池

multiprocessing.Pool：进程池模块，用于创建和管理多个工作进程

三种执行方式详解

方式1：同步执行（apply）

result = pool.apply(task, (10,))

特点：阻塞执行，等待任务完成才继续 工作流程： 1、分配一个进程执行 task(10) 2、主进程等待1秒 3、返回结果 100 适用场景：需要按顺序执行的任务

方式2：异步执行（apply_async）

results = []
for i in range(10):
result = pool.apply_async(task, (i,))
results.append(result)

output = [res.get() for res in results]

特点：非阻塞，立即返回 AsyncResult 对象

工作流程：

1、快速提交10个任务到进程池

2、4个进程并行处理（进程池大小为4）

3、使用 .get() 获取结果时会阻塞等待

执行时间：约3秒（不是10秒！）

1、第1批：4个任务并行（1秒）

2、第2批：4个任务并行（1秒）

3、第3批：2个任务并行（1秒）

方式3：批量映射（map）

pool.map(task, range(5))

特点：简化版的批量处理

工作流程：

1、将 range(5) 分成多个块

2、分配给进程池中的进程

3、等待所有任务完成

等价于：[task(0), task(1), task(2), task(3), task(4)]

执行时间：约2秒（5个任务，4个进程）

三种方式性能对比

使用 with 语句管理资源

with Pool(processes=4) as pool:

自动管理进程池的生命周期

退出时自动关闭进程池

代码1：

from multiprocessing import Pool
import time

def task(x):
time.sleep(1)
return x * x

if __name__ == "__main__":
# 创建包含4个进程的进程池
with Pool(processes=4) as pool:
# 同步执行
result = pool.apply(task, (10,))
print(f"同步结果: {result}")

# 异步执行（推荐）
results = []
for i in range(10):
result = pool.apply_async(task, (i,))
results.append(result)

# 获取所有结果
output = [res.get() for res in results]
print(f"异步结果: {output}")

# 使用 map
print(f"map结果: {pool.map(task, range(5))}")

运行结果：

同步结果: 100
异步结果: [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]
map结果: [0, 1, 4, 9, 16]

5、进程间通信（IPC）

在Python中，实现进程间通信（IPC）的主要方式包括： 管道（Pipe）、队列（Queue）、共享内存、信号量/锁、命名管道（FIFO）、共享文件、Socket通信、消息队列（第三方库）、ZeroMQ、gRPC（跨语言）。 以下介绍队列、管道。

5.1、 使用队列（Queue）

import multiprocessing
import time

def producer(queue, items):
for item in items:
print(f"生产: {item}")
queue.put(item)
time.sleep(0.5)
queue.put(None) # 结束信号

def consumer(queue):
while True:
item = queue.get()
if item is None:
break
print(f"消费: {item}")
time.sleep(1)

if __name__ == "__main__":
queue = multiprocessing.Queue(maxsize=3)

p1 = multiprocessing.Process(target=producer, args=(queue, [1, 2, 3, 4, 5]))
p2 = multiprocessing.Process(target=consumer, args=(queue,))

p1.start()
p2.start()

p1.join()
p2.join()

5.2、 使用管道（Pipe）

from multiprocessing import Pipe, Process

def worker(conn):
conn.send(['hello', 'world'])
data = conn.recv()
print(f"Worker received: {data}")
conn.close()

if __name__ == '__main__':
parent_conn, child_conn = Pipe()
p = Process(target=worker, args=(child_conn,))
p.start()

print(f"Parent received: {parent_conn.recv()}")
parent_conn.send(['parent', 'data'])
p.join()

运行结果：

Parent received: ['hello', 'world']
Worker received: ['parent', 'data']

6、进程的生命周期

6.1、进程状态转换

import multiprocessing
import time
import signal

def process_lifecycle():
"""演示进程的生命周期"""
print("1. 进程创建 – 进入新建状态")
time.sleep(0.5)

print("2. 进程启动 – 进入就绪状态")
# 模拟就绪到运行
print("3. 获得CPU – 进入运行状态")

for i in range(3):
print(f" 运行中… 第{i+1}秒")
time.sleep(1)

# 模拟I/O阻塞
if i == 1:
print("4. 等待I/O – 进入阻塞状态")
time.sleep(2)
print("5. I/O完成 – 返回就绪状态")

print("6. 任务完成 – 进入终止状态")

def signal_handler(signum, frame):
"""信号处理函数"""
print(f"\\n收到信号 {signum}，进程将终止")

if __name__ == "__main__":
# 设置信号处理
signal.signal(signal.SIGTERM, signal_handler)
signal.signal(signal.SIGINT, signal_handler)

p = multiprocessing.Process(target=process_lifecycle)

print("=== 进程生命周期演示 ===")
print(f"进程状态: 未创建")

p.start()
print(f"进程状态: 启动中 (PID: {p.pid})")

# 监控进程状态
while p.is_alive():
print(f"进程状态: 运行中")
time.sleep(1)

print(f"进程状态: 已终止 (退出码: {p.exitcode})")

7、进程间的关系

7.1、 父子进程关系

import os
import time
import signal
from multiprocessing import Process, current_process

def child_process():
"""子进程"""
print(f"子进程 PID: {os.getpid()}, 父进程 PID: {os.getppid()}")

# 子进程成为孤儿进程的情况
time.sleep(5)
print(f"5秒后，父进程可能已结束，新父进程 PID: {os.getppid()}")

# 子进程结束
print("子进程正常结束")

def parent_process():
"""父进程"""
print(f"父进程 PID: {os.getpid()}")

# 创建子进程
child = Process(target=child_process)
child.start()

# 父进程提前结束
print("父进程立即结束，子进程将成为孤儿进程")
# 注意：这里不调用 child.join()

if __name__ == "__main__":
print("=== 孤儿进程演示 ===")
parent = Process(target=parent_process)
parent.start()
parent.join()

# 主进程等待一段时间，观察孤儿进程
time.sleep(10)
print("主进程结束")

7.2 进程组和会话

import os
import time
from multiprocessing import Process

def show_process_info(name):
"""显示进程信息"""
pid = os.getpid()
pgid = os.getpgid(0) # 获取进程组ID
sid = os.getsid(0) # 获取会话ID

print(f"{name}: PID={pid}, PGID={pgid}, SID={sid}")
time.sleep(2)

if __name__ == "__main__":
print("主进程信息:")
show_process_info("主进程")

# 创建子进程（默认在同一进程组）
processes = []
for i in range(3):
p = Process(target=show_process_info, args=(f"子进程{i+1}",))
processes.append(p)
p.start()

for p in processes:
p.join()