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

from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor, as_completed
import time

# CPU 密集型：ProcessPoolExecutor 自动管理进程池
def cpu_task(n):
    return sum(i ** 2 for i in range(n))

with ProcessPoolExecutor(max_workers=4) as executor:
    futures = [executor.submit(cpu_task, 1_000_000) for _ in range(8)]
    for future in as_completed(futures):
        print(f"结果: {future.result()}")

# I/O 密集型：ThreadPoolExecutor（GIL 会释放，线程够用）
def io_task(url):
    import requests
    return requests.get(url, timeout=10).status_code

with ThreadPoolExecutor(max_workers=20) as executor:
    urls = ["https://api.example.com/data"] * 10
    results = list(executor.map(io_task, urls))

# 混合模式：CPU 任务交给进程，I/O 任务用线程
def hybrid_workflow(data):
    # 第 1 步：I/O - 下载数据
    raw = download_data(data['url'])
    # 第 2 步：CPU - 处理数据
    processed = process_pool.submit(cpu_transform, raw).result()
    # 第 3 步：I/O - 上传结果
    return upload_result(processed)

# 带初始化的进程池
import numpy as np

_model = None  # 每个进程独立的模型

def init_worker():
    global _model
    _model = load_heavy_model()  # 每个进程只加载一次

def predict(data):
    return _model.predict(data)

with ProcessPoolExecutor(
    max_workers=4,
    initializer=init_worker,
    maxtasksperchild=100  # 每处理 100 个任务重启进程
) as executor:
    results = list(executor.map(predict, data_batches))

# calc.pyx
from cython.parallel import prange
import numpy as np
cimport numpy as np

def parallel_sum(np.ndarray[np.float64_t, ndim=1] arr):
    cdef:
        double total = 0
        Py_ssize_t i
        Py_ssize_t n = arr.shape[0]

    # nogil: 释放 GIL，允许多线程并行执行 C 代码
    with nogil:
        for i in prange(n, schedule='dynamic'):
            total += arr[i]

    return total

# setup.py
from setuptools import setup
from Cython.Build import cythonize
import numpy as np

setup(
    ext_modules=cythonize("calc.pyx", compiler_directives={'language_level': "3"}),
    include_dirs=[np.get_include()],
)

import time
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor

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

# 基准测试
jobs = 4

# 1. 单线程
start = time.perf_counter()
[cpu_intensive() for _ in range(jobs)]
single = time.perf_counter() - start

# 2. 多线程（CPU 密集，预期更慢或持平）
start = time.perf_counter()
with ThreadPoolExecutor(max_workers=jobs) as e:
    list(e.submit(cpu_intensive) for _ in range(jobs))
thread = time.perf_counter() - start

# 3. 多进程（CPU 密集，预期接近 1/N）
start = time.perf_counter()
with ProcessPoolExecutor(max_workers=jobs) as e:
    list(e.submit(cpu_intensive) for _ in range(jobs))
process = time.perf_counter() - start

print(f"单线程: {single:.2f}s")
print(f"多线程: {thread:.2f}s (x{thread/single:.1f})")
print(f"多进程: {process:.2f}s (x{process/single:.1f})")