Python教程 Episode 19 - 并发与多线程深度实践

欢迎来到"Python教程从零基础到实战"系列的第十九期！

上一期我们聊了 CI/CD 和 DevOps，让你的代码能够自动部署到云端。但不管你用什么部署工具，有一个底层问题绕不过去：**你的程序能不能充分利用多核 CPU？**

今天我们要深入 Python 并发编程的世界——多线程、多进程、协程，以及它们之间的微妙关系。这不仅仅是为了"更快"，更是为了让你的程序在 IO 密集型和 CPU 密集型场景下都能做出正确的选择。

准备好了吗？让我们揭开 GIL 的神秘面纱，掌握真正的性能优化技巧。

---

## 一、为什么需要并发编程？

### 1.1 单线程 vs 多线程的日常对比

想象一下你在一家餐厅打工：

-**单线程**：你一个人站柜台，顾客 A 点餐 → 你去厨房下单 → 等厨师做好 → 端给 A → 再接待 B……一个接一个，排成长队。

-**多线程**：你点餐的同时，让同事 B 去厨房盯着厨师，C 负责上菜。三个人并行工作，效率大幅提升。

但在 Python 的世界里，事情没有这么简单。Python 有一个著名的特性叫 **GIL（全局解释器锁）**，它让很多初学者踩坑。

### 1.2 什么是 GIL？

GIL（Global Interpreter Lock）是 CPython 解释器的一个设计——**在任何时刻，只有一个线程在执行 Python 字节码**。这意味着：

- ✅ 多线程对 IO 密集型任务有用（网络请求、文件读写——等待时释放 GIL）

- ❌ 多线程对 CPU 密集型任务**几乎没有加速效果**（计算时不释放 GIL）

这不是 Python 的缺陷，而是 CPython 的实现选择。其他语言实现（如 Jython、IronPython）没有 GIL。

---

## 二、threading 模块：多线程入门

### 2.1 基本用法：启动多个线程

```python

import threading

import time

defworker(name: str, seconds: int) -> None:

"""模拟一个耗时工作"""

print(f"[{name}] 开始工作...")

    time.sleep(seconds)

print(f"[{name}] 工作完成！")

# ===== 方式一：直接创建并启动线程 =====

t1 = threading.Thread(target=worker, args=("线程-A", 2))

t2 = threading.Thread(target=worker, args=("线程-B", 2))

t1.start()

t2.start()

# 主线程等待所有子线程完成

t1.join()

t2.join()

print("全部完成！")

```

输出：

```

[线程-A] 开始工作...

[线程-B] 开始工作...

[线程-A] 工作完成！

[线程-B] 工作完成！

全部完成！

```

注意看：两个线程几乎同时打印了"开始工作"，说明它们是真正**并行运行**的。总耗时约 2 秒，而不是单线程的 4 秒。

### 2.2 继承 Thread 类的方式

```python

import threading

import time

classTaskThread(threading.Thread):

"""自定义线程类"""

def__init__(self, task_name: str, duration: float):

super().__init__()

self.task_name = task_name

self.duration = duration

defrun(self) -> None:

print(f"[{self.task_name}] 执行中...")

        time.sleep(self.duration)

print(f"[{self.task_name}] 完成！耗时 {self.duration}s")

# 启动 3 个线程

threads = [

    TaskThread("下载图片", 1.5),

    TaskThread("查询数据库", 2.0),

    TaskThread("发送邮件", 1.0),

]

for t in threads:

    t.start()

for t in threads:

    t.join()

print("所有任务完成！")

```

### 2.3 线程安全：当多个线程共享数据时

这是新手最容易踩坑的地方！**多个线程同时修改同一个变量会导致数据错乱**：

```python

import threading

counter = 0

defunsafe_increment(iterations: int) -> None:

"""不安全的计数器（会有竞态条件）"""

global counter

for _ inrange(iterations):

        counter += 1# ← 这不是原子操作！

# 实际上分三步：读取 counter → 加 1 → 写回 counter

threads = []

for _ inrange(10):

    t = threading.Thread(target=unsafe_increment, args=(10000,))

    threads.append(t)

    t.start()

for t in threads:

    t.join()

print(f"期望结果: 100000, 实际结果: {counter}")

# 输出可能是 72341、85632 等——每次都不一样的"惊喜"！

```

这就是**竞态条件（Race Condition）**——你猜不到哪个线程先执行到哪一行，结果就不可预测。

#### 解决方案：使用锁（Lock）

```python

import threading

counter = 0

lock = threading.Lock()

defsafe_increment(iterations: int) -> None:

"""使用锁保证线程安全"""

global counter

for _ inrange(iterations):

with lock:  # ← 自动获取和释放锁

            counter += 1

threads = []

for _ inrange(10):

    t = threading.Thread(target=safe_increment, args=(10000,))

    threads.append(t)

    t.start()

for t in threads:

    t.join()

print(f"期望结果: 100000, 实际结果: {counter}")

# 每次都正确输出 100000 ✅

```

#### 更高级的锁：RLock、Semaphore、Condition

```python

import threading

import time

# ===== RLock：可重入锁（同一个线程可以多次获取同一把锁）=====

rlock = threading.RLock()

defnested_lock():

with rlock:

print("外层锁已获取")

with rlock:  # ← 同一个线程可以再获取

print("内层锁已获取")

# 嵌套锁不会死锁！

# ===== Semaphore：控制并发数量 =====

semaphore = threading.Semaphore(3)  # 最多 3 个线程同时进入

deflimited_worker(task_id: int):

with semaphore:

print(f"[{task_id}] 开始工作（当前并发数: {semaphore._value}）")

        time.sleep(2)

print(f"[{task_id}] 完成")

# 启动 10 个任务，但同一时间只有 3 个在执行

tasks = [threading.Thread(target=limited_worker, args=(i,)) for i inrange(10)]

for t in tasks:

    t.start()

for t in tasks:

    t.join()

print("Semaphored 测试完成！")

```

---

## 三、concurrent.futures：更优雅的并发写法

`threading` 是 Python 的"低层 API"，如果你不想手动管理线程的生命周期，可以用 `concurrent.futures`：

### 3.1 ThreadPoolExecutor：线程池

```python

from concurrent.futures import ThreadPoolExecutor, as_completed

import time

deffetch_url(url: str) -> str:

"""模拟抓取网页内容"""

    time.sleep(1)  # 模拟网络延迟

returnf"{url} 的内容（耗时 1s）"

urls = [

"https://example.com/1",

"https://example.com/2",

"https://example.com/3",

"https://example.com/4",

"https://example.com/5",

]

start = time.time()

with ThreadPoolExecutor(max_workers=3) as executor:

# 提交所有任务

    future_to_url = {executor.submit(fetch_url, url): url for url in urls}

# 按完成顺序处理结果

for future in as_completed(future_to_url):

        url = future_to_url[future]

try:

            result = future.result()

print(f"✅ {result}")

exceptExceptionas e:

print(f"❌ {url} 出错: {e}")

elapsed = time.time() - start

print(f"总耗时: {elapsed:.2f}s（理论值约 2s，因为 5 个任务 3 个并发）")

```

### 3.2 带返回值的并发：map 方法

```python

from concurrent.futures import ThreadPoolExecutor

import time

defcalculate_square(n: int) -> tuple:

    time.sleep(0.5)  # 模拟计算

return (n, n * n)

numbers = list(range(1, 11))  # 1 到 10

with ThreadPoolExecutor(max_workers=4) as executor:

    results = executor.map(calculate_square, numbers)

for num, square in results:

print(f"{num}² = {square}")

```

`map` 的特点是**保持输入输出的顺序**，而 `as_completed` 是按完成顺序返回。

### 3.3 提交任务并设置回调

```python

from concurrent.futures import ThreadPoolExecutor, Future

defdownload_file(filename: str) -> str:

import random

    time.sleep(random.uniform(1, 3))  # 随机耗时

returnf"{filename} 下载完成"

defon_success(future: Future) -> None:

print(f"成功: {future.result()}")

defon_error(future: Future) -> None:

print(f"失败: {future.exception()}")

with ThreadPoolExecutor(max_workers=5) as executor:

    futures = []

for i inrange(10):

        f = executor.submit(download_file, f"file_{i}.dat")

        f.add_done_callback(on_success)  # 成功时调用

        futures.append(f)

# 等待全部完成

from concurrent.futures import wait

    wait(futures)

```

---

## 四、multiprocessing：突破 GIL 限制

当你需要跑 CPU 密集型计算（图像处理、数值计算、机器学习训练）时，多线程不行——因为 GIL 锁着。这时要用**多进程**。

### 4.1 基本用法

```python

import multiprocessing

import time

defcpu_bound_task(n: int) -> int:

"""CPU 密集型任务：计算 1+2+...+n"""

    total = 0

for i inrange(1, n + 1):

        total += i

return total

if__name__ == "__main__":

    nums = [10_000_000] * 4# 4 个超级大的计算

    start = time.time()

# ===== 单线程版本 =====

    results = [cpu_bound_task(n) for n in nums]

    single_time = time.time() - start

print(f"单线程耗时: {single_time:.2f}s")

# ===== 多进程版本 =====

    start = time.time()

with multiprocessing.Pool(processes=4) as pool:

        results = pool.map(cpu_bound_task, nums)

    multi_time = time.time() - start

print(f"多进程耗时: {multi_time:.2f}s")

print(f"加速比: {single_time / multi_time:.2f}x")

# 多进程应该快约 3-4 倍！

```

### 4.2 Process 类手动管理进程

```python

import multiprocessing

import time

defprocess_data(data: list, queue: multiprocessing.Queue) -> None:

"""处理数据并将结果放入队列"""

    result = sum(data)

    queue.put(result)

print(f"收到数据: {data[:3]}..., 结果: {result}")

if__name__ == "__main__":

    data_chunks = [list(range(1000)) for _ inrange(5)]

    queue = multiprocessing.Queue()

    processes = []

for chunk in data_chunks:

        p = multiprocessing.Process(target=process_data, args=(chunk, queue))

        p.start()

        processes.append(p)

for p in processes:

        p.join()

# 从队列取结果

    results = []

whilenot queue.empty():

        results.append(queue.get())

print(f"所有结果: {results}")

print(f"总和: {sum(results)}")

```

### 4.3 进程间通信：共享内存

```python

import multiprocessing

import time

defshared_memory_counter(shared_array: multiprocessing.Array, lock: multiprocessing.Lock):

"""多个进程共享一个整数数组"""

for _ inrange(100000):

with lock:

            shared_array[0] += 1

if__name__ == "__main__":

# 创建一个可共享的整数数组，初始值为 0

    shared_value = multiprocessing.Array('i', [0])  # 'i' 表示 int

    lock = multiprocessing.Lock()

    processes = [

        multiprocessing.Process(target=shared_memory_counter, args=(shared_value, lock))

for _ inrange(4)

    ]

for p in processes:

        p.start()

for p in processes:

        p.join()

print(f"共享内存中的值: {shared_value[0]}")

# 应该等于 400000

```

### 4.4 多进程 + 异步编程结合（生产级写法）

```python

import multiprocessing

import asyncio

from concurrent.futures import ProcessPoolExecutor

defheavy_computation(x: int) -> int:

"""模拟一个耗时的 CPU 计算"""

returnsum(i * i for i inrange(x))

asyncdefasync_heavy_computation(values: list) -> list:

"""用 asyncio  + 进程池跑并发"""

    loop = asyncio.get_running_loop()

with ProcessPoolExecutor() as executor:

        futures = [

            loop.run_in_executor(executor, heavy_computation, v)

for v in values

        ]

        results = await asyncio.gather(*futures)

returnlist(results)

if__name__ == "__main__":

import time

    large_values = [100_000, 200_000, 150_000, 300_000]

    start = time.time()

    results = asyncio.run(async_heavy_computation(large_values))

    elapsed = time.time() - start

for val, result inzip(large_values, results):

print(f"heavy_computation({val}) = {result}")

print(f"总耗时: {elapsed:.2f}s")

```

---

## 五、生产者-消费者模式：三种实现

这是并发编程中最经典的设计模式之一，广泛用于数据管道、任务队列、消息系统等场景。

### 5.1 用 Queue 实现（最简单）

```python

import multiprocessing

import time

import random

defproducer(queue: multiprocessing.Queue, num_items: int):

"""生产者：生成数据放入队列"""

for i inrange(num_items):

        item = f"任务-{i}"

        queue.put(item)

print(f"📦 生产: {item}")

        time.sleep(random.uniform(0.1, 0.5))

defconsumer(queue: multiprocessing.Queue, worker_id: int):

"""消费者：从队列取出数据并处理"""

whileTrue:

try:

            item = queue.get(timeout=1)

if item isNone:  # 死亡信号

break

print(f"🔧 工人-{worker_id} 处理: {item}")

            time.sleep(random.uniform(0.2, 0.8))

except:  # 队列超时

continue

if__name__ == "__main__":

    queue = multiprocessing.Queue(maxsize=10)  # 缓冲区大小 10

    producers = [

        multiprocessing.Process(target=producer, args=(queue, 5)),

    ]

    consumers = [

        multiprocessing.Process(target=consumer, args=(queue, i))

for i inrange(3)

    ]

# 启动

for p in producers + consumers:

        p.start()

# 等待生产者结束

for p in producers:

        p.join()

# 发送死亡信号给消费者

for _ in consumers:

        queue.put(None)

# 等待消费者结束

for p in consumers:

        p.join()

print("🎉 所有生产者和消费者完成！")

```

### 5.2 用 multiprocessing.Manager 实现分布式队列

```python

import multiprocessing

import threading

import queue

import time

classProductionQueue:

"""线程安全的生产者-消费者队列"""

def__init__(self, maxsize: int = 10):

self.q = queue.Queue(maxsize=maxsize)

self.shutdown_event = threading.Event()

defproduce(self, item, timeout: float = 1.0):

try:

self.q.put(item, timeout=timeout)

returnTrue

except queue.Full:

print("队列已满，丢弃任务")

returnFalse

defconsume(self) -> object | None:

try:

returnself.q.get(timeout=0.5)

except queue.Empty:

ifself.shutdown_event.is_set():

returnNone

returnNone

defshutdown(self):

self.shutdown_event.set()

```

### 5.3 用 asyncio 实现（纯异步版本）

```python

import asyncio

import random

import time

asyncdefproducer(queue: asyncio.Queue, name: str, count: int):

"""异步生产者"""

for i inrange(count):

        item = f"{name}-item-{i}"

await queue.put(item)

print(f"[{name}] 生产: {item}")

await asyncio.sleep(random.uniform(0.1, 0.5))

asyncdefconsumer(queue: asyncio.Queue, worker_id: int):

"""异步消费者"""

whileTrue:

try:

            item = await asyncio.wait_for(queue.get(), timeout=1.0)

except asyncio.TimeoutError:

break

print(f"[消费者-{worker_id}] 处理: {item}")

await asyncio.sleep(random.uniform(0.2, 0.8))

        queue.task_done()

asyncdefmain():

    queue = asyncio.Queue(maxsize=20)

# 启动 2 个生产者

    producers = [

        asyncio.create_task(producer(queue, "P1", 5)),

        asyncio.create_task(producer(queue, "P2", 5)),

    ]

# 启动 3 个消费者

    consumers = [

        asyncio.create_task(consumer(queue, i))

for i inrange(3)

    ]

# 等待生产者完成

await asyncio.gather(*producers)

# 等队列清空

await queue.join()

# 等待消费者结束

for c in consumers:

        c.cancel()

print("🎉 异步生产者-消费者完成！")

if__name__ == "__main__":

    asyncio.run(main())

```

---

## 六、进阶技巧：GIL 绕过与混合架构

### 6.1 什么时候用多线程？什么时候用多进程？

| 场景 | 推荐方案 | 原因 |

|------|----------|------|

| 网络爬虫 | ThreadPoolExecutor | IO 等待时 GIL 会释放 |

| 文件批量处理 | ThreadPoolExecutor | IO 密集型 |

| 图像/视频处理 | ProcessPoolExecutor | CPU 密集型，要突破 GIL |

| 数值计算（NumPy/Pandas）| ThreadPoolExecutor 或 ProcessPoolExecutor | NumPy C 扩展会释放 GIL |

| Web 服务器 | ThreadPoolExecutor + asyncio | FastAPI 默认就用 uvicorn workers |

| 实时数据处理 | multiprocessing + Queue | 需要高性能 + 大数据量 |

### 6.2 subprocess 方式运行外部进程

有时候最好的并发不是 Python 原生的，而是调用外部工具：

```python

import subprocess

import concurrent.futures

import os

defping_host(hostname: str) -> str:

"""用系统 ping 命令检测主机是否可达"""

# Windows 用 -n 1, Linux/Mac 用 -c 1

    param = "-n 1"if os.name == "nt"else"-c 1"

try:

        result = subprocess.run(

            ["ping", param, hostname],

capture_output=True,

text=True,

timeout=5,

        )

if result.returncode == 0:

returnf"✅ {hostname}: 可达 ({result.stdout.split('平均')[1][:5] if'平均'in result.stdout else'N/A'})"

else:

returnf"❌ {hostname}: 不可达"

except subprocess.TimeoutExpired:

returnf"⏱️ {hostname}: 超时"

hosts = [

"127.0.0.1",

"8.8.8.8",

"1.1.1.1",

"nonexistent.invalid.domain.xyz",

"www.baidu.com",

]

with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:

    results = executor.map(ping_host, hosts)

for result in results:

print(result)

```

---

## 七、实操练习

### 练习题

**第 1 题：用多线程实现一个简单的下载器**

写一个程序，用线程池模拟同时下载多个 URL 的文件：

```python

from concurrent.futures import ThreadPoolExecutor, as_completed

import urllib.request

import time

defdownload_file(url: str, output_dir: str = "./downloads") -> str:

"""下载文件并保存到本地"""

    filename = url.split("/")[-1] or"index.html"

    filepath = f"{output_dir}/{filename}"

try:

        urllib.request.urlretrieve(url, filepath)

returnf"✅ 下载完成: {filepath}"

exceptExceptionas e:

returnf"❌ 下载失败: {url}, 原因: {e}"

urls = [

"https://www.example.com/",

"https://httpbin.org/get",

"https://jsonplaceholder.typicode.com/posts/1",

]

# TODO: 在这里用 ThreadPoolExecutor 并发下载

# ...

```

参考答案：

```python

with ThreadPoolExecutor(max_workers=5) as executor:

    futures = [executor.submit(download_file, url) for url in urls]

for future in as_completed(futures):

print(future.result())

```

---

**第 2 题：实现一个多线程计数器，验证 GIL 的影响**

分别用单线程和多线程对同一个数做 `+1` 操作 100 万次，观察耗时差异：

```python

import threading

import time

import multiprocessing

defcpu_intensive_work(n: int) -> int:

"""CPU 密集型：计算前 n 个整数的平方和"""

returnsum(i * i for i inrange(n))

# 方式一：单线程

single_start = time.time()

results_single = [cpu_intensive_work(5_000_000) for _ inrange(4)]

single_elapsed = time.time() - single_start

# 方式二：多线程（受 GIL 限制）

defworker_thread(results: list, idx: int, n: int):

    results[idx] = cpu_intensive_work(n)

multi_results = [None] * 4

multi_threads = [

    threading.Thread(target=worker_thread, args=(multi_results, i, 5_000_000))

for i inrange(4)

]

multi_start = time.time()

for t in multi_threads:

    t.start()

for t in multi_threads:

    t.join()

multi_elapsed = time.time() - multi_start

# 方式三：多进程（突破 GIL）

proc_results = [None] * 4

processes = [

    multiprocessing.Process(target=worker_thread, args=(proc_results, i, 5_000_000))

for i inrange(4)

]

proc_start = time.time()

for p in processes:

    p.start()

for p in processes:

    p.join()

proc_elapsed = time.time() - proc_start

print(f"单线程: {single_elapsed:.2f}s")

print(f"多线程: {multi_elapsed:.2f}s（受 GIL 限制，可能比单线程还慢！）")

print(f"多进程: {proc_elapsed:.2f}s（突破 GIL，应该最快）")

```

---

**第 3 题：实现一个带有优先级队列的任务调度器**

用 `queue.PriorityQueue` 实现一个按优先级调度的任务管理器：

```python

import queue

import threading

import time

import heapq

classPriorityTaskScheduler:

"""优先级任务调度器"""

def__init__(self, num_workers: int = 3):

self.task_queue = queue.PriorityQueue()

self.workers = []

self.running = True

for i inrange(num_workers):

            t = threading.Thread(target=self._worker_loop, args=(i,), daemon=True)

            t.start()

self.workers.append(t)

defsubmit_task(self, priority: int, task_name: str, task_func, *args):

"""提交任务，priority 越小越优先"""

self.task_queue.put((priority, task_name, task_func, args))

print(f"📋 提交任务: {task_name}（优先级: {priority}）")

def_worker_loop(self, worker_id: int):

"""Worker 循环：从队列取任务并执行"""

whileself.running:

try:

                priority, task_name, task_func, args = self.task_queue.get(timeout=1.0)

print(f"🔧 工人-{worker_id} 处理: {task_name}（优先级: {priority}）")

                task_func(*args)

self.task_queue.task_done()

except queue.Empty:

continue

defwait_all(self):

"""等待所有任务完成"""

self.task_queue.join()

defstop(self):

self.running = False

# 使用示例

defsimulate_task(name: str, duration: float = 1.0):

print(f"  ⏳ 执行 {name}，耗时 {duration}s...")

    time.sleep(duration)

print(f"  ✅ 完成 {name}")

scheduler = PriorityTaskScheduler(num_workers=3)

scheduler.submit_task(priority=3, task_name="备份数据库", task_func=simulate_task, "DB Backup", 2.0)

scheduler.submit_task(priority=1, task_name="发送紧急邮件", task_func=simulate_task, "Email", 0.5)

scheduler.submit_task(priority=2, task_name="更新索引", task_func=simulate_task, "Index", 1.5)

scheduler.submit_task(priority=1, task_name="告警通知", task_func=simulate_task, "Alert", 0.3)

scheduler.wait_all()

scheduler.stop()

print("🎉 所有任务完成！")

```

---

**第 4 题：用多进程 + 共享内存实现分布式计算框架**

模拟一个小型的"分布式计算"场景，多个 Worker 进程从中央调度器领取任务并汇总结果：

```python

import multiprocessing

import random

import time

defworker_process(worker_id: int, tasks_queue: multiprocessing.Queue, results_dict: multiprocessing.dict):

"""Worker 进程：从任务队列领取任务并计算结果"""

whileTrue:

        task = tasks_queue.get()

if task isNone:  # 停止信号

break

        task_type, payload = task

print(f"[Worker-{worker_id}] 处理: {task_type} -> {payload}")

# 模拟不同类型的任务耗时

if task_type == "compute":

            result = sum(i * i for i inrange(payload))

elif task_type == "random":

            result = random.randint(0, 1000)

else:

            result = -1

        key = f"worker-{worker_id}-{task_type}"

        results_dict[key] = result

print(f"[Worker-{worker_id}] 完成: {key} = {result}")

if__name__ == "__main__":

    task_types = ["compute", "random", "compute", "random", "compute"]

    payloads = [1_000_000, 500, 500_000, 1000, 2_000_000]

    tasks = [(t, p) for t, p inzip(task_types, payloads)]

    tasks.extend([(None,)] * 3)  # 3 个停止信号

    tasks_queue = multiprocessing.Queue()

for task in tasks:

        tasks_queue.put(task)