Python多线程编程

6621 字
17 分钟

Python多线程编程

发布于

Python多线程编程

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

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

背景 / 动机

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

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

前置知识:

  • Python 函数与类的基础用法

核心内容

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

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

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
共享数据要加锁   →  否则结果随机

参考资料

Last updated on