1
import threading
2
import time
3

4
def cpu_bound_task():
5
    """CPU 密集型任务"""
6
    count = 0
7
    for _ in range(10_000_000):
8
        count += 1
9

10
# 单线程
11
start = time.time()
12
cpu_bound_task()
13
cpu_bound_task()
14
print(f"单线程: {time.time() - start:.2f}s")
15

16
# 多线程
17
start = time.time()
18
t1 = threading.Thread(target=cpu_bound_task)
19
t2 = threading.Thread(target=cpu_bound_task)
20
t1.start(); t2.start()
21
t1.join(); t2.join()
22
print(f"多线程: {time.time() - start:.2f}s")

1
单线程: 0.45s
2
多线程: 0.52s  # 多线程反而更慢！

1
// CPython 内部的引用计数结构
2
typedef struct PyObject {
3
    Py_ssize_t ob_refcnt;      // 引用计数
4
    PyTypeObject *ob_type;     // 类型指针
5
} PyObject;
6

7
// 增加引用计数
8
#define Py_INCREF(op) ((void)(++(op)->ob_refcnt))
9

10
// 减少引用计数
11
#define Py_DECREF(op) \
12
    do { \
13
        if (--((op)->ob_refcnt) == 0) \
14
            _Py_Dealloc((PyObject *)(op)); \
15
    } while (0)

1
// 理论上的细粒度锁方案
2
typedef struct PyObject {
3
    Py_ssize_t ob_refcnt;
4
    pthread_mutex_t refcnt_lock;  // 每个对象一个锁
5
    PyTypeObject *ob_type;
6
} PyObject;
7

8
#define Py_INCREF(op) do { \
9
    pthread_mutex_lock(&(op)->refcnt_lock); \
10
    ++(op)->ob_refcnt; \
11
    pthread_mutex_unlock(&(op)->refcnt_lock); \
12
} while(0)

1
import threading
2
import time
3

4
def fibonacci(n):
5
    """CPU 密集型：计算斐波那契数"""
6
    if n <= 1:
7
        return n
8
    return fibonacci(n - 1) + fibonacci(n - 2)
9

10
def benchmark(threads):
11
    start = time.time()
12
    workers = []
13
    for _ in range(threads):
14
        t = threading.Thread(target=lambda: fibonacci(30))
15
        workers.append(t)
16
        t.start()
17
    for t in workers:
18
        t.join()
19
    return time.time() - start
20

21
print(f"1 线程: {benchmark(1):.2f}s")
22
print(f"2 线程: {benchmark(2):.2f}s")
23
print(f"4 线程: {benchmark(4):.2f}s")

1
1 线程: 0.52s
2
2 线程: 0.55s  # 几乎没有改善
3
4 线程: 0.58s  # 甚至更慢！

1
import threading
2
import urllib.request
3
import time
4

5
def download(url):
6
    """I/O 密集型：网络请求"""
7
    try:
8
        urllib.request.urlopen(url)
9
    except:
10
        pass
11

12
urls = ["http://example.com"] * 10
13

14
# 单线程
15
start = time.time()
16
for url in urls:
17
    download(url)
18
print(f"单线程: {time.time() - start:.2f}s")
19

20
# 多线程
21
start = time.time()
22
threads = [threading.Thread(target=download, args=(url,)) for url in urls]
23
for t in threads: t.start()
24
for t in threads: t.join()
25
print(f"多线程: {time.time() - start:.2f}s")

1
单线程: 2.34s
2
多线程: 0.45s  # 显著提升！

1
// CPython 的 GIL 切换逻辑（简化）
2
#define CHECK_INTERVAL 100  // 每 100 个字节码指令检查一次
3

4
static int _Py_CheckInterval = CHECK_INTERVAL;
5

6
void PyEval_AcquireThread(PyThreadState *tstate) {
7
    // 尝试获取 GIL
8
    while (!gil_locked) {
9
        if (_Py_CheckInterval-- <= 0) {
10
            // 让出 CPU，给其他线程机会
11
            _Py_CheckInterval = CHECK_INTERVAL;
12
            // ... 切换逻辑
13
        }
14
    }
15
}

1
# 大量 C 扩展依赖 GIL 的存在
2

3
# NumPy 中的模式
4
import numpy as np
5

6
def compute():
7
    # 创建数组时持有 GIL
8
    arr = np.zeros(1000000)
9

10
    # 释放 GIL 进行计算
11
    # ... C 层面的并行计算
12

13
    # 重新获取 GIL 返回结果
14
    return arr.sum()

1
// CPython 源码中的 GIL 结构（简化）
2
struct _gil {
3
    pthread_mutex_t mutex;      // 互斥锁
4
    pthread_cond_t cond;        // 条件变量
5
    int locked;                 // 锁状态
6
    unsigned long switch_number; // 切换计数
7
};

1
// 获取 GIL
2
void PyEval_AcquireThread(PyThreadState *tstate) {
3
    pthread_mutex_lock(&gil.mutex);
4

5
    while (gil.locked) {
6
        // GIL 被其他线程持有，等待
7
        pthread_cond_wait(&gil.cond, &gil.mutex);
8
    }
9

10
    gil.locked = 1;
11
    pthread_mutex_unlock(&gil.mutex);
12
}
13

14
// 释放 GIL
15
void PyEval_ReleaseThread(PyThreadState *tstate) {
16
    pthread_mutex_lock(&gil.mutex);
17

18
    gil.locked = 0;
19
    // 唤醒等待的线程
20
    pthread_cond_signal(&gil.cond);
21

22
    pthread_mutex_unlock(&gil.mutex);
23
}

1
// 在 C 扩展中释放 GIL
2
static PyObject *
3
my_blocking_operation(PyObject *self, PyObject *args) {
4
    // 执行 Python 代码，持有 GIL
5
    PyObject *result;
6

7
    // 释放 GIL，允许其他线程执行
8
    Py_BEGIN_ALLOW_THREADS
9

10
    // 这里执行阻塞操作（I/O、长时间计算等）
11
    // 其他线程可以同时执行 Python 代码
12
    result = blocking_io_call();
13

14
    // 重新获取 GIL
15
    Py_END_ALLOW_THREADS
16

17
    // 继续执行 Python 代码
18
    return result;
19
}

1
// Python 3.2+ 的切换逻辑
2
#define _Py_EVAL_DEFAULT_INTERVAL 5000  // 5ms
3

4
static inline int
5
eval_frame_handle_pending(PyThreadState *tstate) {
6
    _PyRuntimeState *runtime = &_PyRuntime;
7

8
    // 检查是否需要切换
9
    if (runtime->ceval.gil.drop_request) {
10
        // 释放 GIL
11
        drop_gil(runtime, tstate);
12

13
        // 等待重新获取
14
        take_gil(runtime, tstate);
15
    }
16

17
    return 0;
18
}

1
from multiprocessing import Pool, cpu_count
2
import time
3

4
def cpu_bound(n):
5
    """CPU 密集型任务"""
6
    return sum(i * i for i in range(n))
7

8
if __name__ == "__main__":
9
    data = [10_000_000] * cpu_count()
10

11
    # 多进程（真正并行）
12
    start = time.time()
13
    with Pool() as pool:
14
        results = pool.map(cpu_bound, data)
15
    print(f"多进程: {time.time() - start:.2f}s")
16

17
    # 对比多线程
18
    import threading
19
    start = time.time()
20
    threads = []
21
    for _ in range(cpu_count()):
22
        t = threading.Thread(target=cpu_bound, args=(10_000_000,))
23
        threads.append(t)
24
        t.start()
25
    for t in threads:
26
        t.join()
27
    print(f"多线程: {time.time() - start:.2f}s")

1
多进程: 0.85s
2
多线程: 2.40s

1
import numpy as np
2
import threading
3

4
def matrix_multiply():
5
    # NumPy 的矩阵运算在 C 层执行
6
    # 会释放 GIL，真正并行
7
    a = np.random.rand(1000, 1000)
8
    b = np.random.rand(1000, 1000)
9
    return a @ b
10

11
# 多线程可以并行执行
12
threads = [threading.Thread(target=matrix_multiply) for _ in range(4)]
13
for t in threads: t.start()
14
for t in threads: t.join()
15
# 这个例子中，NumPy 会并行执行！

1
import asyncio
2
import aiohttp
3
import time
4

5
async def fetch(session, url):
6
    async with session.get(url) as response:
7
        return await response.text()
8

9
async def main():
10
    urls = ["https://example.com"] * 10
11

12
    async with aiohttp.ClientSession() as session:
13
        tasks = [fetch(session, url) for url in urls]
14
        results = await asyncio.gather(*tasks)
15
        print(f"Fetched {len(results)} pages")
16

17
start = time.time()
18
asyncio.run(main())
19
print(f"异步耗时: {time.time() - start:.2f}s")

1
# PyPy 使用 JIT 编译器优化性能
2
# 虽然 GIL 仍在，但代码执行更快
3

4
def fibonacci(n):
5
    if n <= 1:
6
        return n
7
    return fibonacci(n - 1) + fibonacci(n - 2)
8

9
# CPython: ~2.5s
10
# PyPy: ~0.3s (JIT 热身后)

1
# Jython (运行在 JVM 上)
2
# 可以真正使用 Java 的多线程
3

4
from java.lang import Thread
5

6
class MyThread(Thread):
7
    def run(self):
8
        # 没有 GIL 限制
9
        compute_heavy_task()
10

11
# 创建多个线程并行执行
12
threads = [MyThread() for _ in range(4)]
13
for t in threads:
14
    t.start()
15
for t in threads:
16
    t.join()

1
// 无 GIL 模式下的引用计数（原子操作）
2
#ifdef Py_GIL_DISABLED
3
    #define Py_INCREF(op) Atomic_Increment(&(op)->ob_refcnt)
4
    #define Py_DECREF(op) do { \
5
        if (Atomic_Decrement(&(op)->ob_refcnt) == 0) \
6
            _Py_Dealloc((PyObject *)(op)); \
7
    } while (0)
8
#else
9
    // 原有的 GIL 保护版本
10
    #define Py_INCREF(op) ((void)(++(op)->ob_refcnt))
11
    #define Py_DECREF(op) ...
12
#endif

1
import _interpreters
2

3
# 创建子解释器（每个有自己的 GIL）
4
interp_id = _interpreters.create()
5

6
# 在子解释器中执行代码
7
_interpreters.run_string(interp_id, """
8
import threading
9
print("Running in subinterpreter")
10
""")

1
# 编译无 GIL 版本的 Python
2
./configure --disable-gil
3
make
4
make install
5

6
# 运行时检查
7
python -c "import sys; print(sys.flags.gil_disabled)"
8
# True: 无 GIL 模式
9
# False: 有 GIL 模式

1
// Java: 真正的多线程并行
2
public class MultiThreadCompute {
3
    public static void main(String[] args) throws Exception {
4
        int cores = Runtime.getRuntime().availableProcessors();
5
        Thread[] threads = new Thread[cores];
6

7
        for (int i = 0; i < cores; i++) {
8
            threads[i] = new Thread(() -> {
9
                // CPU 密集型计算
10
                long sum = 0;
11
                for (long j = 0; j < 100_000_000L; j++) {
12
                    sum += j;
13
                }
14
            });
15
            threads[i].start();
16
        }
17

18
        for (Thread t : threads) {
19
            t.join();
20
        }
21
        // 在 4 核 CPU 上，性能提升约 4 倍
22
    }
23
}

1
// Go: 轻量级并发
2
package main
3

4
import (
5
    "fmt"
6
    "runtime"
7
    "sync"
8
)
9

10
func cpuBound(wg *sync.WaitGroup) {
11
    defer wg.Done()
12
    sum := 0
13
    for i := 0; i < 100_000_000; i++ {
14
        sum += i
15
    }
16
}
17

18
func main() {
19
    cores := runtime.NumCPU()
20
    var wg sync.WaitGroup
21

22
    for i := 0; i < cores; i++ {
23
        wg.Add(1)
24
        go cpuBound(&wg)  // 轻量级 goroutine
25
    }
26

27
    wg.Wait()
28
    fmt.Println("Done")
29
    // 在 4 核 CPU 上，性能提升约 4 倍
30
}