12 并发编程（⚠️ 人工重点）

📋 资料说明

来源：课程 day11
笔记：尚硅谷大模型技术之Python1.0.docx 第 13 章
每日一考：day11_每日一考.md（含选择题 + 编程题）
视频：❌ 跳过（共 17 个视频：进程 → 线程 → GIL → 锁 → 卖票案例）

🟢 🤖 AI 替代率：72%

为什么是这个水平：

基础的多进程/多线程代码（Process/Thread 创建、Pool 使用、Queue 通信）AI 可以准确生成。

但并发编程涉及操作系统原理层面的知识，这是 AI 的薄弱环节：

GIL（全局解释器锁）：Python 的 GIL 导致 CPU 密集型任务在多线程下不会加速——这是 CPython 的实现限制，不是语法特征。AI 知道这个事实，但人需要理解"为什么 GIL 存在"以及"它对你代码的影响"。
线程安全：竞争条件、死锁、活锁——AI 可以写出带锁的代码，但锁的粒度和范围设计（什么是该保护的临界区）需要人类判断。
进程 vs 线程的选型：什么时候用多进程、什么时候用多线程、什么时候用协程——需要理解 I/O 密集型和 CPU 密集型的区别。

👤 人工干预率：28% | 人类重点关注

这是本模块人工干预率最高的章节。 以下是需要人类重点理解的三个方面：

GIL 的本质：GIL 让 Python 多线程对 CPU 密集型任务无效，但对 I/O 密集型任务仍然有效。AI 可以告诉你这个规则，但你需要理解"为什么文件下载用线程没问题，而数值计算用线程却更快不了"。
锁的设计：写 lock.acquire() / release() 很简单，但判断"哪些代码需要保护""锁应该多粗还是多细"需要设计能力。
进程间通信：进程间不共享内存，需要通过 Queue 或 Pipe 通信。这个限制让很多从单线程转来的人困惑。

人类与 AI 协作建议：

AI 生成的并发代码，你先用单线程跑通逻辑，再加并发
调试并发问题永远优先检查"共享数据的访问是否加锁了"
不确定用进程还是线程时，标准决策：I/O 密集型→线程，CPU 密集型→进程
把 AI 当作"写出框架"的工具，你来负责"加锁的位置和范围"

1. 进程（Process）

进程是操作系统分配资源的基本单位，每个进程有独立的内存空间。

通过 Process 创建进程

python

import os
from multiprocessing import Process

def worker(name):
    print(f"进程 {name} (PID: {os.getpid()}) 开始工作")
    for i in range(5):
        print(f"{name}: {i}")

# 创建进程
p1 = Process(target=worker, args=("A",))
p2 = Process(target=worker, args=("B",))

# 启动
p1.start()
p2.start()

# 等待结束（join 让主进程等待子进程完成）
p1.join()
p2.join()

print("所有进程执行完毕")

通过 Process 子类创建

python

class MyProcess(Process):
    def __init__(self, name):
        super().__init__()
        self.name = name
    
    def run(self):  # 重写 run 方法
        print(f"进程 {self.name} 运行中")

p = MyProcess("Worker")
p.start()
p.join()

进程池

python

from multiprocessing import Pool

def square(n):
    return n ** 2

with Pool(4) as pool:  # 4 个进程的池
    results = pool.map(square, range(10))
    print(results)  # [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

进程间通信（Queue）

python

from multiprocessing import Process, Queue

def producer(q):
    for i in range(5):
        q.put(f"数据 {i}")
    q.put(None)  # 结束信号

def consumer(q):
    while True:
        data = q.get()
        if data is None:
            break
        print(f"收到: {data}")

q = Queue()
p1 = Process(target=producer, args=(q,))
p2 = Process(target=consumer, args=(q,))

p1.start()
p2.start()
p1.join()
p2.join()

重要：进程之间不共享内存，每个进程都有独立的变量副本。必须通过 Queue、Pipe 等方式通信。

python

from multiprocessing import Process

number = 0

def add():
    global number
    for _ in range(100):
        number += 1

p1 = Process(target=add)
p2 = Process(target=add)
p1.start()
p2.start()
p1.join()
p2.join()
print(number)  # 可能是 0！因为每个进程有独立的 number 副本

2. 线程（Thread）

线程是 CPU 调度的基本单位，同一进程的线程共享内存空间。

通过 Thread 创建线程

python

import threading
import time

def worker(name, count):
    for i in range(count):
        print(f"线程 {name}: {i}")
        time.sleep(0.1)  # 模拟 I/O 操作

# 创建线程
t1 = threading.Thread(target=worker, args=("A", 5))
t2 = threading.Thread(target=worker, args=("B", 5))

# 启动
t1.start()
t2.start()

# 等待
t1.join()
t2.join()

线程池

python

from concurrent.futures import ThreadPoolExecutor

def fetch_url(url):
    # 模拟网络请求
    return f"{url} 的数据"

urls = ["http://example.com/a", "http://example.com/b", "http://example.com/c"]

with ThreadPoolExecutor(max_workers=3) as executor:
    results = executor.map(fetch_url, urls)
    for result in results:
        print(result)

3. GIL（全局解释器锁）⚠️

GIL 是 CPython 解释器的一个实现限制：同一时刻只有一个线程能执行 Python 字节码。

python

import threading
import time

def cpu_intensive(n):
    """CPU 密集型任务"""
    while n > 0:
        n -= 1

def io_intensive(n):
    """I/O 密集型任务"""
    for _ in range(n):
        time.sleep(0.001)  # 模拟 I/O 等待

# 实验：多线程能否加速？
count = 50000000

# 单线程 CPU 密集
start = time.time()
cpu_intensive(count)
print(f"单线程 CPU: {time.time() - start:.2f}s")

# 多线程 CPU 密集（因为有 GIL，不会加速！）
def worker():
    cpu_intensive(count // 2)

t1 = threading.Thread(target=worker)
t2 = threading.Thread(target=worker)
start = time.time()
t1.start()
t2.start()
t1.join()
t2.join()
print(f"双线程 CPU: {time.time() - start:.2f}s")  # 和单线程差不多

GIL 对开发者的影响：

场景	效果	推荐方案
CPU 密集型	❌ 多线程无加速	使用多进程 `multiprocessing`
I/O 密集型	✅ 多线程有效加速	使用多线程或 asyncio
混合型	⚠️ 部分加速	I/O 部分用线程，计算部分用进程

4. 线程安全问题

线程共享内存，可能导致数据竞争。

python

import threading

# 不安全版本
counter = 0

def increment_bad():
    global counter
    for _ in range(100000):
        counter += 1  # 不是原子操作！
        # 实际上：读取 → 计算 → 写入
        # 可能在这三个步骤中被其他线程打断

threads = [threading.Thread(target=increment_bad) for _ in range(10)]
for t in threads: t.start()
for t in threads: t.join()
print(f"不安全计数器: {counter}")  # 大概率不是 1000000！

互斥锁解决线程安全

python

import threading

lock = threading.Lock()
counter = 0

def increment_safe():
    global counter
    for _ in range(100000):
        with lock:  # 加锁（相当于 lock.acquire() 和 lock.release()）
            counter += 1  # 这段代码同时只能有一个线程执行

threads = [threading.Thread(target=increment_safe) for _ in range(10)]
for t in threads: t.start()
for t in threads: t.join()
print(f"安全计数器: {counter}")  # 1000000

卖票案例

python

import threading

total_tickets = 10
lock = threading.Lock()

def sell_ticket(window):
    global total_tickets
    while True:
        with lock:
            if total_tickets <= 0:
                print(f"窗口 {window}: 已售罄")
                break
            total_tickets -= 1
            print(f"窗口 {window}: 售出 1 张，剩余 {total_tickets}")

windows = [threading.Thread(target=sell_ticket, args=(i,)) for i in range(3)]
for w in windows: w.start()
for w in windows: w.join()

5. 进程 vs 线程对比

对比项	进程 (Process)	线程 (Thread)
内存空间	独立	共享
通信方式	Queue/Pipe	直接访问变量（需加锁）
创建开销	大	小
适合场景	CPU 密集型	I/O 密集型
GIL 影响	不受影响（每个进程有独立 GIL）	受影响
数据安全性	天然安全（不共享）	需要锁保护

练习题

练习 1：多线程打印

使用 threading 创建两个线程，一个打印 "Hello" 10 次，另一个打印 "World" 10 次。

练习 2：多进程求和

使用 multiprocessing 创建两个进程，分别计算 1~50 和 51~100 的和，然后汇总。

练习 3：线程安全计数器（选择题）

以下选项中，哪些是解决线程安全问题的正确方法？ A. 使用 Lock 保护临界区 B. 使用 Queue 通信 C. 使用全局变量 D. 使用 thread_local 存储

参考答案

练习 1：

python

import threading

def print_hello():
    for _ in range(10):
        print("Hello")

def print_world():
    for _ in range(10):
        print("World")

t1 = threading.Thread(target=print_hello)
t2 = threading.Thread(target=print_world)
t1.start()
t2.start()
t1.join()
t2.join()

练习 2：

python

from multiprocessing import Process, Queue

def sum_range(start, end, q):
    q.put(sum(range(start, end + 1)))

q = Queue()
p1 = Process(target=sum_range, args=(1, 50, q))
p2 = Process(target=sum_range, args=(51, 100, q))
p1.start()
p2.start()
p1.join()
p2.join()

total = q.get() + q.get()
print(f"1~100 的和: {total}")  # 5050

练习 3：正确答案 A、B、D（C 的全局变量不加锁是竞态条件）

12 并发编程（⚠️ 人工重点） ​

📋 资料说明 ​

🟢 🤖 AI 替代率：72% ​

👤 人工干预率：28% | 人类重点关注 ​

1. 进程（Process） ​

通过 Process 创建进程 ​

通过 Process 子类创建 ​

进程池 ​

进程间通信（Queue） ​

2. 线程（Thread） ​

通过 Thread 创建线程 ​

线程池 ​

3. GIL（全局解释器锁）⚠️ ​

4. 线程安全问题 ​

互斥锁解决线程安全 ​

卖票案例 ​

5. 进程 vs 线程 对比 ​

练习题 ​

练习 1：多线程打印 ​

练习 2：多进程求和 ​

练习 3：线程安全计数器（选择题） ​

12 并发编程（⚠️ 人工重点）

📋 资料说明

🟢 🤖 AI 替代率：72%

👤 人工干预率：28% | 人类重点关注

1. 进程（Process）

通过 Process 创建进程

通过 Process 子类创建

进程池

进程间通信（Queue）

2. 线程（Thread）

通过 Thread 创建线程

线程池

3. GIL（全局解释器锁）⚠️

4. 线程安全问题

互斥锁解决线程安全

卖票案例

5. 进程 vs 线程对比

练习题

练习 1：多线程打印

练习 2：多进程求和

练习 3：线程安全计数器（选择题）