Python GIL全局解释器锁详解：原理影响与突破方案 - 小确幸生活

import sys
import threading

# 查看GIL切换间隔（Python 3中默认约5ms）
interval = sys.getswitchinterval()
print(f"GIL切换间隔: {interval} 秒")
# Python 3.2+ 默认约 0.005 秒（5毫秒）

# 可以通过 sys.setswitchinterval() 调整
sys.setswitchinterval(0.001)  # 设置为1毫秒

import threading
import time

def cpu_bound_task(n):
    """CPU密集型任务：计算大量平方"""
    result = 0
    for i in range(n):
        result += i * i
    return result

# 单线程执行
start = time.time()
cpu_bound_task(10000000)
cpu_bound_task(10000000)
print(f"单线程耗时: {time.time() - start:.2f}秒")

# 多线程执行
start = time.time()
t1 = threading.Thread(target=cpu_bound_task, args=(10000000,))
t2 = threading.Thread(target=cpu_bound_task, args=(10000000,))
t1.start()
t2.start()
t1.join()
t2.join()
print(f"多线程耗时: {time.time() - start:.2f}秒")
# 多线程可能比单线程更慢！因为GIL切换开销

import threading
import time

def io_bound_task():
    """IO密集型任务：模拟网络请求"""
    time.sleep(1)  # 模拟IO等待
    return "done"

# 单线程执行
start = time.time()
for _ in range(5):
    io_bound_task()
print(f"单线程耗时: {time.time() - start:.2f}秒")  # 约5秒

# 多线程执行
start = time.time()
threads = [threading.Thread(target=io_bound_task) for _ in range(5)]
for t in threads:
    t.start()
for t in threads:
    t.join()
print(f"多线程耗时: {time.time() - start:.2f}秒")  # 约1秒
# 多线程显著提升性能！

from multiprocessing import Pool
import time

def cpu_bound_task(n):
    result = 0
    for i in range(n):
        result += i * i
    return result

if __name__ == '__main__':
    start = time.time()
    
    # 使用多进程并行计算
    with Pool(processes=4) as pool:
        results = pool.map(cpu_bound_task, [10000000, 10000000, 10000000, 10000000])
    
    print(f"多进程耗时: {time.time() - start:.2f}秒")
    # 多进程可以绕过GIL，实现真正的并行

# 使用NumPy等C扩展库
import numpy as np

# NumPy的底层是C实现，会释放GIL
array1 = np.random.rand(10000, 10000)
array2 = np.random.rand(10000, 10000)

# 矩阵运算在C层面执行，不受GIL限制
result = np.dot(array1, array2)
# 可以充分利用多核CPU

import asyncio
import aiohttp

async def fetch(url):
    """异步网络请求"""
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

async def main():
    urls = ["http://example.com"] * 10
    
    # 并发执行10个网络请求
    tasks = [fetch(url) for url in urls]
    results = await asyncio.gather(*tasks)
    
    print(f"完成 {len(results)} 个请求")

asyncio.run(main())
# 异步IO在等待时释放GIL，适合IO密集型任务