曦杨

外观

多线程和多进程

约 2209 字大约 7 分钟

2026-08-24

之前的课程中我们一直在讲异步编程，它适用于 I/O 密集型 任务。但如果遇到 CPU 密集型 任务，或者需要调用同步阻塞的库，多线程就派上用场了。

进程 vs 线程 vs 协程

对比维度🖥️ 进程 (Process)🧵 线程 (Thread)🍃 协程 (Coroutine)
运行载体操作系统内核态操作系统内核态用户态（事件循环）
资源开销极高 独立地址空间、文件描述符等，创建需 fork/clone中等 共享进程地址空间，独立栈空间（数 MB）极低 共享线程栈，状态极小（几 KB）
切换成本最昂贵 涉及 TLB 刷新、页表切换昂贵 内核态切换，数百 CPU 周期非常廉价 纯用户态，数十 CPU 周期
数据共享需 IPC（管道、共享内存、socket 等）共享内存，需同步机制单线程天然安全，但需注意非原子操作
利用多核✅ 原生支持✅ 原生支持❌ 单线程内不支持
适用场景隔离性要求高的多任务CPU 密集型、同步阻塞 I/O海量 I/O 密集型
创建数量级十级别百/千级别万/十万级别

threading 模块

threading 是 Python 标准库中用于多线程编程的模块。

创建线程

import threading
import time


def worker(name: str, duration: int):
    print(f"线程 {name} 开始工作")
    time.sleep(duration)
    print(f"线程 {name} 工作完成")


# 创建线程
t = threading.Thread(target=worker, args=("A", 2))
t.start()  # 启动线程
print("主线程继续执行")
t.join()  # 等待线程结束
print("主线程等待结束")

输出：

线程 A 开始工作
主线程继续执行
...
线程 A 工作完成
主线程等待结束

继承 Thread 类

import threading
import time


class WorkerThread(threading.Thread):
    def __init__(self, name: str, duration: int):
        super().__init__(name=name)
        self._duration = duration

    def run(self) -> None:
        print(f"线程 {self.name} 开始，参数: {self._duration}")
        time.sleep(self._duration)
        print(f"线程 {self.name} 完成")


t = WorkerThread("B", 1)
t.start()
t.join()

守护线程 (Daemon)

主线程退出时，非守护线程会阻止进程退出，守护线程会被强制终止。

import threading
import time


def daemon_worker():
    while True:
        print("守护线程运行中...")
        time.sleep(1)


d = threading.Thread(target=daemon_worker, daemon=True)
d.start()

time.sleep(3)
print("主线程结束，守护线程被强制终止")

注意：守护线程中不应操作资源（如写文件），因为可能在操作过程中被强制终止。

线程安全与竞态条件

多个线程同时访问共享变量时，会出现竞态条件 (Race Condition)：

import threading
import time


class BankAccount:
    def __init__(self, balance):
        self.balance = balance


def withdraw(account, amount, person_name):
    """取款函数 - 有竞态条件漏洞"""
    print(f"{person_name}: 查询余额，当前有 {account.balance} 元")

    # 关键漏洞：检查余额和扣款不是原子操作
    if account.balance >= amount:
        print(f"{person_name}: 余额充足，开始取款...")

        # 这个延迟让另一个线程有机会插进来
        time.sleep(0.1)  # 模拟输入密码、出钞等过程

        # 扣款
        account.balance -= amount
        print(f"{person_name}: ✓ 取款{amount}元成功！剩余 {account.balance} 元")
        return True
    else:
        print(f"{person_name}: ✗ 余额不足，取款失败")
        return False


# 创建一个账户，余额1000元
account = BankAccount(1000)

# 两个人同时取款800元
person1 = threading.Thread(target=withdraw, args=(account, 800, "张三"))
person2 = threading.Thread(target=withdraw, args=(account, 800, "李四"))

# 同时启动
print("=== 两个人同时开始取款 ===")
person1.start()
person2.start()

# 等待两人完成
person1.join()
person2.join()

print("\n=== 最终结果 ===")
print(f"账户余额: {account.balance} 元")
print(f"如果正常，应该只剩: {1000 - 800} = 200 元")
print(f"两人共取出了: {1600 - account.balance} 元")

account.balance -= amount 并非原子操作，它对应三条 CPU 指令：LOAD balance → SUB amount → STORE balance。线程切换可能发生在任意两条指令之间。更关键的是，检查余额和扣款这两个步骤之间也存在时间窗口。

Lock —— 互斥锁

Lock 确保同一时刻只有一个线程可以访问临界区，用锁修复上面的银行账户问题：

import threading
import time


class BankAccount:
    def __init__(self, balance):
        self.balance = balance
        self._lock = threading.Lock()


def withdraw(account: BankAccount, amount: int, person_name: str):
    """取款函数 - 使用锁保证线程安全"""
    with account._lock:  # 获取锁，同一时刻只有一个线程能进入
        print(f"{person_name}: 查询余额，当前有 {account.balance} 元")

        if account.balance >= amount:
            print(f"{person_name}: 余额充足，开始取款...")
            time.sleep(0.1)
            account.balance -= amount
            print(f"{person_name}: ✓ 取款{amount}元成功！剩余 {account.balance} 元")
            return True
        else:
            print(f"{person_name}: ✗ 余额不足，取款失败")
            return False


account = BankAccount(1000)

person1 = threading.Thread(target=withdraw, args=(account, 800, "张三"))
person2 = threading.Thread(target=withdraw, args=(account, 800, "李四"))

print("=== 两个人同时开始取款 ===")
person1.start()
person2.start()
person1.join()
person2.join()

print("\n=== 最终结果 ===")
print(f"账户余额: {account.balance} 元")
print(f"如果正常，应该只剩: {1000 - 800} = 200 元")

锁的常见问题：

import threading


lock = threading.Lock()

# 同一个线程重复 acquire 会死锁
lock.acquire()
lock.acquire()  # 死锁！我锁我自己 —— 因为线程还没释放就再次申请

RLock —— 可重入锁

RLock 允许同一个线程多次 acquire，内部维护一个计数器：

import threading

lock = threading.RLock()


def recurse(n: int):
    with lock:
        if n > 0:
            print(f"Recursing with n={n}")
            recurse(n - 1)  # 同一个线程再次 acquire ✅


recurse(5)  # 正常运行

Semaphore —— 限制并发数

import threading
import time

semaphore = threading.Semaphore(3)


def limited_worker(n: int):
    with semaphore:
        print(f"线程 {n} 进入\n", end="")
        time.sleep(1)
        print(f"线程 {n} 离开\n", end="")


threads = [threading.Thread(target=limited_worker, args=(i,)) for i in range(10)]

for t in threads:
    t.start()
for t in threads:
    t.join()

Event —— 线程间通知

import threading
import time


event = threading.Event()


def waiter():
    print("waiter: 开始等待")
    event.wait()  # 等待事件被设置
    print("waiter: 被唤醒")


def setter():
    print("setter: 1秒后设置事件")
    time.sleep(1)
    event.set()
    print("setter: 事件已设置")


t1 = threading.Thread(target=waiter)
t2 = threading.Thread(target=setter)
t1.start()
t2.start()
t1.join()
t2.join()

Queue —— 线程安全的生产者消费者

import queue
import random
import threading
import time


def producer(q: queue.Queue):
    for i in range(10):
        item = f"item_{i}"
        q.put(item)
        print(f"生产: {item}\n", end="")
        time.sleep(random.random())
    q.put(None)


def consumer(name: str, q: queue.Queue):
    while True:
        item = q.get()
        if item is None:
            q.task_done()
            break
        print(f"{name} 消费: {item}\n", end="")
        q.task_done()


q = queue.Queue(maxsize=5)
threads = [
    threading.Thread(target=producer, args=(q,)),
    threading.Thread(target=consumer, args=("C1", q)),
    threading.Thread(target=consumer, args=("C2", q)),
]

for t in threads:
    t.start()
for t in threads:
    t.join()

queue.Queue 内部已经实现了线程同步，无需额外加锁。

ThreadPoolExecutor —— 线程池

频繁创建线程开销大，使用线程池复用线程：

from concurrent.futures import ThreadPoolExecutor
import time


def fetch_url(url: str) -> str:
    time.sleep(1)  # 模拟网络请求
    return f"{url} 完成"


with ThreadPoolExecutor(max_workers=3) as executor:
    urls = ["url1", "url2", "url3", "url4", "url5"]

    # map 返回结果的顺序与传入顺序一致
    results = executor.map(fetch_url, urls)
    for r in results:
        print(r)

    # 也可以 submit 逐条提交
    futures = [executor.submit(fetch_url, url) for url in urls]
    for f in futures:
        print(f.result())

ThreadPoolExecutor 与 asyncio 配合

import asyncio
import time
from concurrent.futures import ThreadPoolExecutor


def blocking_io() -> str:
    time.sleep(0.5)  # 同步阻塞
    return "文件读取完成"


async def main():
    # to_thread 将同步阻塞函数放到线程池中执行
    result = await asyncio.to_thread(blocking_io)
    print(result)

    # 也可以手动指定执行器
    loop = asyncio.get_running_loop()
    with ThreadPoolExecutor() as pool:
        result = await loop.run_in_executor(pool, blocking_io)
        print(result)


asyncio.run(main())

线程局部数据 (Thread Local)

每个线程拥有独立的副本，互不干扰：

import threading
import time


local_data = threading.local()


def worker(name: str):
    local_data.name = name  # 每个线程独立存储
    local_data.count = 0
    for _ in range(3):
        local_data.count += 1
        print(f"{local_data.name}: {local_data.count}")
        time.sleep(0.1)


threads = [
    threading.Thread(target=worker, args=("A",)),
    threading.Thread(target=worker, args=("B",)),
]

for t in threads:
    t.start()
for t in threads:
    t.join()

多线程常见问题

GIL —— 全局解释器锁

CPython 中有一个 GIL (Global Interpreter Lock)，它保证同一时刻只有一个线程在执行 Python 字节码：

import threading
import time


def cpu_intensive():
    total = 0
    for i in range(50_000_000):
        total += i * i
    return total


# 多线程 vs 单线程 —— 多线程不会更快！
t1 = threading.Thread(target=cpu_intensive)
t2 = threading.Thread(target=cpu_intensive)

start = time.time()
t1.start()
t2.start()
t1.join()
t2.join()
print(f"多线程: {time.time() - start:.2f}s")

start = time.time()
cpu_intensive()
cpu_intensive()
print(f"单线程: {time.time() - start:.2f}s")

GIL 的存在意味着：Python 多线程无法利用多核 CPU 加速 CPU 密集型任务。

那多线程的意义在哪？对于 I/O 密集型 任务，线程在等待 I/O 时会释放 GIL，其他线程可以继续执行，所以仍然有加速效果。

CPU 密集型 —— 应该用多进程

from multiprocessing import Process


def cpu_intensive():
    total = 0
    for i in range(50_000_000):
        total += i * i
    return total


p1 = Process(target=cpu_intensive)
p2 = Process(target=cpu_intensive)
p1.start()
p2.start()
p1.join()
p2.join()

死锁

import threading
import time


lock_a = threading.Lock()
lock_b = threading.Lock()


def task_1():
    with lock_a:
        time.sleep(0.1)
        with lock_b:  # 等待 lock_b
            print("task_1 完成")


def task_2():
    with lock_b:
        time.sleep(0.1)
        with lock_a:  # 等待 lock_a
            print("task_2 完成")


t1 = threading.Thread(target=task_1)
t2 = threading.Thread(target=task_2)
t1.start()
t2.start()
t1.join()
t2.join()
# 死锁！两个线程互相等待对方释放锁

解决死锁的原则：固定锁的获取顺序

import threading
import time


lock_a = threading.Lock()
lock_b = threading.Lock()


def task_1():
    with lock_a:
        time.sleep(0.1)
        with lock_b:
            print("task_1 完成")


def task_2():
    with lock_a:  # 与 task_1 获取锁的顺序一致
        time.sleep(0.1)
        with lock_b:
            print("task_2 完成")

何时用线程，何时用协程？

场景推荐方案
CPU 密集型multiprocessing / ProcessPoolExecutor
同步 I/O 密集型（文件读写、数据库驱动阻塞）threading / ThreadPoolExecutor
异步 I/O 密集型（网络爬虫、Web 服务）asyncio / 协程
调用第三方同步库用 asyncio.to_thread 或 run_in_executor 包装