threading 多线程

threading 提供共享内存的轻量并发：线程间数据零拷贝互通，代价是必须自己处理竞态。受 GIL 全局解释器锁约束，它适合 I/O 密集任务而非 CPU 密集。

创建与生命周期

两种等价写法：传函数（常用）或子类化 Thread：

import threading
 
def worker(n):
    print(f"task {n}")
 
t = threading.Thread(target=worker, args=(1,), name="w1")
t.start()  # 真正创建线程并调度，只能调用一次
t.join()   # 阻塞等待线程结束，可加timeout

start() 之前线程不存在；直接调 run() 只是普通函数调用，不会并发。

daemon 线程：daemon=True 的线程在后台运行，主线程退出时被直接杀掉、不做等待——适合可随时丢弃的心跳、监控任务。它被杀时不走 finally，绝不能持有需要清理的资源（文件、连接）；要优雅退出就用 Event 协作取消。

共享状态与竞态

线程共享进程内存，对同一对象的读改写交错执行即产生竞态条件（race condition）。n += 1 是读-改-写三步，两线程交错就丢更新：

n = 0
 
def inc():
    global n
    for _ in range(100_000):
        n += 1  # 非原子：两线程并发跑inc再join，结果几乎必然<200000

GIL 不救场：它只保证单条字节码原子，n += 1 编译为多条字节码，切换可发生在中间。共享对象的可变性陷阱另见引用与可变性。

互斥：Lock 与 RLock

lock = threading.Lock()
 
def inc():
    global n
    for _ in range(100_000):
        with lock:  # 等价acquire/release，异常路径也保证释放
            n += 1

with lock: 是标准惯用法，把「忘记 release」从隐患变成结构上不可能——与 Python with 语句与上下文管理器同一逻辑。锁的纪律：

临界区只包必要语句，锁内不做阻塞 IO
Lock 不可重入，同线程二次 acquire 即死锁；嵌套 / 递归获取用 RLock
多把锁按固定顺序获取，否则死锁

同步原语

原语	语义	典型场景
`Lock` / `RLock`	互斥访问	保护共享数据
`Semaphore(value)`	计数器，最多 N 线程同时进入	限流：连接池、并发爬虫
`Event`	布尔标志 + 唤醒所有等待者	一对多通知：启动 / 停止信号
`Condition`	Lock + 等待 / 通知	手写生产消费等待语义
`queue.Queue`	线程安全队列	生产者-消费者（首选）

Event：一对多信号

import threading
 
start_event = threading.Event()
 
def worker(name):
    start_event.wait()   # 阻塞直到信号置位
    print(f"{name} 启动")
 
threads = [threading.Thread(target=worker, args=(i,)) for i in range(5)]
for t in threads:
    t.start()            # 线程全部就位，都在等信号
start_event.set()        # 一声令下，全部同时放行

set() 唤醒所有等待线程，clear() 复位后可重复发令——控制一组线程统一起跑 / 统一停止就用它。

协作取消惯用法：daemon 线程里 while not stop.wait(1.0) 循环——Event.wait(timeout) 兼任可中断 sleep，stop.set() 后立即返回 True 退出循环，比裸 time.sleep 能即时响应停止信号。

Condition：等待条件满足

import threading
 
condition = threading.Condition()
current = 0
 
def worker(name):
    global current
    with condition:
        while name != current:    # 被唤醒后必须重查条件，防虚假唤醒
            condition.wait()
        print(f"{name} 按序执行")
        current += 1
        condition.notify_all()    # 唤醒其他线程重查各自条件
 
threads = [threading.Thread(target=worker, args=(i,)) for i in range(5)]
for t in threads:
    t.start()
for t in threads:
    t.join()  # 无论启动顺序如何，输出严格按 0,1,2,3,4

wait() 释放锁并挂起、被唤醒后重新竞争锁，是 Lock 做不到的「等待时不挡人」；上例让多个线程严格按固定次序执行。生产者-消费者等待语义优先用 queue.Queue，手写 Condition 只用于 Queue 表达不了的顺序约束。

队列：生产者-消费者

queue.Queue 自带锁与条件变量，线程安全且自带「满则阻塞写、空则阻塞读」的背压——手写 Condition 同步不如直接用它：

import queue
 
q = queue.Queue(maxsize=100)  # 有界队列提供背压
 
def producer():
    for i in range(5):
        q.put(i)    # 满则阻塞
    q.put(None)     # 毒丸：通知消费者结束
 
def consumer():
    while (item := q.get()) is not None:
        print(item)

完成跟踪的另一路：消费者每取走一项调 q.task_done()，主线程 q.join() 阻塞到队列所有项处理完毕——与毒丸二选一。

线程池：ThreadPoolExecutor

concurrent.futures.ThreadPoolExecutor 负责线程复用，submit 返回 Future 供取结果与异常，是绝大多数 I/O 并发的正确入口：

from concurrent.futures import ThreadPoolExecutor, as_completed
import urllib.request
 
urls = ["https://example.com"] * 5
with ThreadPoolExecutor(max_workers=8) as pool:
    futs = {pool.submit(urllib.request.urlopen, u): u for u in urls}
    for fut in as_completed(futs):
        resp = fut.result()  # 阻塞取结果；任务异常在此重抛

选型

维度	threading	multiprocessing 多进程	asyncio
并行计算	否（GIL）	是（多解释器）	否
I/O 密集	适用	可用但重	高并发最优
共享状态	天然共享，需加锁	需 pickle / IPC	单线程，无锁
切换成本	中（内核线程）	高（进程创建）	低（协程）

经验法则：几十个以内 I/O 并发、调用的是阻塞库 → threading；上万连接 → asyncio；CPU 密集 → multiprocessing 或 C 扩展。

知识花园

探索