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 类
- 定义子类：创建一个类，并继承
threading.Thread。 - 重写 init 方法：在初始化方法中，必须调用
super().__init__()以确保线程对象被正确初始化。你可以在这里接收并保存自定义参数。 - 重写 run 方法：将线程需要执行的业务逻辑代码编写在
run()方法内。当线程启动时，会自动执行这里的代码。 - 启动线程：实例化自定义的线程类，并调用
start()方法启动线程。
class MyThread(threading.Thread):
def __init__(self, name, delay):
super().__init__() # 必须调用父类构造函数
self.name = name
self.delay = delay
def run(self):
# 线程执行的核心逻辑
print(f"线程【{self.name}】启动")
time.sleep(self.delay) # 模拟耗时操作
print(f"线程【{self.name}】结束")
if __name__ == "__main__":
# 创建自定义线程对象
t1 = MyThread("线程-1", 2)
t2 = MyThread("线程-2", 1)
# 启动线程（会自动调用 run 方法）
t1.start()
t2.start()
# 阻塞主线程，等待子线程执行完毕
t1.join()
t2.join()
print("所有线程执行完毕")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 worker():
print("非守护线程：开始干活...")
time.sleep(3) # 模拟耗时 3 秒的工作
print("非守护线程：干完了！")
t = threading.Thread(target=worker)
# t.daemon = False # 默认就是 False，可以不写
t.start()
print("主线程：我要下班了！")
# 主线程结束，但因为 t 是非守护线程，进程会等待 t 执行完毕（耗时约 3 秒）才真正退出。import threading
import time
def worker():
print("守护线程：开始干活...")
time.sleep(3) # 模拟耗时 3 秒的工作
print("守护线程：干完了！") # 这行永远不会被打印
t = threading.Thread(target=worker)
t.daemon = True # 设置为守护线程
t.start()
time.sleep(1) # 稍微等一下，确保子线程打印了第一句话
print("主线程：我要下班了！")
# 主线程结束，发现 t 是守护线程，直接强制终止它，进程瞬间退出。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()waiter: 开始等待
setter: 1秒后设置事件
setter: 事件已设置
waiter: 被唤醒
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 包装 |
