【一问专栏】从零开始理解 Python 的 GIL：一把让人又爱又恨的锁

import threading

def worker():
    for i in range(5):
        print(f"线程 {threading.current_thread().name} 输出: {i}")

# 创建两个线程
t1 = threading.Thread(target=worker, name="线程A")
t2 = threading.Thread(target=worker, name="线程B")

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

看到输出结果时我很困惑：为什么两个线程的输出是交替出现的，而不是真正的"同时"打印？这就是我第一次遇见 GIL 的时刻。

GIL 是什么？

GIL（Global Interpreter Lock，全局解释器锁）是 CPython 解释器中的一个机制。简单来说，它是一把保证同一时刻只有一个线程执行 Python 代码的锁。这意味着即使在多核处理器上，使用多线程的 Python 程序也无法实现真正的并发执行。
因为GIL 实际上是一个互斥锁，它保护着 CPython 解释器的内部状态。当线程想要执行 Python 字节码时，必须先获取这个锁。

用一个生活中的比喻来解释：

想象 Python 解释器是一个厨房，线程是厨师。GIL 就像是厨房里唯一的一把厨刀。即使有很多厨师（线程），但任何时候只有一个人能拿着这把厨刀切菜（执行 Python 代码）。其他厨师要么等待，要么去做不需要厨刀的工作（比如等待水烧开）。

它的工作机制复杂的多

1. 字节码执行保护

Python 代码会被编译成字节码，GIL 确保在执行这些字节码时是线程安全的。

import dis

def simple_function(x):
    return x + 1

# 查看字节码
print("函数字节码:")
dis.dis(simple_function)

每个字节码指令的执行都是原子的，但多个字节码指令组成的操作可能不是原子的，这就是 GIL 需要保护的地方。

2. 检查间隔机制

GIL 不是永久持有的，线程执行一定数量的字节码指令后，会主动释放 GIL，让其他线程有机会执行。

import threading

def demonstrate_gil_switching():
    counter = 0
    
    def increment_counter(thread_name):
        nonlocal counter
        for i in range(100):
            # 在 GIL 的保护下修改变量
            current = counter
            # 模拟一些计算
            result = sum(range(1000))
            counter = current + 1
            print(f"{thread_name}: counter = {counter}")
    
    t1 = threading.Thread(target=increment_counter, args=("Thread-1",))
    t2 = threading.Thread(target=increment_counter, args=("Thread-2",))
    
    t1.start()
    t2.start()
    t1.join()
    t2.join()
    
    print(f"最终结果: {counter}")

3. I/O 操作的 GIL 释放

当线程执行 I/O 操作时，会自动释放 GIL，这让其他线程可以在此期间执行。

import threading
import time

def io_operation(thread_id):
    print(f"线程 {thread_id} 开始执行，获取 GIL")
    
    # 模拟计算（持有 GIL）
    start = time.time()
    while time.time() - start < 1:
        pass  # 占用 CPU
    print(f"线程 {thread_id} 完成计算，准备 I/O")
    
    # I/O 操作会释放 GIL
    time.sleep(2)  # 模拟 I/O 等待，期间释放 GIL
    
    print(f"线程 {thread_id} I/O 完成，重新获取 GIL")

# 创建多个线程演示 GIL 释放
threads = []
for i in range(3):
    t = threading.Thread(target=io_operation, args=(i,))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

为什么 Python 需要这把"锁"？

通过学习，我了解到 GIL 的存在主要有两个原因：

1. 内存管理的安全网

Python 使用引用计数来管理内存。每个对象都有一个计数，记录有多少变量指向它。当计数为 0 时，对象就会被销毁。

如果没有 GIL，两个线程可能同时修改同一个对象的引用计数，导致计数错误，进而引发内存泄漏或者程序崩溃。

2. C 扩展的兼容性

很多 Python 库是用 C 语言写的，GIL 简化了这些 C 扩展的编写，因为它们不需要自己处理复杂的线程安全问题。

GIL 的实际影响

让我通过代码来感受 GIL 的影响：

import threading
import time

# CPU 密集型任务
def cpu_task(n):
    count = 0
    for i in range(n):
        count += i
    return count

# 测试单线程
start_time = time.time()
cpu_task(10**7)
cpu_task(10**7)
single_time = time.time() - start_time

# 测试多线程
start_time = time.time()
t1 = threading.Thread(target=cpu_task, args=(10**7,))
t2 = threading.Thread(target=cpu_task, args=(10**7,))

t1.start()
t2.start()
t1.join()
t2.join()
multi_time = time.time() - start_time

print(f"单线程执行时间: {single_time:.4f}秒")
print(f"多线程执行时间: {multi_time:.4f}秒")

运行这个代码，你会发现多线程版本可能比单线程还要慢！这就是 GIL 的"魔力"——它让多线程在 CPU 密集型任务中变得低效。

重要区分：GIL 不是单例模式

在学习过程中，我一度混淆了 GIL 和单例模式，但它们完全不同：

GIL

是什么：解释器级别的机制
为什么存在：为了保证线程安全和内存管理
控制什么：控制线程执行权限
能否避免：在 CPython 中无法避免

单例模式

是什么：程序设计模式
为什么使用：为了确保类只有一个实例
控制什么：控制对象创建
能否避免：开发者可以选择是否使用

# 单例模式示例
class DatabaseConnection:
    _instance = None
    
    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance

# 使用单例模式
db1 = DatabaseConnection()
db2 = DatabaseConnection()
print(db1 is db2)  # True - 这是我们的设计选择

# GIL 是解释器强加的，我们无法选择

简单来说：单例模式是我们主动选择的设计，GIL 是解释器强加给我们的约束。

如何使用好GIL ？

既然无法避免 GIL，我学习了几种应对策略：

1. 多进程代替多线程

对于 CPU 密集型任务，使用 multiprocessing 模块：

import multiprocessing
import time

def cpu_task(n):
    count = 0
    for i in range(n):
        count += i
    return count

if __name__ == "__main__":
    start_time = time.time()
    
    # 创建两个进程
    p1 = multiprocessing.Process(target=cpu_task, args=(10**7,))
    p2 = multiprocessing.Process(target=cpu_task, args=(10**7,))
    
    p1.start()
    p2.start()
    p1.join()
    p2.join()
    
    print(f"多进程执行时间: {time.time() - start_time:.4f}秒")