threading多线程深入

Python进阶编程专题 · Python多线程编程全面指南

专题：Python进阶编程系统学习

关键词：Python, threading, 多线程, Lock, RLock, Condition, Event, GIL

一、线程基础与创建方式

Python 的 threading 模块提供了丰富的多线程编程接口。线程是操作系统能够进行运算调度的最小单位，它被包含在进程之中，是进程中的实际运作单位。与进程相比，线程更加轻量，创建和切换的开销更小，且同一进程内的多个线程共享内存空间，这使得数据共享变得非常方便。

1.1 使用 Thread 类创建线程

最基本的创建方式是通过实例化 threading.Thread 类，并通过 target 参数指定线程要执行的函数。这是最常用也最直观的方式，适合大多数简单场景。

import threading
import time

def worker(name, delay):
    """模拟一个需要时间的任务"""
    for i in range(5):
        print(f"[{name}] 第 {i+1} 次执行")
        time.sleep(delay)
    print(f"[{name}] 任务完成")

# 创建两个线程
t1 = threading.Thread(target=worker, args=("线程A", 0.5), name="Worker-A")
t2 = threading.Thread(target=worker, args=("线程B", 0.8), name="Worker-B")

t1.start()
t2.start()

print("主线程继续执行其他操作...")

t1.join()
t2.join()

print("所有线程已完成，主线程结束")

1.2 通过继承 Thread 创建线程

对于更复杂的线程逻辑，可以通过继承 Thread 类并重写 run 方法来实现。这种方式将线程代码封装在类中，便于组织和管理复杂的线程行为。

import threading
import random

class DownloadTask(threading.Thread):
    def __init__(self, file_id):
        super().__init__()
        self.file_id = file_id
        self.progress = 0

    def run(self):
        print(f"[{self.name}] 开始下载文件 #{self.file_id}")
        for i in range(100, step=10):
            time.sleep(random.uniform(0.1, 0.3))
            self.progress = i
        self.progress = 100
        print(f"[{self.name}] 下载完成")

tasks = [DownloadTask(i) for i in range(3)]
for t in tasks:
    t.start()
for t in tasks:
    t.join()

1.3 name 与 daemon 参数详解

name 参数为线程设置一个可读的名称，便于调试和日志记录。默认情况下，Python 会为线程自动生成 Thread-1、Thread-2 等名称。通过 threading.current_thread().name 可以在运行时获取当前线程的名称。

daemon 参数决定线程是否为守护线程。守护线程的特点是：当所有非守护线程（即主线程和其他非守护线程）结束时，守护线程会被强制终止，无论其是否执行完毕。这个特性在后台任务、监控线程等场景中非常有用。

import threading
import time

def daemon_worker():
    try:
        while True:
            print("[守护线程] 心跳检测中...")
            time.sleep(1)
    except KeyboardInterrupt:
        print("[守护线程] 被终止")

def main_worker():
    print("[工作线程] 开始工作")
    time.sleep(3)
    print("[工作线程] 工作完成")

d = threading.Thread(target=daemon_worker, daemon=True, name="Heartbeat")
w = threading.Thread(target=main_worker, daemon=False, name="MainTask")

d.start()
w.start()
# 当 w 完成后，主线程退出，守护线程 d 会自动被终止

最佳实践：应避免在守护线程中执行关键资源操作（如数据库写入、文件保存），因为守护线程可能在任何时刻被强制终止，导致资源状态不一致。守护线程适合日志轮转、缓存刷新、监控心跳等非关键任务。

二、线程生命周期管理

理解线程的完整生命周期是编写可靠多线程程序的基础。一个线程从创建到结束会经历新建、就绪、运行、阻塞、终止等多个状态。Python 的 Thread 对象提供了三个核心方法来管理线程的生命周期。

2.1 start() —— 启动线程

start() 方法使线程进入就绪状态，等待操作系统的调度。每个线程对象只能调用一次 start()，重复调用会抛出 RuntimeError。调用 start() 后，线程会自动调用 run() 方法中的代码。

2.2 join() —— 等待线程结束

join(timeout=None) 方法使调用方阻塞，直到目标线程终止或达到超时时间。这是线程同步中最常用的手段之一，用于确保依赖关系——主线程需要等待子线程完成后才能继续执行。

2.3 is_alive() —— 检查线程状态

is_alive() 方法返回线程是否正在运行。在线程的 run() 方法开始执行到执行完毕之间，该方法返回 True，否则返回 False。

import threading
import time

def compute(n):
    print(f"计算 {n} 的平方...")
    time.sleep(2)
    return n * n

threads = []
results = []

for i in range(5):
    t = threading.Thread(target=lambda x=i: results.append(compute(x)), name=f"Calc-{i}")
    threads.append(t)

for t in threads:
    t.start()
    print(f"{t.name} 是否存活: {t.is_alive()}")

for t in threads:
    t.join(timeout=3.0)  # 最多等待3秒
    print(f"{t.name} 已结束: {not t.is_alive()}")

print(f"计算结果: {results}")

生命周期要点：线程启动后不能重启；join(timeout) 设置超时可以避免永久阻塞；is_alive() 在 run() 执行期间返回 True；主线程退出时不会等待守护线程，但会等待所有非守护线程。

2.4 线程状态转换图

状态	描述	触发条件
新建 (New)	Thread 对象已创建，未调用 start()	实例化 Thread 对象
就绪 (Runnable)	等待 CPU 调度	调用 start() 后
运行 (Running)	正在执行 run()	操作系统分配 CPU 时间片
阻塞 (Blocked)	等待锁、IO 或 join()	调用 acquire()、sleep()、join()
终止 (Terminated)	run() 执行完毕或被终止	run() 正常返回或抛出未捕获异常

三、线程安全与锁机制

当多个线程同时访问共享数据时，由于线程调度的不确定性，可能导致数据不一致的问题——这就是竞态条件（Race Condition）。锁机制是解决线程安全问题最基本的手段。

3.1 竞态条件示例

下面的例子演示了没有锁保护时，多个线程同时对同一个计数器执行自增操作导致的结果错误。即使每次只加 1，最终结果也远小于预期值，因为 count += 1 在底层是三步操作（读取、修改、写入），线程切换可能发生在任何一步之间。

import threading

counter = 0
ITERATIONS = 100000

def increment():
    global counter
    for _ in range(ITERATIONS):
        counter += 1  # 不是原子操作！

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

print(f"预期结果: {ITERATIONS * 10}")
print(f"实际结果: {counter}")
print(f"误差: {ITERATIONS * 10 - counter}")
# 输出：预期结果 1000000，实际结果往往远小于此值

3.2 Lock（互斥锁）

threading.Lock 是最基本的锁，只有"锁定"和"未锁定"两种状态。同一个线程多次调用 acquire() 会导致死锁——因为 Lock 不可重入。使用 Lock 时务必配合 try/finally 或 with 语句，确保锁最终被释放。

import threading

counter = 0
lock = threading.Lock()
ITERATIONS = 100000

def safe_increment():
    global counter
    for _ in range(ITERATIONS):
        with lock:  # 推荐使用上下文管理器，自动 acquire/release
            counter += 1

threads = [threading.Thread(target=safe_increment) for _ in range(10)]
for t in threads: t.start()
for t in threads: t.join()
print(f"加锁后结果: {counter}")  # 准确输出 1000000

3.3 RLock（可重入锁）

threading.RLock（可重入锁）与 Lock 的关键区别在于：同一个线程可以多次 acquire 同一个 RLock，而不会发生死锁。RLock 内部维护了一个计数器，记录当前持有线程的 acquire 次数，每次 acquire 计数器加 1，每次 release 减 1，只有减到 0 时其他线程才能获取该锁。

import threading

rlock = threading.RLock()

def outer():
    print("进入 outer 函数，尝试获取锁")
    with rlock:
        print("outer 已获取锁，调用 inner")
        inner()  # RLock 允许在同一个线程中再次获取
    print("outer 释放锁")

def inner():
    print("进入 inner 函数，尝试获取锁")
    with rlock:  # 如果是 Lock，这里会死锁；RLock 不会
        print("inner 成功获取锁（可重入）")
    print("inner 释放锁")

t = threading.Thread(target=outer)
t.start()
t.join()

选择建议：如果锁的获取/释放逻辑简单且不会嵌套，使用 Lock 即可。如果存在递归调用或嵌套锁获取的场景（如一个加锁方法调用另一个同样需要加锁的方法），必须使用 RLock 以避免死锁。

四、线程间通信机制

多线程编程中，线程之间经常需要协作——某个线程需要等待特定条件满足后才能继续执行，或者生产者线程需要将数据传递给消费者线程。Python 提供了多种线程间通信原语。

4.1 Event（事件对象）

threading.Event 是最简单的线程间通信机制之一。它内部维护了一个布尔标志位，通过 set()、clear()、wait() 三个方法实现线程间的信号传递。事件对象特别适合"一次性通知"场景——一个线程等待另一个线程发出信号。

import threading
import time

# 创建三个事件，分别控制三个启动阶段
event_ready = threading.Event()
event_start = threading.Event()
event_stop = threading.Event()

def runner(name):
    print(f"{name} 已就绪，等待发令枪...")
    event_ready.set()  # 通知主线程：我已就绪
    event_start.wait()   # 等待发令枪
    print(f"{name} 起跑！")
    for i in range(1, 6):
        time.sleep(0.3)
        print(f"{name} 跑了 {i} 秒")
    print(f"{name} 到达终点")
    event_stop.set()

event_ready.clear()
t1 = threading.Thread(target=runner, args=("选手A",))
t1.start()
event_ready.wait()  # 等待所有选手就绪
event_ready.clear()
print("所有选手已就绪，倒计时...")
time.sleep(1)
print("砰！发令枪响！")
event_start.set()  # 所有线程同时出发
t1.join()

4.2 Condition（条件变量）

threading.Condition 是比 Event 更灵活的同步原语，它结合了锁和等待/通知机制。条件变量适用于"多次通知"和"复杂条件判断"的场景。使用时需要先获取关联锁，然后调用 wait() 等待条件，或者调用 notify()/notify_all() 唤醒等待线程。

import threading
import time
import random

cv = threading.Condition()
buffer = []
MAX_SIZE = 5

def producer():
    global buffer
    for i in range(10):
        with cv:
            while len(buffer) >= MAX_SIZE:
                print("缓冲区已满，生产者等待...")
                cv.wait()
            item = random.randint(1, 100)
            buffer.append(item)
            print(f"生产者生产: {item}, 缓冲区内 [{len(buffer)}个]")
            cv.notify()  # 通知消费者
        time.sleep(random.uniform(0.1, 0.5))

def consumer():
    global buffer
    for _ in range(10):
        with cv:
            while len(buffer) == 0:
                print("缓冲区为空，消费者等待...")
                cv.wait()
            item = buffer.pop(0)
            print(f"消费者消费: {item}, 缓冲区内 [{len(buffer)}个]")
            cv.notify()  # 通知生产者
        time.sleep(random.uniform(0.2, 0.6))

pt = threading.Thread(target=producer, name="生产者")
ct = threading.Thread(target=consumer, name="消费者")
pt.start()
ct.start()
pt.join()
ct.join()
print("生产消费完成")

4.3 Queue（线程安全队列）

queue.Queue 是 Python 多线程编程中最实用、最推荐的数据交换方式。它内部已经实现了线程安全机制，无需手动加锁。Queue 提供 put()（阻塞式写入）、get()（阻塞式读取）、task_done() 和 join() 等方法，完美支持生产者-消费者模式。

import threading
import queue
import time
import random

q = queue.Queue(maxsize=10)
stop_signal = object()  # 哨兵值，用于通知消费者停止

def producer(count):
    for i in range(count):
        item = f"data-{i}"
        q.put(item)  # 队列满时自动阻塞
        print(f"生产: {item}")
        time.sleep(random.uniform(0.1, 0.3))
    q.put(stop_signal)  # 发送停止信号

def consumer():
    while True:
        item = q.get()  # 队列空时自动阻塞
        if item is stop_signal:
            q.task_done()
            break
        print(f"消费: {item}")
        time.sleep(random.uniform(0.2, 0.5))
        q.task_done()

pt = threading.Thread(target=producer, args=(10,))
ct = threading.Thread(target=consumer)
pt.start()
ct.start()
pt.join()
q.join()  # 等待所有任务处理完成
print("所有任务已完成")

通信方式对比：Event 适合一次性通知；Condition 适合复杂条件判断和多轮通知；Queue 是生产-消费模式的最佳实践，推荐优先使用。Queue 内部封装了锁和条件变量，使用最简洁。

五、local —— 线程本地数据

threading.local 类创建的对象可以在不同线程中保存各自独立的数据副本——每个线程对该对象的属性读写互不影响。这在"避免传递上下文参数"的场景中特别有用，比如在每个线程中保存数据库连接、HTTP 会话、用户请求上下文等。

import threading
import random

# 创建线程本地存储对象
thread_local = threading.local()

class DatabaseConnection:
    def __init__(self, conn_id):
        self.conn_id = conn_id
    def query(self, sql):
        return f"[连接{self.conn_id}] 执行: {sql}"

def get_connection():
    """获取当前线程的数据库连接（惰性初始化）"""
    if not hasattr(thread_local, "db_conn"):
        conn_id = threading.current_thread().name
        thread_local.db_conn = DatabaseConnection(conn_id)
        print(f"[{conn_id}] 创建了新的数据库连接")
    return thread_local.db_conn

def worker(sql):
    conn = get_connection()  # 每个线程第一次调用时创建连接
    result = conn.query(sql)
    print(result)
    # 再次调用，复用已有的连接
    conn2 = get_connection()
    print(f"同一个连接对象? {conn is conn2}")

threads = []
for i in range(3):
    t = threading.Thread(target=worker, args=(f"SELECT * FROM table_{i}",),
                         name=f"Thread-{i}")
    threads.append(t)

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

local 实现原理：threading.local 底层利用了 Python 的字典和线程 ID 作为键来存储和隔离数据。每个线程访问 local 对象时，实际上是通过当前线程的标识符来获取自己独有的数据字典。注意：local 对象本身是全局共享的，只是其属性被隔离到了每个线程。

六、线程池 —— ThreadPoolExecutor

手动创建和管理大量线程会导致两个问题：一是线程创建和销毁的开销很大，二是系统中可同时运行的线程数量受限于操作系统资源。线程池通过复用一组预先创建的线程来解决这些问题。concurrent.futures.ThreadPoolExecutor 是 Python 3.2+ 推荐的线程池实现。

from concurrent.futures import ThreadPoolExecutor, as_completed
import time
import random

def download_url(url):
    print(f"开始下载: {url}")
    time.sleep(random.uniform(0.5, 2.0))
    size = random.randint(1024, 99999)
    print(f"完成下载: {url} ({size} bytes)")
    return (url, size)

urls = [
    "https://example.com/file1.zip",
    "https://example.com/file2.zip",
    "https://example.com/file3.zip",
    "https://example.com/file4.zip",
    "https://example.com/file5.zip",
    "https://example.com/file6.zip",
    "https://example.com/file7.zip",
    "https://example.com/file8.zip",
]

# max_workers=4 表示最多同时运行4个线程
with ThreadPoolExecutor(max_workers=4) as executor:
    # submit 返回 Future 对象
    futures = {executor.submit(download_url, url): url for url in urls}

    # as_completed 在 Future 完成时立即返回
    for future in as_completed(futures):
        url, size = future.result()  # 获取返回值
        print(f"结果: {url} -> {size} bytes")

print("所有下载任务完成")

# 另一种方式：使用 map，按输入顺序返回结果
with ThreadPoolExecutor(max_workers=4) as executor:
    results = executor.map(download_url, urls)
    for result in results:
        print(f"有序结果: {result}")

Executor 最佳实践：使用 with 语句管理线程池生命周期，自动完成清理。submit() 返回 Future 对象，支持回调、取消和超时；map() 按输入顺序返回结果，更简洁但缺乏灵活性。max_workers 一般设置为 CPU 核心数的 5-10 倍（对于 IO 密集型任务）。

七、GIL 深度分析与影响

GIL（Global Interpreter Lock，全局解释器锁）是 CPython 解释器的一个核心设计决策，也是 Python 多线程编程中最重要的性能影响因素。理解 GIL 是写出高效并发代码的前提。

7.1 GIL 是什么

GIL 是一个互斥锁，它保证在任何时刻，同一个进程中只有一个线程在执行 Python 字节码。这意味着即使在多核 CPU 上，Python 多线程也无法真正并行执行计算密集型任务。GIL 的存在简化了 CPython 的内存管理（特别是引用计数），但也牺牲了多线程的并行计算能力。

7.2 IO 密集 vs CPU 密集实测对比

下面的对比实验清晰地展示了 GIL 对不同类型任务的影响：对于 IO 密集型任务，多线程可以显著提升性能（因为线程在等待 IO 时释放 GIL）；对于 CPU 密集型任务，多线程由于 GIL 竞争反而可能比单线程更慢。

IO 密集型任务 —— 多线程大幅提升

import threading
import time

def io_task(n):
    """模拟 IO 操作（网络请求、文件读写等）"""
    for _ in range(n):
        time.sleep(0.01)  # 模拟 IO 等待

# 单线程
start = time.time()
io_task(200)
print(f"单线程耗时: {time.time() - start:.2f}s")

# 多线程
start = time.time()
threads = [threading.Thread(target=io_task, args=(50,))
           for _ in range(4)]
for t in threads: t.start()
for t in threads: t.join()
print(f"4线程耗时: {time.time() - start:.2f}s")
# IO 密集：多线程快 3-4 倍

CPU 密集型任务 —— 多线程无提升甚至更慢

import threading
import time

def cpu_task(n):
    """模拟 CPU 密集计算"""
    result = 0
    for i in range(n * 1000000):
        result += i ** 2

# 单线程
start = time.time()
cpu_task(10)
print(f"单线程耗时: {time.time() - start:.2f}s")

# 多线程
start = time.time()
threads = [threading.Thread(target=cpu_task, args=(2,))
           for _ in range(5)]
for t in threads: t.start()
for t in threads: t.join()
print(f"5线程耗时: {time.time() - start:.2f}s")
# CPU 密集：多线程反而更慢（GIL 切换开销）

7.3 GIL 的应对策略

场景	推荐方案	说明
IO 密集型	多线程 (threading)	GIL 在 IO 等待时释放，多线程有效
CPU 密集型	多进程 (multiprocessing)	每个进程有独立 GIL，可真正并行
混合型	线程 + 进程组合	线程处理 IO，进程处理计算
CPU 密集型（轻量）	协程 (asyncio)	单线程内协作式调度，无 GIL 问题
极致性能需求	C 扩展 / 释放 GIL	C 代码可以显式释放 GIL

GIL 的核心启示：Python 多线程并非"无用"，而是需要根据任务类型选择合适的并发模型。对于 IO 密集型任务（网络爬虫、Web 服务器、文件处理），多线程是非常有效的；对于 CPU 密集型任务（图像处理、科学计算），应使用 multiprocessing 或 asyncio。Python 3.13 引入了"自由线程"（free-threaded）模式作为实验特性，未来有望逐步解决 GIL 的限制。

八、多线程常见陷阱与调试

多线程编程中最棘手的问题往往不是功能实现，而是并发环境下特有的"幽灵"Bug——它们可能只在特定执行顺序下出现，极难复现和调试。

8.1 死锁（Deadlock）

死锁是指两个或多个线程互相等待对方释放锁，导致所有线程都无法继续执行的状态。经典的死锁场景是"锁顺序不一致"——线程 A 持有锁 1 等待锁 2，线程 B 持有锁 2 等待锁 1。

import threading
import time

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

def transfer_from_a_to_b():
    with lock_a:
        time.sleep(0.1)  # 增加死锁概率
        print("[A->B] 已获取锁A，等待锁B...")
        with lock_b:
            print("[A->B] 成功获取锁B，执行转账")

def transfer_from_b_to_a():
    with lock_b:
        time.sleep(0.1)  # 增加死锁概率
        print("[B->A] 已获取锁B，等待锁A...")
        with lock_a:
            print("[B->A] 成功获取锁A，执行转账")

t1 = threading.Thread(target=transfer_from_a_to_b)
t2 = threading.Thread(target=transfer_from_b_to_a)
t1.start()
t2.start()
t1.join(timeout=2)
t2.join(timeout=2)
print(f"线程1存活: {t1.is_alive()} | 线程2存活: {t2.is_alive()}")
# 大概率两个线程都处于死锁状态，程序无法继续

死锁预防策略：1) 固定锁获取顺序——所有线程用相同的顺序获取锁（如总是先 A 后 B）；2) 使用 acquire(timeout) 设置超时，获取失败时释放已持有的锁；3) 使用 threading.RLock 避免单线程内的死锁；4) 尽量缩小加锁范围，减少锁持有时间。

8.2 竞态条件（Race Condition）

竞态条件是指程序的执行结果依赖于线程的调度顺序，当多个线程同时读写共享数据时，由于操作交叉执行导致数据不一致。竞态条件的典型特征是：每次运行结果不同，且难以稳定复现。

import threading

shared_dict = {}

def buggy_update(key, value):
    # 检查 - 修改 - 写入 三步非原子操作
    if key not in shared_dict:     # 步骤1: 检查
        # 这里可能被其他线程切换
        shared_dict[key] = []     # 步骤2: 初始化
    shared_dict[key].append(value)  # 步骤3: 修改

# 正确做法：使用锁保护整个检查-修改过程
lock = threading.Lock()

def safe_update(key, value):
    with lock:  # 原子化检查-修改操作
        if key not in shared_dict:
            shared_dict[key] = []
        shared_dict[key].append(value)

8.3 活锁（Livelock）和锁饥饿

活锁是指线程虽然没有被阻塞，但由于不断地响应对方而无法继续执行。想象两个人在走廊相遇，都试图给对方让路，结果同时向左又同时向右，始终无法通过。与死锁不同，活锁中的线程是在"运行"的，但没有任何实质进展。锁饥饿是指某个线程长期无法获取到所需的锁，被其他线程"饿死"。

调试建议：1) 利用 logging 模块记录线程 ID 和锁获取/释放时间；2) 使用 faulthandler.enable() 在死锁时输出线程堆栈；3) 使用 Py-spy 等第三方工具实时查看线程状态；4) 尽可能编写确定性测试，用 threading.Barrier 控制线程执行顺序来复现竞态条件。

九、最佳实践与综合示例

9.1 线程安全的单例模式

import threading

class ThreadSafeSingleton:
    _instance = None
    _lock = threading.Lock()

    def __new__(cls, *args, **kwargs):
        if cls._instance is None:
            with cls._lock:
                # 双重检查锁定（Double-Checked Locking）
                if cls._instance is None:
                    cls._instance = super().__new__(cls)
        return cls._instance

9.2 综合：带超时的并发任务调度器

import threading
import queue
import time
from concurrent.futures import ThreadPoolExecutor, TimeoutError

class ConcurrentTaskScheduler:
    def __init__(self, max_workers=4):
        self.executor = ThreadPoolExecutor(max_workers=max_workers)
        self.results = queue.Queue()
        self._shutdown = threading.Event()

    def submit_task(self, task_fn, task_name, timeout=5):
        future = self.executor.submit(task_fn)
        future.task_name = task_name
        future.timeout = timeout
        return future

    def collect_results(self, futures, timeout=10):
        for future in futures:
            try:
                result = future.result(timeout=timeout)
                self.results.put((future.task_name, result, None))
            except TimeoutError:
                self.results.put((future.task_name, None, "超时"))
            except Exception as e:
                self.results.put((future.task_name, None, str(e)))

    def shutdown(self):
        self.executor.shutdown(wait=True)

# 使用示例
def fetch_data(url):
    time.sleep(1)  # 模拟网络请求
    return f"数据来自 {url}"

scheduler = ConcurrentTaskScheduler(max_workers=3)
tasks = [
    scheduler.submit_task(lambda: fetch_data("api/endpoint1"), "任务1"),
    scheduler.submit_task(lambda: fetch_data("api/endpoint2"), "任务2"),
]
scheduler.collect_results(tasks)
scheduler.shutdown()
while not scheduler.results.empty():
    name, result, error = scheduler.results.get()
    if error:
        print(f"{name} 失败: {error}")
    else:
        print(f"{name} 成功: {result}")

9.3 选择正确的并发模型

并发模型	适用场景	优点	缺点
threading	IO 密集型	共享内存、编程直观	GIL 限制 CPU 并行
multiprocessing	CPU 密集型	真正并行、独立 GIL	通信开销大、资源重
asyncio	高并发 IO	轻量、可大规模并发	回调复杂、库生态待完善
ThreadPoolExecutor	任务管理	线程复用、API 友好	受限于 GIL

十、核心要点总结

线程创建的方式：直接使用 Thread(target=...) 最为灵活，继承 Thread 类适合复杂逻辑。daemon 守护线程在非守护线程全部退出时自动终止。
生命周期管理：start() 启动，join() 等待，is_alive() 检查运行状态。对已启动的线程重复 start() 会导致 RuntimeError。
锁机制：Lock 是基础互斥锁，不可重入；RLock 可重入，适用于嵌套锁场景。总是使用 with 语句管理锁的生命周期。
线程间通信：Event 适合一次性通知，Condition 适合复杂条件等待，Queue 是生产-消费模式的首选方案。
线程本地数据：threading.local 在每个线程中维护独立数据副本，避免传递上下文参数，适用于数据库连接、会话管理。
线程池：ThreadPoolExecutor 复用线程减少开销，submit() 返回 Future 对象，as_completed() 在任务完成时立即处理。
GIL 的影响：IO 密集任务多线程有效（GIL 在等待时释放），CPU 密集任务多线程无效甚至更慢，应改用多进程。
常见陷阱：死锁（固定锁顺序+超时避免）、竞态条件（原子操作+锁保护）、活锁（退避策略）。多线程 Bug 最难调试，提前预防比事后修复重要得多。