Python GIL 深度解析：全局解释器锁对并发编程的影响与绕过策略

import sys

# Python 3.13+ 可以查询 GIL 状态
try:
    print(f"GIL 状态: {sys._is_gil_enabled()}")
except AttributeError:
    print("当前 Python 版本不支持查询 GIL 状态")

import threading
import time

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

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

# 多线程（预期：更慢！）
threads = [threading.Thread(target=cpu_task, args=(5_000_000,)) for _ in range(2)]
start = time.perf_counter()
for t in threads:
    t.start()
for t in threads:
    t.join()
print(f"多线程: {time.perf_counter() - start:.2f}s")

import threading
import time

def io_task(n):
    time.sleep(1)   # 模拟 I/O 等待
    return n

# 多线程 I/O 操作
threads = [threading.Thread(target=io_task, args=(i,)) for i in range(4)]
start = time.perf_counter()
for t in threads:
    t.start()
for t in threads:
    t.join()
print(f"多线程 I/O: {time.perf_counter() - start:.2f}s")
# → 约 1 秒（线程在 sleep 时会释放 GIL）

import numpy as np
from threading import Thread

# NumPy 操作在 C 层释放 GIL
arr = np.random.rand(10_000_000)

def numpy_task():
    return np.fft.fft(arr)

threads = [Thread(target=numpy_task) for _ in range(4)]
# 这些线程能真正并行！

from multiprocessing import Pool
import time

def cpu_intensive(n):
    return sum(i ** 2 for i in range(n))

if __name__ == "__main__":
    inputs = [5_000_000] * 4
    
    with Pool(processes=4) as pool:
        start = time.perf_counter()
        results = pool.map(cpu_intensive, inputs)
        print(f"多进程: {time.perf_counter() - start:.2f}s")

import asyncio

async def fetch_data(i):
    await asyncio.sleep(1)
    return f"Data {i}"

async def main():
    tasks = [fetch_data(i) for i in range(10)]
    results = await asyncio.gather(*tasks)
    return results

asyncio.run(main())

# 需要从源码编译，启用 --disable-gil
./configure --disable-gil
make