Python GIL 全局解释器锁

GIL（Global Interpreter Lock）是 Python 解释器的全局锁，限制多线程并行执行。

GIL 基本概念

# GIL 确保：
# 1. 同一时刻只有一个线程执行 Python 字节码
# 2. 保护 Python 对象内存管理

# 这意味着：
# 多线程无法真正并行执行 CPU 密集型任务

GIL 对性能的影响

import threading
import time

def cpu_task():
    total = 0
    for _ in range(10000000):
        total += 1

# 单线程
start = time.time()
cpu_task()
cpu_task()
print(f"单线程: {time.time() - start:.2f}s")

# 多线程（受 GIL 影响）
start = time.time()
t1 = threading.Thread(target=cpu_task)
t2 = threading.Thread(target=cpu_task)
t1.start()
t2.start()
t1.join()
t2.join()
print(f"多线程: {time.time() - start:.2f}s")

# 多线程可能更慢（线程切换开销）

I/O 操作释放 GIL

import threading
import time

def io_task():
    time.sleep(1)  # I/O 操作释放 GIL

# 多线程 I/O 密集型任务有效
start = time.time()
threads = [threading.Thread(target=io_task) for _ in range(10)]
for t in threads:
    t.start()
for t in threads:
    t.join()
print(f"耗时: {time.time() - start:.2f}s")  # 约 1 秒（并行）

GIL 释放时机

规避 GIL 策略

多进程替代多线程

from multiprocessing import Process
import time

def cpu_task():
    total = 0
    for _ in range(10000000):
        total += 1

# 多进程绕过 GIL
start = time.time()
processes = [Process(target=cpu_task) for _ in range(4)]
for p in processes:
    p.start()
for p in processes:
    p.join()
print(f"多进程: {time.time() - start:.2f}s")  # 真正并行

使用 ProcessPoolExecutor

from concurrent.futures import ProcessPoolExecutor

def cpu_task(n):
    total = 0
    for _ in range(n):
        total += 1
    return total

with ProcessPoolExecutor(max_workers=4) as executor:
    results = executor.map(cpu_task, [10000000] * 4)
    print(list(results))

C 扩展释放 GIL

# C 扩展可以手动释放 GIL
# 示例：NumPy 的计算操作

import numpy as np

# NumPy 大量计算时释放 GIL
arr = np.random.rand(1000000)
result = np.dot(arr, arr)  # C 层执行，释放 GIL

使用协程

import asyncio

async def async_task(n):
    await asyncio.sleep(n)
    return n

async def main():
    tasks = [async_task(1) for _ in range(10)]
    results = await asyncio.gather(*tasks)
    print(results)

asyncio.run(main())  # 单线程并发，不受 GIL 影响

GIL 切换机制

# Python 3.2+ 使用基于时间的切换
# 每 5ms（默认）检查是否需要切换线程

import sys
print(sys.getswitchinterval())  # 默认 0.005（5毫秒）

# 设置切换间隔
sys.setswitchinterval(0.01)  # 10毫秒

GIL 存在原因

任务类型选择

性能测试对比

import threading
import multiprocessing
import time

def task():
    sum(range(10000000))

# 测试函数
def benchmark(func, count):
    start = time.time()
    workers = [func(target=task) for _ in range(count)]
    for w in workers:
        w.start()
    for w in workers:
        w.join()
    return time.time() - start

print(f"多线程: {benchmark(threading.Thread, 4):.2f}s")
print(f"多进程: {benchmark(multiprocessing.Process, 4):.2f}s")

要点总结

• GIL 限制同一时刻只有一个线程执行字节码
• CPU 密集型任务受 GIL 影响无法真正并行
• I/O 操作自动释放 GIL，多线程有效
• 多进程绕过 GIL，适合 CPU 密集型
• C 扩展可手动释放 GIL（如 NumPy）
• 协程单线程并发，不受 GIL 限制
• 根据任务类型选择合适的并发方案