大家好，我是正在实战各种 AI 项目的程序员晚枫。
CPython 如何实现多线程？线程状态如何管理？线程间如何通信？这一讲，结合 GIL 深入理解 Python 的多线程机制。
📖 开篇：Python 线程不是操作系统的线程
Python 有自己的线程概念——threading 模块：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
| import threading
import time
def worker(n):
print(f'线程 {n} 开始')
time.sleep(1)
print(f'线程 {n} 结束')
threads = [threading.Thread(target=worker, args=(i,)) for i in range(3)]
for t in threads:
t.start()
for t in threads:
t.join()
print('全部完成')
|
Python 的 threading.Thread 在 CPython 中底层是操作系统的原生线程（pthread 或 Windows threads），但受 GIL 限制。
🧵 PyThreadState（线程状态）
每个 Python 线程都有一个 PyThreadState 对象：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
|
typedef struct _ts {
struct _ts *prev;
struct _ts *next;
PyInterpreterState *interp;
struct _frame *frame;
int recursion_depth;
PyObject *exc_type;
PyObject *exc_value;
PyObject *exc_traceback;
unsigned long thread_id;
char held;
} PyThreadState;
|
线程状态链表
1
2
3
4
5
| interp
↓
all_tstate链表: [tstate_0] <-> [tstate_1] <-> [tstate_2] <-> ...
↓ ↓ ↓
frame_a frame_b frame_c
|
🔄 线程调度
GIL + 线程调度 = 协作式调度
1
2
3
4
5
6
7
| 线程 A ──> 执行字节码（持有 GIL）
↓ (15ms 或 IO)
释放 GIL ───────────────────┐
↓
线程 B ──> 获取 GIL ──> 执行字节码 ──> 释放 GIL
↓ (IO 或时间片耗尽)
等待 GIL <───────────────────┘
|
线程切换的实际过程
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
| import threading
counter = 0
def increment():
global counter
for _ in range(10**6):
counter += 1
t1 = threading.Thread(target=increment)
t2 = threading.Thread(target=increment)
t1.start(); t2.start()
t1.join(); t2.join()
print(counter)
|
原因：counter += 1 其实是 4 条字节码指令，两线程交错执行时会丢失中间结果。
🛡️ 线程同步原语
Lock（互斥锁）
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
| import threading
lock = threading.Lock()
counter = 0
def safe_increment():
global counter
for _ in range(10**6):
with lock:
counter += 1
t1 = threading.Thread(target=safe_increment)
t2 = threading.Thread(target=safe_increment)
t1.start(); t2.start()
t1.join(); t2.join()
print(counter)
|
RLock（可重入锁）
同一个线程可以多次获取同一把锁：
1
2
3
4
5
6
7
8
9
10
11
12
13
| lock = threading.Lock()
def outer():
with lock:
inner()
def inner():
with lock:
print('OK')
rlock = threading.RLock()
|
Semaphore（信号量）
控制同时访问的线程数量：
1
2
3
4
5
6
7
8
9
|
semaphore = threading.Semaphore(5)
def access_resource():
with semaphore:
time.sleep(0.1)
threads = [threading.Thread(target=access_resource) for _ in range(20)]
|
📊 线程间通信
Queue（线程安全队列）
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
| import threading
import queue
q = queue.Queue()
def producer():
for i in range(10):
q.put(i)
print(f'生产: {i}')
def consumer():
while True:
item = q.get()
print(f'消费: {item}')
q.task_done()
producer_thread = threading.Thread(target=producer)
consumer_thread = threading.Thread(target=consumer, daemon=True)
producer_thread.start()
consumer_thread.start()
producer_thread.join()
|
Event（事件）
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
| import threading
import time
event = threading.Event()
def waiter():
print('等待中...')
event.wait()
print('收到信号！')
def setter():
time.sleep(2)
event.set()
t1 = threading.Thread(target=waiter)
t2 = threading.Thread(target=setter)
t1.start(); t2.start()
t1.join(); t2.join()
|
Condition（条件变量）
用于复杂的线程协调：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
| import threading
buffer = []
MAX_SIZE = 10
condition = threading.Condition()
def producer():
for i in range(20):
with condition:
while len(buffer) >= MAX_SIZE:
condition.wait()
buffer.append(i)
condition.notify()
def consumer():
while True:
with condition:
while len(buffer) == 0:
condition.wait()
item = buffer.pop(0)
condition.notify()
print(f'消费: {item}')
|
⚠️ 常见陷阱
陷阱1：守护线程
1
2
3
4
|
d = threading.Thread(target=background_task, daemon=True)
d.start()
|
陷阱2：死锁
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
|
lock_a = threading.Lock()
lock_b = threading.Lock()
def task1():
with lock_a:
time.sleep(0.1)
with lock_b:
print('task1 done')
def task2():
with lock_b:
time.sleep(0.1)
with lock_a:
print('task2 done')
|
💡 本节作业
- 验证 counter += 1 在多线程下的丢失更新问题
- 用 Lock 修复上述问题
- 用 Queue 实现一个生产者-消费者模式
🎯 本讲总结
PyThreadState：每个 Python 线程的状态对象，包含栈帧、异常状态、GIL 持有状态。
线程调度：GIL 释放 -> 调度器选择 -> GIL 获取 -> 执行，协作式调度。
同步原语：Lock（互斥）、RLock（可重入）、Semaphore（计数信号量）。
线程通信：Queue（队列）、Event（事件）、Condition（条件变量）。
📚 推荐教材
《Python 编程从入门到实践（第 3 版）》 | 《流畅的 Python（第 2 版）》 | 《CPython 设计与实现》
🔗 课程导航
← 上一讲：GIL 全局解释器锁 | 下一讲：模块导入系统 →
💬 联系我
主营业务：AI 编程培训、企业内训、技术咨询
🎓 AI 编程实战课程
想系统学习 AI 编程？程序员晚枫的 AI 编程实战课 帮你从零上手！
微信进群