1
# GIL: Global Interpreter Lock
2
# CPython 的互斥锁，保证同一时刻只有一个线程执行 Python 字节码
3

4
# 问题：Python 字节码执行时，解释器需要锁定
5
# 同一进程内的所有线程共享 GIL
6

7
import threading
8
import time
9

10
# CPU 密集型任务
11
def cpu_bound(n):
12
    return sum(i * i for i in range(n))
13

14
# 由于 GIL，多线程对 CPU 密集型任务无效
15
# 应该使用多进程绕过 GIL

1
# CPU 密集型：应该用多进程
2
import multiprocessing
3

4
def cpu_task(n):
5
    return sum(i * i for i in range(n))
6

7
if __name__ == "__main__":
8
    with multiprocessing.Pool(4) as pool:
9
        results = pool.map(cpu_task, [10**7] * 8)

1
# 1. C 扩展：长时间计算的 C 函数可释放 GIL
2
# 2. ctypes：调用 C 库时释放 GIL
3
# 3. PyPy：使用 JIT 编译器，减少 GIL 影响
4
# 4. 其他 Python 实现：Jython, IronPython 无 GIL
5

6
# 示例：使用 numpy（底层 C 实现）
7
import numpy as np
8
# numpy 操作不经过 Python 字节码，GIL 影响小
9
a = np.random.rand(10000, 10000)
10
b = np.dot(a, a.T)  # 并行计算

1
import asyncio
2

3
# async def 定义协程函数
4
async def fetch(url):
5
    await asyncio.sleep(1)  # 模拟 IO 操作
6
    return f"data from {url}"
7

8
# 运行协程
9
async def main():
10
    result = await fetch("example.com")
11
    print(result)
12

13
asyncio.run(main())
14

15
# 协程不能直接调用
16
# fetch("example.com")  # 不会执行！

1
import asyncio
2

3
async def fetch(url, delay):
4
    await asyncio.sleep(delay)
5
    return f"data from {url}"
6

7
async def main():
8
    # gather: 并发执行多个协程
9
    results = await asyncio.gather(
10
        fetch("url1", 1),
11
        fetch("url2", 2),
12
        fetch("url3", 0.5)
13
    )
14
    print(results)
15

16
asyncio.run(main())

1
import asyncio
2

3
async def main():
4
    # create_task: 创建后台任务
5
    task = asyncio.create_task(fetch("url1", 1))
6

7
    # 任务状态
8
    print(task.done())  # False
9
    print(task.result())  # 等待完成并获取结果
10

11
    # wait: 等待一组任务
12
    task1 = asyncio.create_task(fetch("url1", 1))
13
    task2 = asyncio.create_task(fetch("url2", 2))
14

15
    done, pending = await asyncio.wait([task1, task2])
16

17
    for task in done:
18
        print(task.result())
19

20
asyncio.run(main())

1
import asyncio
2

3
class AsyncResource:
4
    async def __aenter__(self):
5
        print("获取资源")
6
        await asyncio.sleep(0.1)
7
        return self
8

9
    async def __aexit__(self, exc_type, exc_val, exc_tb):
10
        print("释放资源")
11
        await asyncio.sleep(0.1)
12

13
async def main():
14
    async with AsyncResource() as resource:
15
        print("使用资源")
16

17
asyncio.run(main())

1
import asyncio
2

3
# 异步生成器
4
async def async_generator():
5
    for i in range(5):
6
        await asyncio.sleep(0.1)
7
        yield i
8

9
async def main():
10
    # 异步迭代
11
    async for item in async_generator():
12
        print(item)
13

14
    # 异步列表推导式
15
    results = [item async for item in async_generator()]
16
    print(results)
17

18
asyncio.run(main())

1
import threading
2
import time
3

4
def worker(name, seconds):
5
    print(f"线程 {name} 开始")
6
    time.sleep(seconds)
7
    print(f"线程 {name} 结束")
8

9
# 创建线程
10
t1 = threading.Thread(target=worker, args=("A", 1))
11
t2 = threading.Thread(target=worker, args=("B", 2))
12

13
t1.start()  # 启动线程
14
t2.start()
15

16
t1.join()   # 等待线程结束
17
t2.join()
18

19
print("所有线程结束")

1
import threading
2

3
# 互斥锁
4
counter = 0
5
lock = threading.Lock()
6

7
def increment():
8
    global counter
9
    with lock:  # 自动获取和释放锁
10
        counter += 1
11

12
threads = [threading.Thread(target=increment) for _ in range(1000)]
13
for t in threads: t.start()
14
for t in threads: t.join()
15
print(counter)  # 1000

1
import threading
2

3
# Queue 是线程安全的
4
from queue import Queue
5

6
q = Queue()
7
q.put("item")
8
item = q.get()  # 阻塞直到有元素
9

10
# ThreadLocal 线程本地存储
11
local_data = threading.local()
12

13
def worker():
14
    local_data.value = threading.current_thread().name
15
    print(local_data.value)
16

17
threads = [threading.Thread(target=worker) for _ in range(3)]
18
for t in threads: t.start()

1
import threading
2

3
class ReadWriteLock:
4
    def __init__(self):
5
        self._read_ready = threading.Condition(threading.Lock())
6
        self._readers = 0
7

8
    def acquire_read(self):
9
        with self._read_ready:
10
            self._readers += 1
11

12
    def release_read(self):
13
        with self._read_ready:
14
            self._readers -= 1
15
            if self._readers == 0:
16
                self._read_ready.notify_all()
17

18
    def acquire_write(self):
19
        self._read_ready.acquire()
20
        while self._readers > 0:
21
            self._read_ready.wait()
22

23
    def release_write(self):
24
        self._read_ready.release()
25

26
# RWLock 允许并发读，但写时独占

1
import multiprocessing
2

3
def worker(name):
4
    return f"Worker {name}"
5

6
if __name__ == "__main__":
7
    # 创建进程池
8
    with multiprocessing.Pool(4) as pool:
9
        results = pool.map(worker, ["A", "B", "C", "D"])
10
        print(results)  # ['Worker A', 'Worker B', 'Worker C', 'Worker D']
11

12
    # 单个进程
13
    p = multiprocessing.Process(target=worker, args=("X",))
14
    p.start()
15
    p.join()

1
import multiprocessing
2

3
# Queue - 进程间队列
4
def producer(queue):
5
    for i in range(5):
6
        queue.put(i)
7

8
def consumer(queue):
9
    while not queue.empty():
10
        print(queue.get())
11

12
if __name__ == "__main__":
13
    queue = multiprocessing.Queue()
14
    p1 = multiprocessing.Process(target=producer, args=(queue,))
15
    p2 = multiprocessing.Process(target=consumer, args=(queue,))
16

17
    p1.start()
18
    p1.join()
19
    p2.start()
20
    p2.join()
21

22
# Pipe - 双工管道
23
def proc1(conn):
24
    conn.send("hello")
25
    msg = conn.recv()
26
    print(msg)
27

28
if __name__ == "__main__":
29
    conn1, conn2 = multiprocessing.Pipe()
30
    p = multiprocessing.Process(target=proc1, args=(conn1,))
31
    p.start()
32
    print(conn2.recv())  # hello
33
    conn2.send("world")
34
    p.join()

1
import multiprocessing
2

3
# 共享值
4
value = multiprocessing.Value('i', 0)
5
lock = multiprocessing.Lock()
6

7
def increment():
8
    with lock:
9
        value.value += 1
10

11
# 共享数组
12
arr = multiprocessing.Array('i', [1, 2, 3])
13

14
# Manager - 共享数据结构
15
manager = multiprocessing.Manager()
16
shared_dict = manager.dict()
17
shared_list = manager.list()

1
# CPU 密集型：多进程
2
import multiprocessing
3

4
def compute(data):
5
    return sum(i * i for i in range(data))
6

7
with multiprocessing.Pool(4) as pool:
8
    results = pool.map(compute, [10**7] * 10)
9

10
# IO 密集型：asyncio（首选）或 threading
11
import asyncio
12

13
async def fetch_all(urls):
14
    async with aiohttp.ClientSession() as session:
15
        tasks = [fetch(session, url) for url in urls]
16
        return await asyncio.gather(*tasks)
17

18
# 混合型：多进程 + 协程
19
# 每个进程运行一个事件循环

1
# asyncio 版本
2
import asyncio
3

4
async def producer(queue):
5
    for i in range(10):
6
        await queue.put(i)
7
        await asyncio.sleep(0.1)
8

9
async def consumer(queue):
10
    while True:
11
        item = await queue.get()
12
        if item is None:
13
            break
14
        print(item)
15

16
async def main():
17
    queue = asyncio.Queue()
18
    await asyncio.gather(
19
        producer(queue),
20
        consumer(queue)
21
    )
22

23
asyncio.run(main())
24

25
# threading 版本
26
import threading
27
from queue import Queue
28

29
def producer(queue):
30
    for i in range(10):
31
        queue.put(i)
32

33
def consumer(queue):
34
    while True:
35
        item = queue.get()
36
        if item is None:
37
            break
38
        print(item)
39

40
queue = Queue()
41
threads = [
42
    threading.Thread(target=producer, args=(queue,)),
43
    threading.Thread(target=consumer, args=(queue,))
44
]
45
for t in threads: t.start()
46
for t in threads: t.join()