Python多线程编程

发布于 2026年8月1日

Python多线程编程

线程 = 一条独立的执行流水线。 Python 里因为 GIL,多线程跑不满多核 CPU,但对网络请求、文件读写这类 I/O 密集型任务,依然是显著的提速利器。

类比：奶茶店的多个「下单窗口」—— 客人再多也能并行接待(I/O 等待);但如果只有一个收银员结账(GIL),纯算账这种计算活儿,窗口再多也快不起来。

背景 / 动机

程序为什么会卡?最常见的场景不是 CPU 忙,而是在等——等网络响应、等磁盘读写、等数据库返回。等待期间 CPU 其实是空闲的,白白浪费。

多线程的思路就是：让一个人的等待时间,变成另一个人的工作时间。

前置知识：

• Python 函数与类的基础用法

核心内容

1. 核心概念：线程、进程与 GIL

概念	是什么	内存	开销
进程	独立运行的程序实例	相互隔离	重
线程	进程内的执行单元	共享	轻
GIL	CPython 的全局解释器锁	—	—

GIL 一句话：CPython 同一时刻只允许一个线程执行 Python 字节码。但它会在线程发生 I/O 等待时自动让出,所以：

时间线（两个线程共享一个 GIL）
──────────────────────────────────────────
线程A: [计算] [发起网络请求 → 让出GIL] [计算] [完成]
线程B:         [  获得GIL → 趁机计算  ]
──────────────────────────────────────────

别被「Python 多线程没用」带偏
GIL 只卡CPU 密集型任务。I/O 密集时线程在等待中让出 GIL,照样能并行提速。

	多进程 multiprocessing	多线程 threading	异步 asyncio
适合场景	CPU 密集型	I/O 密集型	I/O 密集型(单线程)
利用多核	✅	❌ 受 GIL 限制	❌
资源开销	大	小	最小
上手难度	中	低	中(async/await)
典型场景	图像处理、数值计算	爬虫、批量请求	高并发服务

2. 关键写法

2.1 创建线程：threading.Thread

import threading
import time

def download(url):
    print(f"🔽 开始下载 {url}")
    time.sleep(2)              # 模拟网络等待（会释放 GIL）
    print(f"✅ 完成 {url}")

t = threading.Thread(target=download, args=("https://a.com",))
t.start()                      # 启动线程
t.join()                       # 主线程等它跑完

批量并发：

urls = ["https://a.com", "https://b.com", "https://c.com"]

threads = [threading.Thread(target=download, args=(u,)) for u in urls]
for t in threads:
    t.start()
for t in threads:
    t.join()                   # 逐个等待，全部跑完再继续

join() 不能省

1. 主线程/主进程的代码执行完毕后，Python 解释器不会立即退出。
2. 解释器会检查是否还有非守护的子线程和子进程在运行。
3. 有 → 解释器继续等待；
   没有 → 解释器退出，所有守护线程和守护子进程被强制销毁。

简记：所有非守护的子线程和子进程都结束，程序才真正退出。
join() 阻塞调用线程（通常是主线程），直到被 join 的线程终止，或超时。
❌ 没有 join —— 可能在子线程结束前就用了结果；
✅ 有 join —— 确保拿到结果。

守护线程：

import threading
import time

def daemon_task():
    while True:
        print('守护线程正在运行中...')
        time.sleep(1)

def normal_task():
    time.sleep(3)
    print('普通子线程结束')

if __name__ == '__main__':
    # daemon=True，说明 t1 是守护线程，主线程结束，t1 直接被回收
    t1 = threading.Thread(target=daemon_task, daemon=True)
    # t2 不是守护线程，主线程要等待 t2 结束之后才能结束
    t2 = threading.Thread(target=normal_task)

    t1.start()
    t2.start()
    t2.join()
    print('主线程执行完毕，守护线程直接销毁...')

2.2 线程安全：Lock

多个线程同时改同一个变量会发生竞态：

import threading
import time

num = 0

def add():
    global num
    for _ in range(100_000):
        current = num
        time.sleep(0)  # 不睡，但让出 GIL，立刻重新排队
        num = current + 1  # 非原子操作，多线程互相覆盖 → 结果偏小

if __name__ == "__main__":
    t1 = threading.Thread(target=add)
    t2 = threading.Thread(target=add)
    t1.start()
    t2.start()
    t1.join()
    t2.join()
    print("预期200_000，实际结果：", num)    # 期望 200_000，实际通常不是

2.2.1 互斥锁

lock = threading.Lock()

def add():
    global num
    for _ in range(100_000):
        with lock:        # 等价于 acquire() + try/finally + release()
            current = num
            time.sleep(0)
            num = current + 1

临界区要尽量小
with lock: 包裹的代码越短越好，锁太久等于把并发又变回串行。

2.2.2 可重入锁 RLock

函数嵌套调用时需要重复获取锁，普通 Lock 同一线程重复 acquire 会死锁，RLock 支持同一线程多次加锁，需要同等次数释放：

import threading

rlock = threading.RLock()

def transfer(account_from, account_to, amount):
    with rlock:
        # 获取锁后，调用内部方法
        deduct(account_from, amount)
        deposit(account_to, amount)

def deduct(account, amount):
    with rlock:      # 同一线程再次获取，RLock 允许
        account.balance -= amount

def deposit(account, amount):
    with rlock:      # 同一线程再次获取，RLock 允许
        account.balance += amount

死锁规避方案

1. 统一所有线程获取锁的顺序；
2. 设置 acquire 超时 lock.acquire(timeout=3)；
3. 减少嵌套锁；
4. 使用 RLock 替代普通 Lock。

2.3 线程同步工具

2.3.1 信号量 Semaphore：限制最大并发线程

控制同时运行线程数量：

'''
    Lock           → 独占，一次 1 个      → 保护共享资源
    Semaphore(n)   → 共享，一次 n 个      → 限制并发数量
    RLock          → 可重入，同一线程多次获取 → 函数嵌套调用
'''
import threading
import time

# 同时允许 3 个线程访问
sem = threading.Semaphore(3)

def access_resource(thread_id):
    print(f"线程{thread_id}: 等待进入...")
    with sem:
        print(f"线程{thread_id}: ✅ 进入，正在使用资源")
        time.sleep(2)
        print(f"线程{thread_id}: ❌ 离开")

if __name__ == "__main__":
    threads = []
    for i in range(8):
        t = threading.Thread(target=access_resource, args=(i,))
        threads.append(t)
        t.start()

    for t in threads:
        t.join()
    print("全部完成")

2.3.2 事件 Event：线程间通知机制

核心方法：

• event.set()：设置标志为 True，唤醒等待线程
• event.wait(timeout)：阻塞等待标志为 True
• event.clear()：重置标志为 False

import threading
import time

event = threading.Event()

def wait_task():
    print('子线程等待信号...')
    event.wait()    # 阻塞，直到 event 被 set
    print('收到信号，并且开始执行任务')

def send_signal():
    time.sleep(3)
    print('发送通知信号')
    event.set()    # 通知所有等待的线程

if __name__ == '__main__':
    t1 = threading.Thread(target=wait_task)
    t2 = threading.Thread(target=send_signal)
    t1.start()
    t2.start()
    t1.join()
    t2.join()

2.3.3 条件变量 Condition：复杂生产消费模型

Condition = 锁 + 等待队列。线程在条件不满足时 wait() 等待，条件满足时被 notify_all() 唤醒。
适合「生产者-消费者」这类一写多读的复杂模型。

wait() / notify_all() 机制：

wait() 三步曲：释放锁 → 进入等待队列 → 阻塞
notify_all()：唤醒等待队列 → 回到锁池重新抢锁

被唤醒 ≠ 立即执行：抢到锁的线程才从 wait() 的下一行继续

锁池 vs 等待队列：

锁池     ：等待拿锁的线程，参与锁竞争
等待队列 ：已 wait() 的线程，不参与竞争
wait() → 进等待队列；notify_all() → 回到锁池抢锁

完整可运行示例：

import threading
import time
import random

BUFFER = []                    # 共享仓库
MAX, TOTAL = 5, 20             # 仓库容量 / 总任务数
produced = consumed = 0
done = False
cond = threading.Condition()   # 锁 + 等待队列

def producer(name):
    global produced, done
    while True:
        with cond:
            if produced >= TOTAL:
                done = True
                cond.notify_all()
                return
            while len(BUFFER) >= MAX and not done:
                cond.wait()             # 仓库满 → 等待
            if done:                    # 第二次检查，防止过度生产
                cond.notify_all()
                return
            produced += 1
            BUFFER.append(f"{name}-P{produced}")
            cond.notify_all()           # 唤醒消费者
        time.sleep(random.uniform(0.1, 0.3))

def consumer(name):
    global consumed, done
    while True:
        with cond:
            while not BUFFER and not done:
                cond.wait()             # 仓库空 → 等待
            if not BUFFER:              # 没货且生产结束 → 退出
                done = True
                cond.notify_all()
                return
            item = BUFFER.pop(0)
            consumed += 1
            if consumed >= TOTAL:
                done = True
                cond.notify_all()
                return
            cond.notify_all()
        time.sleep(random.uniform(0.1, 0.3))

if __name__ == "__main__":
    ts = [threading.Thread(target=producer, args=(f"P{i+1}",)) for i in range(2)] + \
         [threading.Thread(target=consumer, args=(f"C{i+1}",)) for i in range(3)]
    for t in ts:
        t.start()
    for t in ts:
        t.join()
    print(f"完成！总生产 {produced}，总消费 {consumed}")

为什么用while 不是 if？
被唤醒 ≠ 条件一定满足：可能有虚假唤醒，或货已被其他消费者抢走。必须 while 重新检查，否则会越界 pop。

惊群效应：notify_all() 唤醒全部线程，但只有抢到锁的那个能干活，其余又回去等待，浪费 CPU。单个唤醒用 notify()，但可能造成线程饥饿。

2.4 线程池：ThreadPoolExecutor

手写线程又要 start 又要 join，还很啰嗦。用线程池：

from concurrent.futures import ThreadPoolExecutor
import time

def download(url):
    time.sleep(1)
    return f"✅ {url} 完成"

urls = [f"https://a.com/page{i}" for i in range(10)]

with ThreadPoolExecutor(max_workers=5) as pool:   # 最多 5 个并发
    results = list(pool.map(download, urls))       # 按顺序拿回结果

for r in results:
    print(r)

默认优先用线程池
自动回收线程、能取返回值、可限并发数，代码更短。只有极特殊的场景才需要手写 Thread。

3. 判断标准：该不该用多线程？

场景	结论
大量网络请求 / 文件读写，CPU 大部分时间在等	✅ 多线程
纯计算（循环、图像处理），想加速	❌ 换多进程
海量连接的网络服务	🤔 考虑 asyncio
任务很少且很快	❌ 别用，单线程足够

实战示例：并发下载并统计耗时

import time
import urllib.request
from concurrent.futures import ThreadPoolExecutor

URLS = [
    "https://www.example.com/",
    "https://docs.python.org/3/",
    "https://www.google.com/",
]

def fetch(url):
    with urllib.request.urlopen(url, timeout=5) as resp:
        return len(resp.read())

def run(mode, func, urls):
    start = time.time()
    result = func(urls)
    print(f"{mode}：{time.time()-start:.2f}s, 返回字节数 {result}")
    return result

if __name__ == "__main__":
    # 顺序执行
    run("顺序", lambda us: [fetch(u) for u in us], URLS)

    # 多线程并发
    with ThreadPoolExecutor(max_workers=3) as pool:
        run("并发", lambda us: list(pool.map(fetch, us)), URLS)

实测预期
网络请求大部分时间在等响应 → 并发通常比顺序快 2~3 倍。换成纯 CPU 计算时，多线程反而更慢 —— 这是 GIL 的锅，不是线程的错。

常见坑 & 避坑指南

坑	现象	解决方案
忘记 join()	主线程不等子线程就继续，用结果时可能还没算好（守护线程则会被强制终止）	需要结果时先 t.join()
共享变量不加锁	计数结果随机偏小（竞态）	with lock: 保护临界区
CPU 密集任务用多线程	反而更慢	改用 multiprocessing
锁里再申请同一把锁	死锁，程序卡死	改用 RLock（可重入锁）
线程内直接 print	输出乱序穿插	接受，或收集结果后统一打印
任务很快却硬开线程	启动开销 > 收益	单线程就够了

最常见的误解
一看程序慢就无脑加线程 → CPU 密集 → 更慢。判断标准一句话：等 I/O 用线程，算 CPU 用进程。

结语

极简口诀：

网络/文件等待多  →  多线程提速明显
纯计算跑不动     →  换多进程
海量连接服务     →  上 asyncio
共享数据要加锁   →  否则结果随机

参考资料

• Python threading 官方文档
• concurrent.futures 官方文档
• Python GIL 详解（Real Python）
• asyncio 官方文档