Python进阶系列之-多任务：进程、线程与协程
Python进阶系列之-多任务：进程、线程与协程
写在前面：在写代码时，你有没有遇到过“一边下载电影，一边还要听歌，结果程序卡死”的情况？这就是单任务时代的痛点！为了榨干CPU的性能，让程序飞起来，我们必须掌握多任务编程。多任务能充分利用CPU资源，大幅提升程序执行效率，是Python进阶的必学知识点，也是面试的高频考点。
本文从「并发与并行」的核心区别出发，系统讲解多进程、多线程的实现方式、底层特性与常见坑点，深入解析互斥锁如何解决线程安全问题，最后拆解with open背后的上下文管理器原理，带你彻底理清进程、线程的恩怨情仇。
一、多任务核心基础
1.1 什么是多任务
同一时间内执行多个任务，就叫多任务。比如电脑同时运行微信、浏览器、音乐播放器，是操作系统级别的多任务；程序里一边下载文件一边处理数据，是代码级别的多任务。
多任务的核心价值：充分利用CPU多核资源，减少程序等待时间，提升执行效率。
1.2 面试必问：并发 vs 并行
这是多任务的基础概念题，二者有本质区别：
- 并发：多个任务同时请求同一个CPU资源，但同一瞬间CPU只能执行一个任务，于是安排它们交替执行。看起来好像是同时执行的，其实只是CPU切换得太快了。单核CPU只能实现并发。
- 并行：多个任务真正同时执行。前提是：你的电脑得是多核CPU！
1.3 进程与线程的关系
- 进程：是CPU分配资源的最小单位。简单理解，一个.exe可执行程序就是一个进程。
- 线程：是进程的执行路径，执行单元，是CPU调度的最小单位。
- 多进程：多个.exe同时执行，例如QQ、微信、飞秋一起运行。
- 多线程：进程内部的多任务执行，例如在微信里，同时和张三、李四聊天。
🚗 大白话比喻：
进程 = 车，线程 = 车道。
如果是单车道，就叫单线程（容易堵车）；如果是多车道，就叫多线程（畅通无阻）。
1.4 补充：GIL（全局解释器锁）
在深入多线程之前，必须了解Python的一个特殊机制——GIL（Global Interpreter Lock）。
GIL是CPython解释器中的一个互斥锁，它确保同一时刻只有一个线程能执行Python字节码。这意味着，即使在多核CPU上，Python的多线程也无法真正并行执行CPU密集型任务。但在IO密集型任务中，线程会在IO等待时主动释放GIL，因此多线程依然能带来显著的并发收益。
GIL 并不是一直持有不放，它会在两种情况下主动释放：
- 遇到 IO 阻塞时：比如网络请求、文件读写、time.sleep 时，线程会主动让出 GIL，其他线程可以趁机执行。这也是 IO 密集型任务用多线程依然能提速的核心原因。
- 按时间片轮转：Python3.2 之后，解释器会按固定时间片检查，到期后强制当前线程释放 GIL，让其他线程有机会执行。
正因为如此：
CPU 密集型任务：线程大部分时间在计算，GIL 切换带来额外开销，多线程反而可能比单线程还慢
IO 密集型任务：线程大部分时间在等待，GIL 会频繁让出，多线程能显著提升效率
结论：
- CPU密集型任务 → 多进程（绕过GIL）
- IO密集型任务 → 多线程或协程（GIL不是瓶颈）
二、多进程编程
Python的multiprocessing模块提供了完整的多进程支持，可以轻松实现多任务并行执行。
多进程的思路很简单：导包 -> 创建进程对象关联函数 -> 开启进程。
2.1 多进程入门案例
实现“一边敲代码，一边听音乐”的多任务效果：
import multiprocessing
import time
# 任务1：敲代码
def coding():
for i in range(10):
print(f'敲代码...{i}')
time.sleep(0.1)
# 任务2：听音乐
def music():
for i in range(10):
print(f'听音乐--------{i}')
time.sleep(0.1)
if __name__ == '__main__':
# 创建进程对象，关联目标函数
p1 = multiprocessing.Process(target=coding)
p2 = multiprocessing.Process(target=music)
# 开启进程
p1.start()
p2.start()
注意：Windows系统下，多进程代码必须写在
if __name__ == '__main__':内部，否则会出现进程递归创建的报错。
为什么 Windows 必须加这行？
答：Windows 系统创建子进程默认采用 spawn 模式：子进程会重新导入并执行当前整个 Python 脚本。如果不做 if name == ‘main’ 判断，子进程会再次执行创建进程的代码，导致无限递归创建进程，最终报错。Linux/macOS 默认采用 fork 模式，直接复制父进程内存空间，没有这个问题，但加上这行属于跨平台兼容的最佳实践。
2.2 带参数的多进程（重点）
给任务函数传参有两种标准方式：
args：元组形式传参，参数顺序必须和函数形参完全一致kwargs：字典形式传参，键名必须和函数形参名完全一致
import multiprocessing
import time
def coding(name, num):
for i in range(num):
print(f'{name} 正在敲第 {i} 遍代码...')
time.sleep(0.1)
def music(name, count):
for i in range(count):
print(f'{name} 正在听第 {i} 首音乐...')
time.sleep(0.1)
if __name__ == '__main__':
# args元组传参
p1 = multiprocessing.Process(target=coding, args=('乔峰', 10), name='进程1_敲代码')
print(f'进程p1的名字为: {p1.name}')
# kwargs字典传参
p2 = multiprocessing.Process(target=music, kwargs={'count': 5, 'name':'虚竹'}, name='进程2_听音乐')
print(f'进程p2的名字为: {p2.name}')
p1.start()
p2.start()
2.3 获取进程ID
每个进程都有唯一的进程ID（PID），方便我们管理和监控进程：
os.getpid()：获取当前进程的PIDos.getppid()：获取当前进程的父进程PIDmultiprocessing.current_process().pid：也可获取当前进程PID
重点结论： 进程之间的数据是相互隔离的！ 子进程相当于把父进程的资源拷贝了一份，互不干扰。
import multiprocessing
import time
import os
def coding(name, num):
for i in range(num):
print(f'{name} 正在敲第 {i} 遍代码...')
time.sleep(0.1)
print(f'当前进程PID: {os.getpid()}, 父进程PID: {os.getppid()}')
if __name__ == '__main__':
p1 = multiprocessing.Process(target=coding, args=('乔峰', 10))
p1.start()
print(f'main主进程PID: {os.getpid()}')
2.4 进程间数据相互隔离
不同进程拥有独立的内存空间，数据互不共享。一个进程修改全局变量，不会影响其他进程。
import multiprocessing, time, os
my_list = [] # 全局变量
def write_data():
for i in range(1, 6):
my_list.append(i)
print(f'add: {i}')
print(f'write_data函数: {my_list}')
def read_data():
time.sleep(3) # 确保写完毕
print(f'read_data函数: {my_list}') # 发现读不到数据，因为数据隔离！
if __name__ == '__main__':
p1 = multiprocessing.Process(target=write_data)
p2 = multiprocessing.Process(target=read_data)
p1.start()
p2.start()
# 获取main进程的id
print(f'main进程的id为: {os.getpid()}')
运行后会发现，read_data函数读到的列表依然是空的——因为两个进程各有一份独立的变量副本，修改互不影响。
2.5 进程间通信（IPC）
虽然进程间数据隔离，但有时我们需要在进程间传递数据。Python提供了多种IPC方式，最常用的是Queue（队列）：
from multiprocessing import Process, Queue
def producer(q):
q.put('hello')
q.put('world')
def consumer(q):
print(q.get())
print(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()
其他IPC方式还包括Pipe（管道）、Manager（共享内存管理器）等。
multiprocessing.Queue 是进程安全的队列，内部自带互斥锁，多个进程同时写入、读取都不会出现数据错乱，非常适合传递简单的消息、小体积数据。
如果需要传递大量数据或者更复杂的共享对象，可以使用 multiprocessing.Manager 或者 shared_memory 共享内存。
2.6 主进程与子进程的生命周期
默认情况下，主进程会等待所有子进程执行完毕后，才会结束整个程序。
import multiprocessing
import time
def work():
for i in range(10):
print(f'工作中... {i}')
time.sleep(0.3)
if __name__ == '__main__':
p1 = multiprocessing.Process(target=work)
p1.start()
time.sleep(1)
print('主进程代码执行完毕！')
运行后可以看到，主进程打印完结束语句后，程序并没有立即退出，会等待子进程循环结束后才终止。
join ()：等待子进程结束
如果需要主进程等待子进程全部执行完毕后，再执行后续收尾逻辑，可以使用 join() 方法：
import multiprocessing
import time
def work():
time.sleep(2)
print("子任务执行完成")
if __name__ == '__main__':
p = multiprocessing.Process(target=work)
p.start()
p.join() # 主进程阻塞在这里，等待p执行完毕再往下走
print("主进程收尾逻辑")
如果有多个子进程，可以逐个调用 join()，实现 “所有子任务完成后再统一处理结果” 的同步逻辑。
2.7 守护进程与终止进程
如果主进程干完了，想让子进程跟着死，怎么办？两种方式：
- 设置守护进程（推荐）：p1.daemon = True（像守护骑士一样，主公死，骑士亡）。
- 手动关闭：p1.terminate()（不推荐，子进程会变成僵尸进程，不立即释放资源）。
import multiprocessing
import time
def work():
for i in range(10):
print(f'工作中... {i}')
time.sleep(0.3)
if __name__ == '__main__':
p1 = multiprocessing.Process(target=work)
# 方式1：设置为守护进程（推荐）
# p1.daemon = True
p1.start()
time.sleep(1)
# 方式2：手动终止进程（不推荐，可能产生僵尸进程）
p1.terminate()
print('主进程执行结束了!')
⚠️ 注意
terminate()不会立即释放资源，可能产生僵尸进程；守护进程会随主进程正常销毁，更推荐使用。
好的，下面是为您博客补充的 2.8 进程池（Pool） 详细内容。可直接复制到文章中，放在 2.7 节之后。
2.8 进程池（Pool）：高效并行计算
为什么需要进程池？
在前面的例子中，我们手动创建一个个进程对象（Process），然后逐个 start() 和 join()。这种方式在小规模任务中没问题，但当任务数量很大（比如上千个）时，手动管理进程会变得繁琐且低效：
- 创建和销毁进程的开销较大
- 系统能同时运行的进程数受 CPU 核数和内存限制
- 手动管理难以控制并发数量，容易耗尽系统资源
进程池（Pool） 正是为了解决这些问题而生。它预先创建一组进程（池），将任务分配给池中的空闲进程执行，任务完成后进程不销毁，而是继续执行下一个任务，从而大幅降低创建/销毁进程的开销，并能有效控制并发数量。
2.8.1 进程池的基本使用
multiprocessing.Pool 提供了便捷的进程池接口，常用方法如下：
| 方法 | 说明 | 是否阻塞 |
|---|---|---|
apply(func, args) |
同步执行，一个任务执行完后才能提交下一个 | ✅ 阻塞 |
apply_async(func, args, callback) |
异步执行，非阻塞，可同时提交多个任务 | ❌ 非阻塞 |
map(func, iterable) |
同步版 map，将可迭代对象中的每个元素交给进程池处理 | ✅ 阻塞 |
map_async(func, iterable, callback) |
异步版 map | ❌ 非阻塞 |
close() |
关闭进程池，不再接受新任务 | - |
join() |
等待所有任务执行完毕（必须先 close） | ✅ 阻塞 |
基础示例：计算平方
import multiprocessing
import time
def square(x):
time.sleep(0.5)
return x * x
if __name__ == '__main__':
# 创建进程池，默认使用 CPU 核心数
pool = multiprocessing.Pool(processes=4)
# 同步方式：apply
result = pool.apply(square, (10,))
print(f'同步结果: {result}')
# 异步方式：apply_async，返回 AsyncResult 对象
async_result = pool.apply_async(square, (5,))
print(f'异步结果: {async_result.get()}') # get() 会阻塞等待结果
# 批量任务：map
numbers = [1, 2, 3, 4, 5]
results = pool.map(square, numbers)
print(f'map 结果: {results}')
pool.close()
pool.join()
输出示例（每次运行可能有微小差异，但结果一致）：
注意：
pool.close()和pool.join()必须成对出现，且join()必须在close()之后调用。
2.8.2 异步提交与回调函数
apply_async 和 map_async 是非阻塞的，可以同时提交多个任务，并通过 callback 参数指定任务完成后的回调函数。
import multiprocessing
import time
def worker(num):
time.sleep(1)
return num * 2
def callback(result):
print(f'任务完成，结果: {result}')
if __name__ == '__main__':
pool = multiprocessing.Pool(processes=3)
# 异步提交多个任务，每个任务完成后自动调用 callback
for i in range(5):
pool.apply_async(worker, args=(i,), callback=callback)
pool.close()
pool.join()
print('所有任务执行完毕')
输出示例（顺序可能不同，因为任务完成时间有先后）：
回调函数在主进程中执行，适合用于更新 UI、记录日志等操作。
2.8.3 进程池的上下文管理器
Python 3.3+ 支持使用 with 语句管理进程池，自动调用 close() 和 join()，代码更简洁：
import multiprocessing
def cube(x):
return x ** 3
if __name__ == '__main__':
with multiprocessing.Pool(processes=4) as pool:
results = pool.map(cube, range(10))
print(results) # [0, 1, 8, 27, 64, 125, 216, 343, 512, 729]
2.8.4 进程池 vs 手动创建进程
| 对比项 | 手动创建 Process | 进程池 Pool |
|---|---|---|
| 创建开销 | 每次任务都创建/销毁进程，开销大 | 预创建进程，复用，开销小 |
| 并发控制 | 需手动管理数量，容易失控 | 通过 processes 参数精确控制 |
| 任务提交 | 需手动 start/join | 提供 apply/map 等便捷方法 |
| 结果获取 | 需通过队列或共享内存 | 直接返回结果，支持回调 |
| 适用场景 | 少量、长期运行的后台进程 | 大量、短小的并行任务 |
2.8.5 实战：并行计算 π 值
蒙特卡洛方法估算 π 是一个经典的并行计算案例，非常适合用进程池加速。
import multiprocessing
import random
import math
def estimate_pi_part(n):
"""计算 n 个随机点落在圆内的数量"""
count = 0
for _ in range(n):
x = random.random()
y = random.random()
if x*x + y*y <= 1:
count += 1
return count
def estimate_pi(total_points, processes=4):
"""使用进程池并行估算 π"""
points_per_process = total_points // processes
with multiprocessing.Pool(processes=processes) as pool:
counts = pool.map(estimate_pi_part, [points_per_process] * processes)
total_in_circle = sum(counts)
pi_estimate = 4 * total_in_circle / total_points
return pi_estimate
if __name__ == '__main__':
total = 10_000_000 # 1千万个点
pi = estimate_pi(total, processes=8)
print(f'估算 π ≈ {pi:.6f}')
print(f'误差: {abs(math.pi - pi):.6f}')
运行后输出类似：
性能提升：使用 8 进程比单进程快约 6~7 倍（受 CPU 核心数和任务调度影响）。
2.8.6 注意事项
- 进程池中的异常处理：如果 worker 函数抛出异常，
get()或map()会重新抛出该异常，需要在主进程中捕获。 - 不要在进程池中使用全局变量：每个子进程都有独立的内存空间，修改全局变量不会影响其他进程。
- 进程池不适合 IO 密集型任务：IO 密集型更适合线程池或协程，因为进程切换开销大于线程。
- Windows 上的限制：Windows 下进程池必须在
if __name__ == '__main__':中创建，否则会递归报错。
三、多线程编程
线程的操作和进程非常类似，把 multiprocessing 换成 threading 即可。线程是进程内部的执行单元，threading模块提供了多线程支持。多线程开销比多进程小，尤其适合IO密集型任务。
3.1 多线程入门案例
同样实现“边敲代码边听音乐”，多线程代码和多进程结构非常相似：
import threading
import time
def coding():
for i in range(10):
print(f'正在敲第 {i} 遍代码! -------')
time.sleep(0.1)
def music():
for i in range(10):
print(f'正在听第 {i} 首音乐!')
time.sleep(0.1)
if __name__ == '__main__':
t1 = threading.Thread(target=coding)
t2 = threading.Thread(target=music)
t1.start()
t2.start()
3.2 多线程内存执行模型
多线程共享同一个进程的资源，多个线程抢占CPU执行，内存结构如下图所示：
每个线程有自己独立的栈空间，但共享进程的堆内存和全局资源，因此线程间通信比进程简单得多，但也带来了数据安全问题。
3.3 带参数的多线程
和多进程一样，线程也支持args元组传参和kwargs字典传参：
import threading
import time
def coding(name, num):
for i in range(num):
print(f'{name} 正在敲第 {i} 遍代码! -------')
time.sleep(0.1)
def music(name, count):
for i in range(count):
print(f'{name} 正在听第 {i} 首音乐!')
time.sleep(0.1)
if __name__ == '__main__':
t1 = threading.Thread(target=coding, args=('乔峰', 9))
t2 = threading.Thread(target=music, kwargs={'name': '虚竹', 'count': 10})
t1.start()
t2.start()
3.4 线程执行顺序：抢占式调度
多线程的执行顺序是随机的，因为Python采用抢占式调度：哪个线程抢到CPU资源，哪个线程就执行。
import threading
import time
def get_info():
time.sleep(0.1)
ct = threading.current_thread()
print(f'当前的线程对象是: {ct}')
if __name__ == '__main__':
for i in range(10):
t = threading.Thread(target=get_info)
t.start()
多次运行会发现，线程的打印顺序每次都不一样。
3.5 主线程与守护线程
和进程的生命周期规则一致：
- 默认主线程会等待所有子线程执行完毕再结束
- 设置守护线程后，主线程结束则守护线程立即终止
import threading
import time
def work():
for i in range(10):
print(f'工作中... {i}')
time.sleep(0.3)
if __name__ == '__main__':
t1 = threading.Thread(target=work)
# 设置守护线程
t1.setDaemon(True)
t1.start()
time.sleep(1)
print('主线程执行结束了!')
和进程对应，线程也支持 join() 方法，用于让主线程等待指定子线程执行完毕后再继续：
import threading
import time
def work():
time.sleep(2)
print("子线程任务完成")
t = threading.Thread(target=work)
t.start()
t.join() # 主线程阻塞等待
print("主线程继续执行")
3.6 线程间资源共享
结论：同一进程的多个线程，可以共享该进程的全局变量资源！
import threading, time
my_list = []
def write_data():
for i in range(1, 6):
my_list.append(i)
print(f'add: {i}')
print(f'write_data函数: {my_list}')
def read_data():
time.sleep(2)
print(f'read_data函数: {my_list}') # 能读到数据！
if __name__ == '__main__':
t1 = threading.Thread(target=write_data)
t2 = threading.Thread(target=read_data)
t1.start()
t2.start()
同样，主线程默认也会等待子线程结束。可以通过t1.setDaemon(True)或 t1.daemon = True 设置守护线程。
3.7 线程安全问题与互斥锁
问题产生
多个线程共享全局变量虽然方便，但引来了一个大Boss：多个线程同时修改同一个变量时，资源竞争导致数据不安全。比如两个线程各累加100万次，最终结果却不到200万。因为在 g_num += 1 这一步（读取、计算、写回）时，容易被其他线程打断。
根本原因：累加操作不是原子性的，线程在读取变量后、写回内存前，可能被其他线程抢占，导致数据覆盖。+= 不是原子操作。原子操作指的是不可被 CPU 调度打断、要么全部执行要么完全不执行的操作。g_num += 1 看似一行代码，实际分为三步：
- 读取全局变量 g_num 的值到寄存器
- 在寄存器中执行 +1 计算
- 将计算结果写回全局变量内存
这三步中间随时可能被其他线程抢占，导致两个线程读到同一个旧值，分别加 1 后写回，最终相当于只加了 1 次，出现数据丢失。
小知识：Python 中单个字节码指令是原子的，比如简单的赋值、单个方法调用；但复合运算、多行逻辑一定不是原子操作。
import threading
g_num = 0
def get_sum1():
for i in range(1000000):
global g_num
g_num += 1
print(f'get_sum1计算结果: {g_num}')
def get_sum2():
for i in range(1000000):
global g_num
g_num += 1
print(f'get_sum2计算结果: {g_num}')
if __name__ == '__main__':
t1 = threading.Thread(target=get_sum1)
t2 = threading.Thread(target=get_sum2)
t1.start()
t2.start()
多次运行会发现，最终结果几乎永远小于2000000。
解决方案：互斥锁
互斥锁（Lock）可以保证同一时间只有一个线程能操作共享数据，把非原子操作变成原子操作，彻底解决线程安全问题。
使用步骤：
- 创建锁对象：
mutex = threading.Lock() - 操作共享数据前加锁：
mutex.acquire() - 操作完毕释放锁：
mutex.release()
import threading
g_num = 0
# 创建互斥锁
mutex = threading.Lock()
def get_sum1():
mutex.acquire() # 加锁
for i in range(1000000):
global g_num
g_num += 1
mutex.release() # 释放锁
print(f'get_sum1计算结果: {g_num}')
def get_sum2():
mutex.acquire()
for i in range(1000000):
global g_num
g_num += 1
mutex.release()
print(f'get_sum2计算结果: {g_num}')
if __name__ == '__main__':
t1 = threading.Thread(target=get_sum1)
t2 = threading.Thread(target=get_sum2)
t1.start()
t2.start()
加锁后，无论运行多少次，最终结果都会是准确的2000000。
💡 注意
加锁会牺牲部分性能，换来数据安全；
避坑：必须使用同一把锁，且加锁范围要合理，千万别忘了 release()及时释放锁，否则会导致死锁！
好的，以下是为您博客新增的 3.8 死锁：成因、避免与调试 详细内容，可直接插入到 3.7 节之后。
3.8 死锁：成因、避免与调试
3.8.1 什么是死锁
死锁（Deadlock） 是指两个或多个线程（或进程）在执行过程中，因争夺资源而造成的一种互相等待的状态，导致所有线程都无法继续执行，程序永久阻塞。
死锁是并发编程中最棘手的 Bug 之一，它不像普通异常那样可以被捕获，一旦发生，程序就“僵死”在那里，既不报错也不退出。是多线程面试的高频考点。
3.8.2 死锁产生的四个必要条件
死锁必须同时满足以下四个条件才会发生（只要破坏任意一个，死锁就不会产生）：
- 互斥条件：资源在同一时刻只能被一个线程占用（如互斥锁）。
- 持有并等待条件：线程已经持有了至少一个资源，同时在等待其他线程持有的资源。
- 不可剥夺条件：线程已获得的资源在未使用完之前不能被其他线程强行剥夺，只能由持有者主动释放。
- 循环等待条件：存在一个线程等待链，每个线程都在等待下一个线程持有的资源，形成一个闭环。
3.8.3 经典死锁示例
import threading
import time
# 创建两把锁
lock1 = threading.Lock()
lock2 = threading.Lock()
def thread_a():
print("线程A: 尝试获取锁1...")
lock1.acquire()
print("线程A: 成功获取锁1")
time.sleep(0.1) # 模拟一些操作，确保死锁更容易发生
print("线程A: 尝试获取锁2...")
lock2.acquire()
print("线程A: 成功获取锁2")
lock2.release()
lock1.release()
print("线程A: 释放所有锁")
def thread_b():
print("线程B: 尝试获取锁2...")
lock2.acquire()
print("线程B: 成功获取锁2")
time.sleep(0.1)
print("线程B: 尝试获取锁1...")
lock1.acquire()
print("线程B: 成功获取锁1")
lock1.release()
lock2.release()
print("线程B: 释放所有锁")
if __name__ == '__main__':
t1 = threading.Thread(target=thread_a)
t2 = threading.Thread(target=thread_b)
t1.start()
t2.start()
t1.join()
t2.join()
print("程序结束")
运行后，程序大概率会永远卡住，不会打印“程序结束”。这是因为：
- 线程A拿到了锁1，等待锁2
- 线程B拿到了锁2，等待锁1
- 双方互不相让，形成循环等待
3.8.4 如何避免死锁
方法一：固定加锁顺序（最常用、最可靠）
所有线程按照相同的顺序获取锁，破坏循环等待条件。
def thread_a():
lock1.acquire()
lock2.acquire()
# ... 临界区代码 ...
lock2.release()
lock1.release()
def thread_b():
# 与线程A保持相同的加锁顺序：先lock1，再lock2
lock1.acquire()
lock2.acquire()
# ... 临界区代码 ...
lock2.release()
lock1.release()
方法二：使用超时机制（acquire(timeout)）
给 acquire 设置超时时间，如果指定时间内无法获取锁，就放弃并释放已持有的锁，避免死锁。
def thread_a():
if lock1.acquire(timeout=2):
print("线程A: 获取锁1成功")
time.sleep(0.1)
if lock2.acquire(timeout=2):
print("线程A: 获取锁2成功")
lock2.release()
else:
print("线程A: 获取锁2超时，释放锁1")
lock1.release()
else:
print("线程A: 获取锁1超时")
方法三：使用可重入锁 RLock
RLock（可重入锁）允许同一线程多次获取锁而不会死锁，适合递归调用或同一线程需要多次加锁的场景。但它不能解决不同线程间的死锁问题。
lock = threading.RLock()
def recursive_func(n):
if n <= 0:
return
lock.acquire()
print(f"递归第{n}层")
recursive_func(n - 1)
lock.release()
recursive_func(3) # 正常运行，不会死锁
方法四：使用 with 语句自动释放锁
with 语句可以确保锁在使用完毕后自动释放，即使发生异常也不会遗漏 release()，减少因忘记释放锁导致的潜在问题。
def safe_lock():
with lock1:
with lock2:
# 临界区代码
pass
3.8.5 如何调试死锁
死锁发生后，程序不会崩溃，而是“卡死”。可以通过以下手段排查：
- 使用
Ctrl+C中断程序，查看堆栈信息
当程序卡住时，按下 Ctrl+C，Python 会打印当前所有线程的堆栈，帮助你定位哪些线程在等待什么锁。
^C
Traceback (most recent call last):
File "deadlock.py", line 35, in thread_a
lock2.acquire()
KeyboardInterrupt
- 使用
threading.enumerate()查看活跃线程
import threading
import time
def check_threads():
while True:
threads = threading.enumerate()
print(f"当前活跃线程数: {len(threads)}")
for t in threads:
print(f" 线程: {t.name}, 是否存活: {t.is_alive()}")
time.sleep(1)
# 在死锁程序中启动监控线程
monitor = threading.Thread(target=check_threads, daemon=True)
monitor.start()
- 使用
logging模块记录锁的获取与释放
在生产环境中，可以通过日志记录每次加锁和解锁的操作，帮助事后分析死锁原因。
import logging
import threading
logging.basicConfig(level=logging.DEBUG)
lock1 = threading.Lock()
lock2 = threading.Lock()
def thread_a():
logging.debug("尝试获取锁1")
lock1.acquire()
logging.debug("成功获取锁1")
# ...
lock1.release()
logging.debug("释放锁1")
3.8.6 死锁检测工具
faulthandler模块：Python 3.3+ 内置，可以在程序卡住时通过信号触发打印所有线程的堆栈。py-spy：第三方工具，可以实时查看运行中 Python 程序的线程状态和堆栈。gdb+python-dbg：高级调试工具，适用于复杂的死锁场景。
死锁是并发编程中必须警惕的问题。养成良好加锁习惯（如固定顺序、使用 with），可以大大降低死锁风险。如果不幸遇到死锁，善用调试工具能帮你快速定位问题。
四、协程详解
4.1 什么是协程
协程（Coroutine），也叫“微线程”，是一种用户态下的轻量级线程。它允许在单线程内实现多任务的并发执行，任务的切换由程序自身控制，而不是由操作系统内核调度。
核心特点：
- 轻量级：一个线程中可以包含成千上万个协程，每个协程只占用极小的内存（约几 KB）。
- 协作式调度：协程主动让出执行权（通过 await），而不是被操作系统抢占。
- 无锁竞争：由于是单线程执行，不存在共享资源的锁冲突，避免了死锁和竞态条件。
总结： 协程最适合 IO 密集型 应用（如网络爬虫、Web 服务器、数据库访问），因为它能在等待 IO 时自动让出 CPU，转而执行其他协程，极大提高吞吐量。而对于 CPU 密集型计算，协程并不能带来加速（因为单线程），此时应使用多进程或多线程。
4.2 协程 vs 线程
- 线程：由操作系统调度，切换开销大，存在锁的开销
- 协程：由用户程序调度，切换开销极小，单线程内实现多任务，没有锁的开销
类比理解：
- 线程就像多个工人（线程）同时干活，但工人们需要争抢工具（锁），切换成本高。
- 协程就像一个工人（单线程）在不同任务之间快速切换，每次切换前主动保存现场，切换后恢复，几乎没有额外开销。
4.3 简单示例
Python中可以通过asyncio模块实现协程：
import asyncio
async def task1():
for i in range(3):
print(f'任务1执行中...{i}')
await asyncio.sleep(0.1)
async def task2():
for i in range(3):
print(f'任务2执行中...{i}')
await asyncio.sleep(0.1)
async def main():
await asyncio.gather(task1(), task2())
asyncio.run(main())
4.4 协程的底层原理：事件循环
Python 的协程基于 事件循环（Event Loop） 实现。事件循环是一个无限循环，负责监听和分发事件（如 IO 就绪、定时器到期），并调度协程的执行。
工作流程：
- 事件循环启动后，将协程对象注册为任务（Task）。
- 当协程执行到
await时，它会让出控制权，事件循环挂起该协程，并切换到下一个就绪的协程。 - 当
await的条件满足（如 IO 完成、定时器到达），事件循环会将协程唤醒，继续执行后续代码。 - 所有协程执行完毕，事件循环结束。
import asyncio
async def hello():
print("Hello")
await asyncio.sleep(1) # 模拟 IO 等待
print("World")
# 创建事件循环并运行
asyncio.run(hello())
注意：
asyncio.run()是 Python 3.7+ 引入的顶级入口，它会自动创建事件循环、运行协程，并在完成后关闭循环。
4.5 关键概念：协程、任务与 Future
- 协程函数：用
async def定义的函数，调用后返回一个协程对象。 - 协程对象：协程函数的返回值，需要通过
await或在事件循环中执行。 - 任务（Task）：对协程的进一步封装，用于并发调度。通过
asyncio.create_task()创建。 - Future：表示一个未来的结果，类似于 JavaScript 中的 Promise。
await的本质就是等待 Future 完成。
import asyncio
async def my_coro():
await asyncio.sleep(1)
return 42
async def main():
# 创建任务（立即开始调度）
task = asyncio.create_task(my_coro())
# 等待任务完成
result = await task
print(result) # 42
asyncio.run(main())
4.6 并发执行多个协程
asyncio.gather() 可以同时运行多个协程，并等待它们全部完成，返回结果列表。
import asyncio
async def fetch_data(url):
print(f"开始请求: {url}")
await asyncio.sleep(1) # 模拟网络延迟
print(f"完成请求: {url}")
return f"数据来自 {url}"
async def main():
urls = ["http://a.com", "http://b.com", "http://c.com"]
# 并发执行三个请求
results = await asyncio.gather(
fetch_data(urls[0]),
fetch_data(urls[1]),
fetch_data(urls[2])
)
print(results)
asyncio.run(main())
输出（三个请求几乎同时开始，总耗时约 1 秒）：
如果使用同步方式顺序执行，则需要 3 秒。协程的并发优势一目了然。
4.7 阻塞调用与线程池：正确混用同步与异步
1. 为什么需要阻塞调用与线程池？
asyncio的事件循环是单线程的，任何同步阻塞调用（如time.sleep(), requests.get(), 同步文件操作）都会卡死整个循环，导致所有协程无法执行。
解决方案： 使用loop.run_in_executor(executor, func, *args)将阻塞函数放到线程池（或进程池）中执行，让事件循环继续运行。
2. 正确做法
- 使用
aiohttp、aiofiles等异步库替代同步库。 - 对于无法避免的同步代码，用
run_in_executor。 - 注意线程池的生命周期管理，推荐使用
with ThreadPoolExecutor()。
3. 代码示例
# 示例：在协程中执行同步阻塞函数
import asyncio
import time
from concurrent.futures import ThreadPoolExecutor
def blocking_io():
"""模拟同步阻塞IO操作"""
print("开始阻塞IO...")
time.sleep(2) # 同步阻塞
print("阻塞IO完成")
return "结果"
async def main():
print("主协程开始")
loop = asyncio.get_running_loop()
# 错误方式：直接调用会阻塞事件循环
# result = blocking_io()
# 正确方式：放到线程池中执行
with ThreadPoolExecutor() as pool:
result = await loop.run_in_executor(pool, blocking_io)
print(f"获取到结果: {result}")
print("主协程结束")
asyncio.run(main())
4.8 常见误区与注意事项
- 协程不是并行：协程是单线程内的并发，无法利用多核 CPU。要实现真正的并行，需结合多进程（
multiprocessing）。 - 不要在协程中使用 time.sleep()：它会阻塞整个线程，导致所有协程停滞。应使用
await asyncio.sleep()。 - 协程中不要执行 CPU 密集型任务：长时间的计算会阻塞事件循环，导致其他协程无法执行。可将 CPU 密集任务放到线程池或进程池中。
- 避免在协程中直接调用同步 IO 库：如
requests、open()等，它们会阻塞线程。应使用异步库（如aiohttp、aiofiles）。 - asyncio.run() 只能调用一次：在一个脚本中通常只调用一次，作为入口。
4.9 本节核心总结
- 协程：用户态轻量级并发单元，切换开销极小，适合 IO 密集型场景。
- 事件循环：协程的调度引擎，负责管理任务的挂起与唤醒。
async/await：定义协程和挂起点的关键字。asyncio.gather()：并发运行多个协程，等待全部完成。- 注意事项：避免阻塞调用、CPU 密集型任务、使用专用异步库。
协程是现代 Python 高并发编程的核心，掌握它将使你能够编写高效的网络服务、爬虫和实时应用。下一章将介绍 Python 的垃圾回收机制，深入了解内存管理。
五、上下文管理器
5.1 从with open说起
我们操作文件时，通常会用with open语法，它的好处是：代码块执行完毕后，会自动关闭文件，无需手动调用close()，还能自动处理异常场景。
with open('./1.txt', 'r', encoding='utf-8') as f:
print(f.read())
# 离开with代码块后，文件自动关闭
为什么with open能自动释放资源？底层的核心就是上下文管理器。
5.2 自定义上下文管理器
一个类只要实现了__enter__()和__exit__()两个魔法方法，它就是上下文管理器：
__enter__()：进入with代码块前自动执行，返回值会赋值给as后的变量__exit__()：离开with代码块后自动执行，负责释放资源、处理异常
我们可以自定义一个文件类，模拟open的效果：
class MyFile(object):
def __init__(self, file_name, file_model):
self.file_name = file_name
self.file_model = file_model
self.fp = None
def __enter__(self):
# 进入with前执行：打开文件
self.fp = open(self.file_name, self.file_model, encoding='utf-8')
return self
def __exit__(self, exc_type, exc_val, exc_tb):
# 离开with后执行：关闭文件
self.fp.close()
print('释放资源成功!')
if __name__ == '__main__':
with MyFile('./1.txt', 'r') as mf:
print(mf.fp.read())
5.3 使用装饰器实现上下文管理器
除了通过类实现 __enter__ 和 __exit__ 方法，Python 还提供了一种更简洁的方式——使用 @contextmanager 装饰器。它可以将一个普通的生成器函数快速转变为上下文管理器。
5.3.1 装饰器方式的基本用法
from contextlib import contextmanager
@contextmanager
def my_open(file_name, mode):
fp = open(file_name, mode, encoding='utf-8')
try:
yield fp # yield 之前的代码相当于 __enter__
finally:
fp.close() # finally 中的代码相当于 __exit__
工作原理：
yield之前的代码在进入with块时执行（相当于__enter__）。yield后面的值会赋给as后的变量。yield之后的代码在离开with块时执行（相当于__exit__），无论是否发生异常都会执行finally块。
使用示例：
with my_open('test.txt', 'r') as f:
print(f.read())
# 离开 with 块后，文件自动关闭
5.3.2 与类方式的对比
| 对比项 | 类方式 | 装饰器方式 |
|---|---|---|
| 代码量 | 较多（需定义类、__enter__、__exit__） |
较少（一个函数 + yield） |
| 可读性 | 结构清晰，适合复杂逻辑 | 简洁直观，适合简单场景 |
| 异常处理 | 通过 __exit__ 参数手动处理 |
依赖 try/finally 自动处理 |
| 状态管理 | 可通过 self 维护状态 |
通过局部变量维护状态 |
选择建议：
- 如果上下文逻辑简单（如打开/关闭资源），优先使用装饰器方式。
- 如果需要精细控制异常处理（如抑制异常、记录日志），或者需要维护多个状态变量，建议使用类方式。
5.3.3 __exit__ 的三个参数（装饰器方式的底层机制）
虽然装饰器方式不需要我们手动编写 __exit__，但 @contextmanager 内部实际上还是利用了 __exit__ 机制。当 with 块中发生异常时，yield 语句会抛出异常，finally 块中的代码仍然会执行。如果希望抑制异常（不让它传播出去），可以在 yield 周围使用 try/except 捕获。
__exit__ 的三个参数在装饰器方式中同样存在，只不过由 contextmanager 内部自动处理：
- exc_type：异常类型，如果 with 块内无异常则为 None
- exc_val：异常对象，包含异常的具体信息
- exc_tb：异常追踪栈，记录异常发生的代码位置
如果 exit 返回 True，会抑制 with 块内的异常，不再向外抛出；返回 False 或无返回值，则异常会正常向上传递。利用这个特性，可以在上下文管理器中统一捕获和处理异常。
示例：在装饰器方式中捕获异常
from contextlib import contextmanager
@contextmanager
def safe_open(file_name, mode):
fp = None
try:
fp = open(file_name, mode, encoding='utf-8')
yield fp
except FileNotFoundError as e:
print(f"文件不存在: {e}")
# 不重新抛出异常，相当于抑制了异常
finally:
if fp:
fp.close()
with safe_open('不存在的文件.txt', 'r') as f:
if f:
print(f.read())
5.3.4 使用 contextmanager 的注意事项
yield只能有一个：生成器函数中只能有一个yield语句，用于将资源返回给with块。- 必须使用
try/finally：确保资源在发生异常时也能被正确释放。 - 不要忘记
yield后的代码：如果yield之后没有清理代码，资源将不会被释放。 @contextmanager只能用于简单的上下文管理器：如果需要同时管理多个资源或实现嵌套上下文，建议使用类方式或ExitStack。
5.3.5 简化写法：contextlib 装饰器
对于简单的上下文管理器，不需要写类，使用 contextlib.contextmanager 装饰器 + 生成器可以更简洁地实现，也是日常开发的主流写法：
from contextlib import contextmanager
@contextmanager
def my_open(file_name, mode):
# __enter__ 部分的逻辑
fp = open(file_name, mode, encoding='utf-8')
yield fp # yield 之前是进入逻辑，yield之后是退出逻辑
# __exit__ 部分的逻辑
fp.close()
print("释放资源成功!")
# 使用方式和原生with完全一致
with my_open('./1.txt', 'r') as f:
print(f.read())
yield 将代码分割为两部分：yield 之前的代码在进入 with 块时执行，对应 __enter__；yield 之后的代码在离开 with 块时执行，对应 __exit__。
5.3.6 实战：计时器上下文管理器
import time
from contextlib import contextmanager
@contextmanager
def timer(label="耗时"):
start = time.time()
try:
yield
finally:
end = time.time()
print(f"{label}: {end - start:.4f} 秒")
# 使用
with timer("数据处理"):
sum(range(10_000_000))
5.4 执行原理解析
上下文管理器的完整执行流程可以通过下图直观理解：
- 执行
MyFile('./1.txt', 'r')，调用__init__初始化对象 - 进入
with代码块前，自动调用__enter__方法，打开文件并返回对象给mf - 执行
with内部的代码，读取文件内容 - 离开
with代码块时，自动调用__exit__方法，关闭文件释放资源
5.5 如何选择多任务方案
面对不同的业务场景，选择合适的并发模型至关重要：
- CPU 密集型任务（如视频解码、科学计算）→ 多进程（绕过 GIL，利用多核）
- IO 密集型 + 高并发（如 Web 服务器、爬虫）→ 协程（单线程高效，切换开销最小）
- IO 密集型 + 中等规模（如文件处理、数据库操作）→ 多线程（简单易用，开发成本低）
- 需要分布式部署 → 多进程 + 消息队列（如 Celery）
口诀：CPU 密集用进程，IO 密集用协程，简单场景用线程。
算得多（CPU 密集）→ 选多进程，绕过 GIL 用满多核
等得多（IO 密集）→ 选多线程，简单好上手
并发高（海量 IO）→ 选协程，开销最小吞吐量最高
六、全文总结
1. 进程、线程、协程对比
| 特性 | 进程 | 线程 | 协程 |
|---|---|---|---|
| 调度者 | 操作系统 | 操作系统 | 用户程序 |
| 切换开销 | 最大（上下文切换、页表切换） | 较大（内核态切换） | 极小（用户态切换） |
| 内存占用 | 独立地址空间，占用大 | 共享地址空间，占用中等 | 共享地址空间，占用极小 |
| 数据共享 | 进程间通信（IPC）复杂 | 共享变量（需加锁） | 直接共享，无需锁 |
| GIL 影响 | 无影响（每个进程独立GIL） | 有影响（CPU密集受限） | 无影响（单线程） |
| 适用场景 | CPU 密集型、多核并行 | IO 密集型、中等并发 | 高并发 IO 密集型 |
| 创建数量 | 几十到几百 | 几千 | 几十万到百万 |
| 典型库 | multiprocessing |
threading |
asyncio |
2. 核心知识点
- 并发是交替执行，并行是真正同时执行；多核CPU才能实现并行
- 进程是资源分配单位，线程是调度执行单位；进程间数据隔离，线程间数据共享
- 多线程共享资源会引发线程安全问题，通过互斥锁可以解决
- 上下文管理器通过
__enter__和__exit__实现资源自动释放，是with语法的底层支撑
多任务编程的核心是平衡效率与安全，根据业务场景选择合适的多任务方案，才能写出高效且稳定的代码。
写在最后
到这里，我们已经完整走过了 Python 多任务编程的三大支柱：进程、线程与协程。从并发并行的概念辨析，到 GIL 的底层约束；从手动创建进程/线程，到进程池/线程池的优雅管理；从互斥锁解决线程安全，到死锁的预防与调试；再到协程的事件循环与异步编程，最后落地到上下文管理器的封装思想——这一路下来，相信你对 Python 的并发模型已经有了系统而深入的理解。
多任务编程的核心不在于掌握多少 API，而在于理解每种方案的适用边界。CPU 密集型用多进程，IO 密集型用协程，中等并发用线程——没有银弹，只有权衡。而无论选择哪种方案，线程安全、资源管理、死锁预防这三个问题始终如影随形，需要你在实践中不断积累经验。
最后送你一句话：“并发编程的难点不在于让代码跑起来，而在于让它一直正确地跑下去。” 希望这篇博客能成为你多任务编程路上的一个坚实起点。如果觉得有帮助，欢迎点赞、收藏、转发，你的支持是我持续输出的最大动力。有任何疑问或想法，欢迎在评论区留言交流。
更多推荐
3
0- 0
已为社区贡献3条内容
所有评论(0)