一、线程基础概念

1.1 什么是线程

线程（Thread）是程序中的独立执行流，允许程序同时执行多个任务。Python 的 threading 模块提供了跨平台一致的线程管理接口。

1.2 创建线程的两种方式

方式一：直接传入可调用对象（推荐）

import threading

def worker(num):
    print(f"Worker {num}")

thread = threading.Thread(target=worker, args=(1,))
thread.start()
thread.join()

方式二：继承 Thread 类

class MyThread(threading.Thread):
    def run(self):
        print("Thread running")

thread = MyThread()
thread.start()

1.3 守护线程（Daemon Thread）

守护线程是一种后台运行的线程，当程序中只剩下守护线程时，Python 程序会自动退出。

# 设置守护线程
thread = threading.Thread(target=func, daemon=True)
thread.start()

特点：

• 主程序退出时自动终止
• 适合后台任务、定时检查、心跳监测等场景
• 退出时资源可能未正确释放

---

二、数据竞争（Race Condition）

2.1 什么是竞态条件

当多个线程同时访问和修改共享资源时，最终结果依赖于线程执行的顺序，导致数据不一致。

2.2 实际案例

import threading

counter = 0

def add():
    global counter
    for _ in range(100000):
        counter += 1  # 非原子操作，存在竞态条件

threads = [threading.Thread(target=add) for _ in range(2)]
for t in threads:
    t.start()
for t in threads:
    t.join()
print(counter)  # 结果可能小于 200000

---

三、锁（Lock）机制

3.1 锁的基本概念

锁是一种同步原语，用于保护共享资源，确保同一时间只有一个线程可以访问该资源。

锁的两种状态：

• 锁定（locked）：线程正在使用共享资源
• 未锁定（unlocked）：资源空闲，可被线程获取

3.2 锁的核心原理

互斥锁（Mutual Exclusion Lock）确保在同一时刻只有一个线程可以进入临界区（Critical Section）——即访问共享资源的代码段。

3.3 锁的使用方法

方法一：with 语句（推荐）

lock = threading.Lock()
counter = 0

def safe_add():
    global counter
    with lock:  # 自动 acquire() 和 release()
        counter += 1

方法二：手动 acquire() / release()

lock = threading.Lock()
lock.acquire()
try:
    counter += 1  # 临界区代码
finally:
    lock.release()  # 确保锁释放

推荐使用 with 语句，因为更安全、更简洁，不容易忘记释放锁。

3.4 锁的阻塞机制

• 当锁处于”未锁定”状态：调用 acquire() 的线程会立即获得锁
• 当锁处于”锁定”状态：其他线程调用 acquire() 时会被阻塞等待，直到持有锁的线程释放锁

3.5 非阻塞获取锁

if lock.acquire(blocking=False):  # 不阻塞，立即返回
    try:
        pass
    finally:
        lock.release()

---

四、锁的注意事项与最佳实践

4.1 避免死锁

死锁通常由以下原因导致：

• 多把锁交叉获取
• 忘记释放锁
• 嵌套锁顺序不一致

预防措施：

• 固定加锁顺序
• 使用 with 语句
• 使用带超时的 acquire(timeout)
• 优先使用可重入锁（RLock）

4.2 最小化锁的持有时间

# 不好的写法：持有锁时间过长
with lock:
    do_something()
    do_other_thing()
    time.sleep(2)  # 不必在锁内

# 更好的写法
with lock:
    update_shared_data()
time.sleep(2)  # 锁外执行

4.3 锁的代价

• 上下文切换开销
• 等待开销
• 代码复杂度上升
• 死锁风险

---

五、可重入锁（RLock）

5.1 RLock 的特点

可重入锁（Reentrant Lock）允许同一线程多次获取锁，内部维护计数器。

核心特性：

• 同一线程 acquire() 多少次，需 release() 多少次
• 计数器归 0 时才真正释放锁
• 仅当前持有锁的线程可释放锁

5.2 使用场景

适用于递归函数或同一线程需多次进入临界区的场景：

lock = threading.RLock()

def recursive_function():
    with lock:
        print("进入临界区")
        recursive_function()  # 同一线程可以再次获取锁

---

六、其他同步机制

6.1 信号量（Semaphore）

用于控制同时访问共享资源的线程数量：

semaphore = threading.Semaphore(2)  # 允许最多2个线程同时访问

def access_resource():
    with semaphore:
        print(f"Thread {threading.current_thread().name} accessing resource")

6.2 条件变量（Condition）

用于线程之间的通信和协调，常用于生产者-消费者模型：

condition = threading.Condition()
queue = []

def producer():
    with condition:
        queue.append("item")
        condition.notify()  # 通知等待的线程

def consumer():
    with condition:
        while not queue:
            condition.wait()  # 等待队列非空
        item = queue.pop(0)

6.3 事件（Event）

允许线程等待一个事件发生：

event = threading.Event()

def worker():
    print("Waiting for event...")
    event.wait()  # 等待事件被设置
    print("Event fired!")

# 在主线程中触发事件
event.set()

6.4 队列（Queue）

线程安全的队列，官方推荐优先使用：

import queue

q = queue.Queue(maxsize=10)

def producer():
    q.put("item")

def consumer():
    item = q.get()  # 自动处理锁与条件变量

优势：

• 内部实现线程安全
• 支持阻塞与超时
• 简化同步逻辑
• 可扩展性强

---

七、GIL（全局解释器锁）的影响

7.1 GIL 的作用

Python 的全局解释器锁（GIL）确保同一时刻只有一个线程执行 Python 字节码。

7.2 GIL 的影响

• CPU 密集型任务：多线程无法利用多核优势，性能提升有限甚至下降，建议使用 multiprocessing 模块
• I/O 密集型任务：线程在等待 I/O 时会释放 GIL，多线程仍能有效提升并发性能

---

八、@dataclass 装饰器

8.1 基本概念

@dataclass 是 Python 3.7 引入的装饰器，位于标准库 dataclasses 模块中，用于自动为类生成常用的特殊方法。

8.2 自动生成的方法

• __init__()：按字段顺序初始化
• __repr__()：提供包含类名和所有字段值的清晰表示
• __eq__()：基于字段值进行相等性比较

8.3 使用示例

from dataclasses import dataclass

@dataclass
class Task:
    id: str
    subject: str
    description: str
    status: str          # pending | in_progress | completed
    owner: str | None    # Agent name
    blockedBy: list[str]  # Dependency task IDs

8.4 高级用法

@dataclass
class Task:
    id: str
    status: str = "pending"

    def __post_init__(self):
        # 初始化后自动验证
        if self.status not in ["pending", "in_progress", "completed"]:
            raise ValueError(f"Invalid status: {self.status}")

---

九、JSON 与字典解包

9.1 读取 JSON 文件

import json
from pathlib import Path

def load_task(task_id: str) -> Task:
    return Task(**json.loads(_task_path(task_id).read_text()))

9.2 ** 字典解包操作符

• 将字典的键值对解包为函数的关键字参数
• 字典的键必须与函数参数名匹配

data = {"name": "Alice", "age": 30}
# Person(**data) 等价于 Person(name="Alice", age=30)

十、常用字符串方法

方法	功能	返回值	默认行为
strip()	去除两端字符	字符串	去除空白字符
split()	按分隔符分割	列表	按空白字符分割
splitlines()	按换行符分割	列表	不保留换行符
startswith()	检查是否以指定前缀开头	布尔值	大小写敏感

---

总结

• 多线程编程的核心是正确处理共享资源的访问，避免竞态条件
• 锁（Lock） 是最基础的同步机制，推荐使用 with 语句确保安全
• 可重入锁（RLock） 解决同一线程多次加锁问题
• 高级同步机制（Semaphore、Condition、Event、Queue）适用于特定场景
• GIL 限制了 CPU 密集型任务的并行性能，但 I/O 密集型任务仍受益
• @dataclass 简化数据类的定义，自动生成样板代码
• 字典解包 ** 方便从 JSON 等数据格式直接创建对象

---