线程基础与 GIL

Python 的线程模型让你在单个进程内并发执行多个任务。理解 GIL（全局解释器锁）是掌握 Python 线程的关键——它决定了什么任务适合用线程、什么任务应该选择其他方案。

GIL：全局解释器锁

GIL 是 CPython 解释器中的一个互斥锁，确保同一时刻只有一个线程执行 Python 字节码。这不是语言规范，而是 CPython 实现的技术选择。

为什么需要 GIL

CPython 的内存管理依赖引用计数。每个对象维护一个引用计数器，当计数降为零时对象被回收。引用计数的增减必须是原子操作，否则会出现：

# 假设没有 GIL 的场景（仅作说明）
obj = SomeObject()  # 引用计数 = 1

# 线程 A: obj2 = obj  → 读取计数(1) → 准备写入(2)
# 线程 B: del obj     → 读取计数(1) → 准备写入(0) → 对象被释放
# 线程 A: 写入计数(2) → 但对象已不存在！

为每个对象加锁会严重降低单线程性能。GIL 用一个全局锁保护整个解释器状态，换取简单和效率。

GIL 的实际影响

CPU 密集型任务受限：计算素数时，即使创建多个线程，也只有一个线程在执行 Python 代码：

import threading
import time

def compute_primes(n):
    """计算前 n 个素数"""
    primes = []
    num = 2
    while len(primes) < n:
        is_prime = True
        for p in primes:
            if p * p > num:
                break
            if num % p == 0:
                is_prime = False
                break
        if is_prime:
            primes.append(num)
        num += 1
    return primes

# 单线程
start = time.time()
compute_primes(5000)
single_time = time.time() - start
print(f"单线程: {single_time:.2f}秒")

# 两个线程
start = time.time()
t1 = threading.Thread(target=compute_primes, args=(2500,))
t2 = threading.Thread(target=compute_primes, args=(2500,))
t1.start()
t2.start()
t1.join()
t2.join()
multi_time = time.time() - start
print(f"双线程: {multi_time:.2f}秒")
print(f"加速比: {single_time / multi_time:.2f}x")

输出示例：

单线程: 1.23秒
双线程: 1.31秒
加速比: 0.94x

双线程反而更慢，因为线程切换开销超过了并行收益。

I/O 密集型任务有效：当线程等待网络、磁盘或用户输入时，GIL 会释放，其他线程可以执行：

import threading
import time

def download_simulation(url):
    """模拟下载（实际是 sleep）"""
    print(f"开始下载 {url}")
    time.sleep(2)  # sleep 期间会释放 GIL
    print(f"完成下载 {url}")

# 单线程
start = time.time()
for i in range(3):
    download_simulation(f"file{i}.dat")
single_time = time.time() - start
print(f"单线程总耗时: {single_time:.2f}秒\n")

# 多线程
start = time.time()
threads = []
for i in range(3):
    t = threading.Thread(target=download_simulation, args=(f"file{i}.dat",))
    threads.append(t)
    t.start()

for t in threads:
    t.join()
multi_time = time.time() - start
print(f"多线程总耗时: {multi_time:.2f}秒")
print(f"加速比: {single_time / multi_time:.2f}x")

输出示例：

单线程总耗时: 6.01秒

开始下载 file0.dat
开始下载 file1.dat
开始下载 file2.dat
完成下载 file0.dat
完成下载 file1.dat
完成下载 file2.dat
多线程总耗时: 2.01秒
加速比: 2.99x

三个下载几乎同时完成，因为 time.sleep() 释放了 GIL。

何时释放 GIL

这些操作会释放 GIL：

• I/O 操作（read(), write(), send(), recv()）
• time.sleep()
• C 扩展中显式释放（如 NumPy 的矩阵运算）
• 等待锁或条件变量

纯 Python 计算代码不会主动释放 GIL，但解释器每执行约 5ms 字节码会切换线程（可配置）。

创建和管理线程

Thread 类的基本用法

import threading
import time

def worker(name, delay):
    """工作线程函数"""
    print(f"{name} 开始")
    time.sleep(delay)
    print(f"{name} 完成")

# 方式1：传递目标函数
t1 = threading.Thread(target=worker, args=("线程A", 1))
t1.start()  # 启动线程
t1.join()   # 等待线程结束

# 方式2：继承 Thread 类
class WorkerThread(threading.Thread):
    def __init__(self, name, delay):
        super().__init__()
        self.worker_name = name
        self.delay = delay
    
    def run(self):
        """线程的入口方法"""
        print(f"{self.worker_name} 开始")
        time.sleep(self.delay)
        print(f"{self.worker_name} 完成")

t2 = WorkerThread("线程B", 1)
t2.start()
t2.join()

守护线程（Daemon Thread）

守护线程在主程序退出时自动终止，不会阻止程序退出：

import threading
import time

def background_task():
    while True:
        print("后台任务运行中...")
        time.sleep(1)

# 普通线程
normal_thread = threading.Thread(target=background_task)
normal_thread.start()
# 如果取消下面的注释，程序永远不会退出
# time.sleep(3)

# 守护线程
daemon_thread = threading.Thread(target=background_task, daemon=True)
daemon_thread.start()
time.sleep(3)  # 主线程等待3秒
print("主程序退出")
# daemon_thread 会自动终止

使用场景：

• 后台日志写入
• 定期清理缓存
• 监控服务状态

注意：守护线程可能在任意时刻被强制终止，不应执行需要清理资源的操作（如关闭文件、数据库连接）。

线程生命周期

import threading
import time

def task():
    time.sleep(1)

t = threading.Thread(target=task)
print(f"创建后: {t.is_alive()}")  # False

t.start()
print(f"启动后: {t.is_alive()}")  # True

t.join()  # 等待线程结束
print(f"结束后: {t.is_alive()}")  # False

join(timeout) 可以设置超时：

import threading
import time

def slow_task():
    time.sleep(5)

t = threading.Thread(target=slow_task)
t.start()

# 最多等待2秒
t.join(timeout=2)

if t.is_alive():
    print("线程仍在运行")
else:
    print("线程已完成")

线程同步

多个线程访问共享数据时，必须同步以避免竞态条件。

Lock：互斥锁

import threading

counter = 0
lock = threading.Lock()

def increment():
    global counter
    for _ in range(100000):
        lock.acquire()
        counter += 1
        lock.release()

# 不使用锁的错误示例
counter = 0
threads = [threading.Thread(target=lambda: globals().__setitem__('counter', counter + 1) or None) 
           for _ in range(100000)]
# 上面的例子仍然不安全，仅作说明

# 正确示例
counter = 0
threads = [threading.Thread(target=increment) for _ in range(2)]
for t in threads:
    t.start()
for t in threads:
    t.join()

print(f"最终计数: {counter}")  # 应该是 200000

使用上下文管理器更安全：

def increment_safe():
    global counter
    for _ in range(100000):
        with lock:  # 自动 acquire 和 release
            counter += 1

RLock：可重入锁

允许同一线程多次获取锁：

import threading

rlock = threading.RLock()

def recursive_function(n):
    with rlock:
        print(f"层级 {n}")
        if n > 0:
            recursive_function(n - 1)  # 同一线程再次获取 rlock

recursive_function(3)

如果用普通 Lock，第二次 acquire() 会死锁。

Semaphore：信号量

限制同时访问资源的线程数量：

import threading
import time

# 最多3个线程同时访问
semaphore = threading.Semaphore(3)

def access_resource(name):
    print(f"{name} 等待...")
    with semaphore:
        print(f"{name} 获得访问权")
        time.sleep(2)
    print(f"{name} 释放")

threads = [threading.Thread(target=access_resource, args=(f"线程{i}",)) 
           for i in range(6)]

for t in threads:
    t.start()
for t in threads:
    t.join()

输出示例：

线程0 等待...
线程1 等待...
线程2 等待...
线程3 等待...
线程4 等待...
线程5 等待...
线程0 获得访问权
线程1 获得访问权
线程2 获得访问权
（2秒后）
线程0 释放
线程3 获得访问权
线程1 释放
线程4 获得访问权
...

Event：事件信号

用于线程间简单的通知机制：

import threading
import time

event = threading.Event()

def waiter():
    print("等待事件...")
    event.wait()  # 阻塞直到事件被设置
    print("事件已触发！")

def setter():
    time.sleep(2)
    print("设置事件")
    event.set()  # 唤醒所有等待的线程

t1 = threading.Thread(target=waiter)
t2 = threading.Thread(target=setter)

t1.start()
t2.start()
t1.join()
t2.join()

Event 常用于启动信号、停止标志：

stop_event = threading.Event()

def worker():
    while not stop_event.is_set():
        print("工作中...")
        time.sleep(1)
    print("收到停止信号")

t = threading.Thread(target=worker)
t.start()
time.sleep(3)
stop_event.set()  # 通知线程停止
t.join()

Condition：条件变量

结合锁和等待/通知机制：

import threading
import time
import random

condition = threading.Condition()
items = []

def producer():
    for i in range(5):
        time.sleep(random.uniform(0.1, 0.5))
        with condition:
            items.append(i)
            print(f"生产: {i}")
            condition.notify()  # 通知消费者

def consumer():
    while True:
        with condition:
            while not items:  # 必须在循环中检查
                condition.wait()  # 释放锁并等待
            item = items.pop(0)
            print(f"消费: {item}")
            if item == 4:  # 最后一个
                break

t1 = threading.Thread(target=producer)
t2 = threading.Thread(target=consumer)
t1.start()
t2.start()
t1.join()
t2.join()

线程间通信

Queue：线程安全队列

queue.Queue 内部使用锁，是线程间传递数据的首选：

import threading
import queue
import time

q = queue.Queue()

def producer(name):
    for i in range(3):
        time.sleep(0.5)
        q.put(f"{name}-{i}")
        print(f"{name} 生产: {i}")

def consumer(name):
    while True:
        item = q.get()  # 阻塞直到有数据
        if item is None:  # 哨兵值表示结束
            break
        print(f"{name} 消费: {item}")
        q.task_done()  # 标记任务完成

# 启动生产者和消费者
producers = [threading.Thread(target=producer, args=(f"P{i}",)) for i in range(2)]
consumers = [threading.Thread(target=consumer, args=(f"C{i}",)) for i in range(2)]

for t in producers + consumers:
    t.start()

for t in producers:
    t.join()

# 发送停止信号
for _ in consumers:
    q.put(None)

for t in consumers:
    t.join()

Queue 的其他方法：

• put(item, block=True, timeout=None): 放入数据
• get(block=True, timeout=None): 取出数据
• put_nowait(item): 非阻塞放入，满时抛出 queue.Full
• get_nowait(): 非阻塞取出，空时抛出 queue.Empty
• qsize(): 近似大小（不精确）
• empty(), full(): 状态检查（不可靠，仅供参考）

线程局部存储

每个线程拥有独立的变量副本：

import threading

thread_local = threading.local()

def worker(value):
    thread_local.data = value  # 每个线程独立
    print(f"{threading.current_thread().name}: {thread_local.data}")

threads = [threading.Thread(target=worker, args=(i,)) for i in range(3)]
for t in threads:
    t.start()
for t in threads:
    t.join()

输出示例：

Thread-1: 0
Thread-2: 1
Thread-3: 2

使用场景：

• 数据库连接池（每个线程独立连接）
• HTTP 请求上下文
• 线程专属缓存

常见陷阱与调试

竞态条件

# 错误示例
class BankAccount:
    def __init__(self):
        self.balance = 0
    
    def deposit(self, amount):
        # 读取、计算、写入三步不是原子操作
        current = self.balance
        current += amount
        self.balance = current

# 正确示例
class BankAccount:
    def __init__(self):
        self.balance = 0
        self.lock = threading.Lock()
    
    def deposit(self, amount):
        with self.lock:
            self.balance += amount

死锁

两个线程互相等待对方持有的锁：

lock1 = threading.Lock()
lock2 = threading.Lock()

def thread_a():
    with lock1:
        time.sleep(0.1)
        with lock2:  # 等待 lock2
            print("Thread A")

def thread_b():
    with lock2:
        time.sleep(0.1)
        with lock1:  # 等待 lock1
            print("Thread B")

# 这会死锁！

预防死锁：

1. 固定锁的获取顺序
2. 使用 timeout 参数
3. 使用 RLock 或无锁数据结构

# 解决方案：统一顺序
def thread_a():
    with lock1:
        with lock2:
            print("Thread A")

def thread_b():
    with lock1:  # 与 thread_a 顺序一致
        with lock2:
            print("Thread B")

实战案例：多线程 HTTP 下载器

import threading
import queue
import time
from urllib.request import urlopen

def download_file(url):
    """下载文件并返回大小"""
    try:
        with urlopen(url, timeout=10) as response:
            data = response.read()
            return len(data)
    except Exception as e:
        return f"错误: {e}"

def worker(task_queue, result_queue):
    """工作线程"""
    while True:
        url = task_queue.get()
        if url is None:  # 停止信号
            break
        
        size = download_file(url)
        result_queue.put((url, size))
        task_queue.task_done()

# 待下载的URL列表
urls = [
    "https://www.python.org",
    "https://docs.python.org/3/",
    "https://pypi.org",
]

task_queue = queue.Queue()
result_queue = queue.Queue()

# 创建工作线程
num_threads = 3
threads = []
for _ in range(num_threads):
    t = threading.Thread(target=worker, args=(task_queue, result_queue))
    t.start()
    threads.append(t)

# 添加任务
start_time = time.time()
for url in urls:
    task_queue.put(url)

# 等待所有任务完成
task_queue.join()

# 停止工作线程
for _ in range(num_threads):
    task_queue.put(None)

for t in threads:
    t.join()

# 收集结果
results = []
while not result_queue.empty():
    results.append(result_queue.get())

elapsed = time.time() - start_time
print(f"\n下载完成，耗时: {elapsed:.2f}秒")
for url, size in results:
    print(f"{url}: {size} 字节")

小结

• GIL 限制了 CPU 密集型任务的并行，但不影响 I/O 密集型任务
• Lock 是最基本的同步原语，防止竞态条件
• Queue 是线程间通信的首选，内部已加锁
• Event 用于简单的通知机制
• Condition 适合生产者-消费者模式
• threading.local 为每个线程提供独立存储
• 始终警惕死锁和竞态条件

下一节学习 多进程，它通过独立内存空间绕过 GIL 限制。

---

练习：答案