线程生命周期

线程从创建到销毁，经历以下状态：

常用线程方法

import threading
import time

def worker():
    time.sleep(1)

t = threading.Thread(target=worker)

print(t.is_alive())  # False（未启动）
t.start()
print(t.is_alive())  # True（运行中）
t.join()             # 阻塞主线程，等待 t 完成
print(t.is_alive())  # False（已结束）

# 获取当前活跃线程数
print(f"活跃线程数：{threading.active_count()}")
# 获取所有线程列表
print(f"所有线程：{threading.enumerate()}")

五、线程同步：Lock 和 RLock

5.1 为什么需要同步？—— 竞态条件

import threading

count = 0# 共享变量

def add():
    global count
    for _ in range(1000000):
        count += 1# ⚠️ 这不是原子操作！

t1 = threading.Thread(target=add)
t2 = threading.Thread(target=add)
t1.start()
t2.start()
t1.join()
t2.join()

print(f"期望值：2000000，实际值：{count}")
# 实际值往往小于 2000000！

💥 问题根因：count += 1 实际是三步：读取 → 加1 → 写回。两个线程可能同时读到相同的旧值。

5.2 使用 Lock 解决

import threading

count = 0
lock = threading.Lock()  # 创建互斥锁

def add():
    global count
    for _ in range(1000000):
        with lock:  # 自动 acquire() 和 release()
            count += 1

t1 = threading.Thread(target=add)
t2 = threading.Thread(target=add)
t1.start()
t2.start()
t1.join()
t2.join()

print(f"结果：{count}")  # 稳定输出 2000000 ✅

5.3 RLock（可重入锁）

import threading

rlock = threading.RLock()

def outer():
    with rlock:
        print("外层获得锁")
        inner()  # 同一线程可以再次获得锁

def inner():
    with rlock:
        print("内层获得锁")

outer()  # 如果用 Lock 会死锁，RLock 不会

🔑 Lock vs RLock：

• Lock：同一线程不能重复获取，否则死锁
• RLock：同一线程可多次获取，内部计数，全部释放才真正解锁

六、线程通信：Event、Condition、Queue

6.1 Event（事件通知）

一个线程发信号，其他线程等待。

import threading
import time

event = threading.Event()

def waiter():
    print("等待信号...")
    event.wait()  # 阻塞，直到 event 被 set
    print("收到信号，继续执行！")

def setter():
    time.sleep(2)
    print("发送信号！")
    event.set()  # 唤醒所有等待的线程

threading.Thread(target=waiter).start()
threading.Thread(target=setter).start()

6.2 Condition（条件变量）

更精细的控制：等待某个条件满足。

import threading
import time

condition = threading.Condition()
data_ready = False

def consumer():
    global data_ready
    with condition:
        whilenot data_ready:
            condition.wait()  # 释放锁并等待
        print("消费者：拿到数据了！")
        data_ready = False

def producer():
    global data_ready
    time.sleep(1)
    with condition:
        data_ready = True
        condition.notify()  # 唤醒一个等待者
        print("生产者：数据准备好了！")

threading.Thread(target=consumer).start()
threading.Thread(target=producer).start()

6.3 Queue（线程安全队列）⭐推荐

最常用、最安全的线程间通信方式。

import threading
import queue
import time

q = queue.Queue(maxsize=5)  # 最多存5个元素

def producer():
    for i in range(10):
        q.put(f"产品-{i}")  # 队列满时自动阻塞
        print(f"生产：产品-{i}")

def consumer():
    while True:
        item = q.get()  # 队列空时自动阻塞
        print(f"消费：{item}")
        q.task_done()  # 标记完成

threading.Thread(target=producer).start()
threading.Thread(target=consumer, daemon=True).start()

q.join()  # 等待所有任务完成
print("全部处理完毕！")

七、守护线程（Daemon）

💡 守护线程：当所有非守护线程结束时，守护线程会被强制终止，不管它有没有执行完。

import threading
import time

def background_task():
    while True:
        print("后台监控中...")
        time.sleep(1)

# 设置为守护线程
t = threading.Thread(target=background_task, daemon=True)
t.start()

time.sleep(3)
print("主线程结束，守护线程自动终止")
# 程序在此退出，不会无限循环

⚠️ 注意：daemon=True 必须在 start() 之前设置。

八、线程局部数据（Thread-Local）

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

import threading

local_data = threading.local()

def process(name):
    local_data.value = name  # 每个线程设置自己的 value
    print(f"[{name}] 设置 value = {name}")
    print(f"[{name}] 读取 value = {local_data.value}")  # 读到的是自己的

t1 = threading.Thread(target=process, args=("Alice",))
t2 = threading.Thread(target=process, args=("Bob",))
t1.start()
t2.start()
t1.join()
t2.join()

输出：

[Alice] 设置 value = Alice
[Bob] 设置 value = Bob
[Alice] 读取 value = Alice
[Bob] 读取 value = Bob

🔑 典型用途：Web 框架中为每个请求线程保存用户会话信息（如 Flask 的 g 对象）。

九、线程池（ThreadPoolExecutor）

手动管理线程很麻烦，线程池帮你自动管理。

场景：你一个人在家，有6个快递要寄出，但每次只能跑一趟驿站。如果自己去，要跑6趟，很累。

用线程池：相当于叫了3个快递员（max_workers=3）同时帮你跑，6个快递分两批就送完了。

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

# 模拟寄快递：每个快递需要不同时间打包
def send_package(package_id):
    cost = random.randint(1, 4)  # 打包耗时 1~4 秒
    print(f"📦 开始处理包裹 {package_id}（预计 {cost} 秒）")
    time.sleep(cost)
    result = f"包裹 {package_id} 已寄出"
    return result

# 有 6 个包裹要寄
packages = [101, 102, 103, 104, 105, 106]

# 创建线程池：3 个快递员同时干活
with ThreadPoolExecutor(max_workers=3) as executor:
    # 把所有包裹交给快递员
    futures = {executor.submit(send_package, pkg): pkg for pkg in packages}

    # 谁先完成就先取谁的结果
    for future in as_completed(futures):
        pkg = futures[future]
        try:
            result = future.result()
            print(f"✅ {result}")
        except Exception as e:
            print(f"❌ 包裹 {pkg} 出问题了: {e}")

输出效果（大致是这样）：

📦 开始处理包裹 101（预计 3 秒）
📦 开始处理包裹 102（预计 1 秒）
📦 开始处理包裹 103（预计 4 秒）
✅ 包裹 102 已寄出        ← 102 最快完成
📦 开始处理包裹 104（预计 2 秒）
✅ 包裹 101 已寄出
📦 开始处理包裹 105（预计 3 秒）
✅ 包裹 104 已寄出
📦 开始处理包裹 106（预计 2 秒）
✅ 包裹 106 已寄出
✅ 包裹 105 已寄出
✅ 包裹 103 已寄出

线程池的好处（对比自己手动搞线程）

一句话总结

线程池 = 一个“任务调度员”，你只管交任务，它自动分配线程、控制并发、收集结果、清理资源。

9.1 更简洁的写法：map

from concurrent.futures import ThreadPoolExecutor

def square(n):
    return n * n

with ThreadPoolExecutor(max_workers=4) as executor:
    results = list(executor.map(square, [1, 2, 3, 4, 5]))
    print(results)  # [1, 4, 9, 16, 25]

9.2 线程池参数建议

十、GIL：Python 多线程的"天花板"