Karene's Blog
文章
合集Python 并发编程第 2 / 7 篇

多线程

约 23 分钟

从 threading.Thread 原语到 ThreadPoolExecutor 的完整路径。多线程是 Python 最常用的并发方案，简单、好用、大多数 I/O 密集场景够用。

1. threading.Thread 基本用法

1.1 最简单的线程

import threading

def worker(n):
    print(f"线程 {n} 开始")
    # ... 执行任务 ...
    print(f"线程 {n} 结束")

# 创建线程
threads = [threading.Thread(target=worker, args=(i,)) for i in range(5)]

# 启动（不是 run！）
for t in threads:
    t.start()

# 等待所有线程完成
for t in threads:
    t.join()

print("全部完成")

1.2 守护线程（daemon）

t = threading.Thread(target=background_task, daemon=True)
t.start()
# 主线程结束时，daemon 线程会被自动终止

守护线程适合后台任务（心跳检测、日志刷盘），主线程退出时不需要等它。

1.3 join() 的作用

t.start()
t.join()   # 阻塞当前线程，直到 t 执行完毕
print("t 已完成")

不调用 join()，主线程不会等子线程——程序可能在子线程还没跑完时就退出了。

2. 竞态条件：为什么需要锁

2.1 一个计数器的例子

import threading

counter = 0

def increment():
    global counter
    for _ in range(100_000):
        counter += 1   # 这一行不是原子操作！

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

print(counter)
# 期望 1_000_000，实际可能是 283_457、412_891... 每次都不同

2.2 为什么会这样

counter += 1 看起来是一行代码，实际分三步：

  1. 读取 counter 的当前值
  2. 加 1
  3. 写回 counter

两个线程可能同时读到同一个值，各自加 1 后写回——结果只加了 1，而不是 2。这就是竞态条件（Race Condition）。

2.3 用 Lock 修复

import threading

counter = 0
lock = threading.Lock()

def increment():
    global counter
    for _ in range(100_000):
        with lock:        # 自动获取和释放锁
            counter += 1  # 临界区：同一时刻只有一个线程执行

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

print(counter)   # 1_000_000（加锁保证正确性）

with lock 是最推荐的写法——自动获取锁，执行完自动释放，即使出异常也不会忘记释放。

3. 同步原语速览

Python 的 threading 模块提供了多种同步工具：

原语说明典型场景
Lock互斥锁，同一时刻只有一个线程持有保护共享变量
RLock可重入锁，同一线程可多次获取嵌套锁场景（防止死锁）
Semaphore信号量，控制同时访问的线程数量限制并发数（如限速器）
Event线程间事件通知一个线程通知，多个线程等待
Condition条件变量，支持 wait/notify生产者-消费者模式
Barrier栅栏，所有线程到齐后同时继续多阶段并行计算

常用示例

import threading
import time

# --- Semaphore：限制最多 3 个线程同时执行 ---
semaphore = threading.Semaphore(3)

def limited_task(n):
    with semaphore:
        print(f"任务 {n} 执行中")
        time.sleep(1)

# --- Event：线程间通知 ---
event = threading.Event()

def producer():
    time.sleep(1)
    print("数据就绪")
    event.set()   # 通知消费者

def consumer():
    event.wait()  # 阻塞直到 event 被 set
    print("开始处理")

4. ThreadPoolExecutor 重点讲解

concurrent.futures.ThreadPoolExecutor 是现代 Python 多线程的首选 API。

4.1 executor.map vs submit + as_completed

from concurrent.futures import ThreadPoolExecutor, as_completed
import time

def fetch(url):
    time.sleep(1)
    return f"result of {url}"

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

方式一：map —— 简单，保持输入顺序

with ThreadPoolExecutor(max_workers=4) as executor:
    results = list(executor.map(fetch, urls))
# results 按 urls 的顺序返回，不管哪个先完成
# results = ['result of https://a.com', 'result of https://b.com', ...]

方式二：submit + as_completed —— 灵活，按完成顺序

with ThreadPoolExecutor(max_workers=4) as executor:
    futures = {executor.submit(fetch, url): url for url in urls}

    for future in as_completed(futures):
        url = futures[future]
        try:
            result = future.result()
            print(f"{url}: {result}")
        except Exception as e:
            print(f"{url} 失败: {e}")
对比mapsubmit + as_completed
结果顺序按输入顺序按完成顺序
错误处理第一个异常就抛出可以逐个捕获
灵活性低（固定模式）高（可以做更复杂的逻辑）
适用场景简单的批量处理需要"先完成先处理"或错误隔离

4.2 future 对象

submit() 返回一个 Future 对象，代表一个异步操作的结果：

future = executor.submit(fetch, url)

# 获取结果（阻塞直到完成）
result = future.result()

# 获取结果，带超时
result = future.result(timeout=10)

# 检查是否完成
future.done()          # True / False

# 获取异常（如果任务抛了异常）
future.exception()     # Exception 对象，或 None

# 取消任务（如果还没开始执行）
future.cancel()        # True / False

4.3 max_workers 怎么选

场景建议值原因
I/O 密集（网络请求）10-100线程大部分时间在等，多开无妨
I/O 密集（文件读写）10-50受磁盘 I/O 限制
CPU 密集CPU 核心数但建议用多进程代替

5. 实战模式：限速器（Rate Limiter）

API 通常有频率限制（如每秒 10 次请求）。用 Semaphore 可以实现简单的限速：

import threading
import time
from concurrent.futures import ThreadPoolExecutor

# 信号量限制同时最多 5 个请求
semaphore = threading.Semaphore(5)
# 额外的锁控制每秒请求速率
rate_lock = threading.Lock()
request_count = 0

def rate_limited_fetch(url):
    """带限速的 HTTP 请求。"""
    global request_count

    with semaphore:   # 限制并发数
        with rate_lock:
            request_count += 1
            if request_count >= 10:
                print("达到速率上限，等待...")
                time.sleep(1)
                request_count = 0

        # 实际的请求逻辑
        time.sleep(0.1)   # 模拟 HTTP 请求
        return f"OK: {url}"

urls = [f"https://api.example.com/item/{i}" for i in range(20)]

with ThreadPoolExecutor(max_workers=5) as executor:
    results = list(executor.map(rate_limited_fetch, urls))

print(f"完成 {len(results)} 个请求")

6. 线程安全注意事项

6.1 不是所有操作都线程安全

# 这些操作是原子的（CPython 实现保证）：
# - counter += 1  （实际上是 .load + .add + .store，不原子！）
# - list.append(x) 是原子的
# - dict[key] = value 是原子的

# 但组合操作不是原子的：
# - if key not in d: d[key] = value  ← 竞态条件
# - if not lst: lst.append(x)  ← 竞态条件

6.2 用 Queue 代替共享变量

import queue
import threading

q = queue.Queue()   # 线程安全的队列

def producer():
    for i in range(10):
        q.put(i)

def consumer():
    while True:
        item = q.get()
        if item is None:
            break
        print(f"处理: {item}")
        q.task_done()

t1 = threading.Thread(target=producer)
t2 = threading.Thread(target=consumer)
t1.start(); t2.start()
t1.join(); q.put(None); t2.join()

queue.Queue 内部已处理好锁，是线程间传递数据的首选方式。

7. 小结

知识点要记住的
Thread 基本用法start() 启动，join() 等待，daemon=True 守护
竞态条件共享变量并发修改导致数据不一致
Lockwith lock: 保护临界区，永远用上下文管理器
ThreadPoolExecutor首选 API，map 适合简单场景，submit + as_completed 适合复杂场景
max_workersI/O 密集可以设大，CPU 密集不超过核心数
queue.Queue线程间传数据的首选方式，内部已加锁

下一篇：python-multiprocessing，当 GIL 阻碍你时，多进程是 CPU 密集任务的解药。