【一起学 Python】第 68 天:全局解释器锁 (GIL) 的现状与 3.14 的无 GIL 模式

import threadingimport timecounter = 0def increment():    global counter    for _ in range(1000000):        counter += 1# 即使使用多线程，由于 GIL 的存在，不会真正并行threads = [threading.Thread(target=increment) for _ in range(4)]for t in threads:    t.start()for t in threads:    t.join()print(counter)  # 结果不确定，因为 counter += 1 不是原子操作

import threadingimport timedef cpu_intensive():    total = 0    for i in range(10000000):        total += i    return total# 单线程start = time.time()cpu_intensive()cpu_intensive()print(f"单线程：{time.time() - start:.2f}秒")# 多线程（由于 GIL，不会更快）start = time.time()t1 = threading.Thread(target=cpu_intensive)t2 = threading.Thread(target=cpu_intensive)t1.start()t2.start()t1.join()t2.join()print(f"多线程：{time.time() - start:.2f}秒")  # 时间相近甚至更慢

import threadingimport timedef io_intensive():    time.sleep(1)  # 模拟 I/O 等待，此时会释放 GIL# 多线程处理 I/O 是有效的start = time.time()threads = [threading.Thread(target=io_intensive) for _ in range(4)]for t in threads:    t.start()for t in threads:    t.join()print(f"4 个 I/O 任务：{time.time() - start:.2f}秒")  # 约 1 秒，而非 4 秒

from multiprocessing import Poolimport timedef cpu_intensive(n):    total = 0    for i in range(n):        total += i    return totalif __name__ == "__main__":    # 多进程可以真正并行    start = time.time()    with Pool(4) as p:        results = p.map(cpu_intensive, [10000000] * 4)    print(f"多进程：{time.time() - start:.2f}秒")

import numpy as npimport threading# NumPy 操作可以并行def numpy_operation():    a = np.random.rand(1000, 1000)    return np.dot(a, a)# NumPy 在计算时释放 GIL，可以真正并行threads = [threading.Thread(target=numpy_operation) for _ inrange(4)]for t in threads:    t.start()for t in threads:    t.join()

import asyncioasync def fetch_data(url):    await asyncio.sleep(1)  # 模拟网络请求    return f"Data from {url}"async def main():    urls = ["url1", "url2", "url3", "url4"]    tasks = [fetch_data(url) for url in urls]    results = await asyncio.gather(*tasks)    print(results)asyncio.run(main())  # 4 个任务约 1 秒完成

import threadingclass ThreadSafeCounter:    """线程安全计数器"""    def __init__(self):        self._value = 0        self._lock = threading.Lock()    def increment(self):        with self._lock:            self._value += 1    def get_value(self):        with self._lock:            return self._value# 使用counter = ThreadSafeCounter()threads = []for _ in range(10):    t = threading.Thread(target=lambda: [counter.increment() for _ in range(1000)])    threads.append(t)    t.start()for t in threads:    t.join()print(f"最终计数：{counter.get_value()}")  # 10000（准确）

import threadingfrom typing import Any, Optionalclass ThreadSafeCache:    """线程安全缓存"""    def __init__(self):        self._cache: dict[str, Any] = {}        self._lock = threading.Lock()    def get(self, key: str) -> Optional[Any]:        with self._lock:            return self._cache.get(key)    def set(self, key: str, value: Any) -> None:        with self._lock:            self._cache[key] = value    def delete(self, key: str) -> bool:        with self._lock:            if key in self._cache:                del self._cache[key]                return True            return False# 使用cache = ThreadSafeCache()cache.set("user:1", {"name": "张三"})print(cache.get("user:1"))

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutorimport timedef cpu_task(n):    return sum(range(n))def io_task(n):    time.sleep(1)    return n# CPU 密集型 -> 多进程with ProcessPoolExecutor(max_workers=4) as executor:    results = list(executor.map(cpu_task, [10000000] * 4))# I/O 密集型 -> 多线程with ThreadPoolExecutor(max_workers=4) as executor:    results = list(executor.map(io_task, range(4)))