银河it-Python 并发与并行：从 GIL 到自由线程的工程真相

import threading
import time

def cpu_bound(n: int) -> int:
    total = 0
    for i in range(n):
        total += i * i
    return total

def run_single():
    start = time.perf_counter()
    cpu_bound(20_000_000)
    print(f"single: {time.perf_counter() - start:.3f}s")

def run_threads():
    threads = [
        threading.Thread(target=cpu_bound, args=(20_000_000,))
        for _ in range(4)
    ]
    start = time.perf_counter()
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    print(f"threads: {time.perf_counter() - start:.3f}s")

if __name__ == "__main__":
    run_single()
    run_threads()

import threading
import time
import urllib.request

URLS = [
    "https://example.com",
    "https://httpbin.org/get",
    "https://www.python.org",
] * 5

def fetch(url: str) -> bytes:
    with urllib.request.urlopen(url, timeout=10) as resp:
        return resp.read()

def run_sequential():
    start = time.perf_counter()
    for url in URLS:
        fetch(url)
    print(f"sequential: {time.perf_counter() - start:.3f}s")

def run_threaded():
    threads = [threading.Thread(target=fetch, args=(url,)) for url in URLS]
    start = time.perf_counter()
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    print(f"threaded: {time.perf_counter() - start:.3f}s")

import threading

counter = 0

def worker():
    global counter
    for _ in range(100_000):
        counter += 1

threads = [threading.Thread(target=worker) for _ in range(4)]
for t in threads:
    t.start()
for t in threads:
    t.join()

print(counter)  # 通常小于 400000

import threading

counter = 0
lock = threading.Lock()

def worker():
    global counter
    for _ in range(100_000):
        with lock:
            counter += 1

from multiprocessing import Pool
import time

def cpu_bound(n: int) -> int:
    total = 0
    for i in range(n):
        total += i * i
    return total

if __name__ == "__main__":
    start = time.perf_counter()
    with Pool(4) as pool:
        results = pool.map(cpu_bound, [20_000_000] * 4)
    print(f"multiprocessing: {time.perf_counter() - start:.3f}s")

from multiprocessing import shared_memory
import numpy as np

# 创建共享内存块
shm = shared_memory.SharedMemory(create=True, size=8 * 1024 * 1024)
arr = np.ndarray((1024, 1024), dtype=np.float64, buffer=shm.buf)

# 子进程可以通过名称附加到同一块内存
# shm_child = shared_memory.SharedMemory(name=shm.name)
# arr_child = np.ndarray((1024, 1024), dtype=np.float64, buffer=shm_child.buf)

import asyncio
import aiohttp

async def fetch(session: aiohttp.ClientSession, url: str) -> str:
    async with session.get(url) as resp:
        return await resp.text()

async def main():
    urls = ["https://example.com"] * 20
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
    return results

# asyncio.run(main())

import asyncio
import time

async def bad_task():
    time.sleep(5)  # 阻塞整个事件循环
    return "done"

async def main():
    await asyncio.gather(bad_task(), bad_task(), bad_task())

import asyncio
import time
from concurrent.futures import ThreadPoolExecutor

def blocking_io():
    time.sleep(5)
    return "done"

async def main():
    loop = asyncio.get_running_loop()
    with ThreadPoolExecutor() as pool:
        result = await loop.run_in_executor(pool, blocking_io)
    return result

result = await asyncio.to_thread(blocking_io)

import sys

print(sys._is_gil_enabled())  # Python 3.13+ 自由线程构建返回 False

from concurrent import interpreters

interp = interpreters.create()
interp.run("print('hello from subinterpreter')")

import time
import threading

def diagnose(func, *args, n=4):
    # 单线程
    start = time.perf_counter()
    func(*args)
    single = time.perf_counter() - start

    # 多线程
    threads = [threading.Thread(target=func, args=args) for _ in range(n)]
    start = time.perf_counter()
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    threaded = time.perf_counter() - start

    print(f"single: {single:.3f}s")
    print(f"threaded x{n}: {threaded:.3f}s")
    print(f"speedup: {single * n / threaded:.2f}x")