Python thread

import threading
import time
import random
from queue import Queue
from concurrent.futures import ThreadPoolExecutor
import sys


# 1. 线程创建方式1：继承Thread类并重写run方法
class MyThread(threading.Thread):
    def __init__(self, name, delay):
        super().__init__(name=name)  # 设置线程名称
        self.delay = delay

    def run(self):
        """线程执行的核心逻辑"""
        print(f"[{self.name}] 开始执行 (继承Thread类)")
        time.sleep(self.delay)
        print(f"[{self.name}] 执行结束")


# 2. 线程创建方式2：通过target函数创建
def thread_func(name, delay):
    """作为线程目标的函数"""
    print(f"[{name}] 开始执行 (target函数)")
    try:
        time.sleep(delay)
        # 模拟可能发生的异常
        if random.random() < 0.3:
            raise ValueError(f"[{name}] 发生随机错误")
    except Exception as e:
        print(f"[{name}] 捕获异常: {e}", file=sys.stderr)
    finally:
        print(f"[{name}] 函数执行完毕")


# 3. 线程同步：使用Lock处理共享资源竞争
class SharedResource:
    def __init__(self):
        self.count = 0
        self.lock = threading.Lock()  # 互斥锁
        # self.lock = threading.RLock()  # 可重入锁（适合递归场景）

    def increment(self, thread_name):
        """安全地递增计数器（演示锁的使用）"""
        with self.lock:  # 自动获取和释放锁，避免死锁风险
            print(f"[{thread_name}] 准备修改计数器，当前值: {self.count}")
            current = self.count
            time.sleep(0.1)  # 模拟耗时操作，放大竞争问题
            self.count = current + 1
            print(f"[{thread_name}] 修改后计数器: {self.count}")


# 4. 线程通信：使用Event实现简单信号传递
def event_waiter(event, name):
    """等待事件触发的线程"""
    print(f"[{name}] 等待事件触发...")
    event.wait()  # 阻塞等待事件被设置
    print(f"[{name}] 事件已触发，继续执行")


# 5. 线程通信：使用Condition实现复杂条件等待
def condition_worker(cond, name, threshold):
    """基于Condition等待特定条件的线程"""
    with cond:
        print(f"[{name}] 等待条件满足...")
        cond.wait_for(lambda: shared_counter >= threshold)  # 等待条件成立
        print(f"[{name}] 条件满足 (shared_counter={shared_counter})，开始工作")
        time.sleep(0.5)


# 6. 线程通信：使用Queue实现安全的数据传递
def queue_producer(q, name):
    """向队列生产数据的线程"""
    for i in range(3):
        data = f"{name}_data_{i}"
        q.put(data)  # 放入数据，队列满时会阻塞
        print(f"[{name}] 生产数据: {data}，队列大小: {q.qsize()}")
        time.sleep(random.uniform(0.1, 0.5))
    q.put(None)  # 发送结束信号
    print(f"[{name}] 生产结束")


def queue_consumer(q, name):
    """从队列消费数据的线程"""
    while True:
        data = q.get()  # 获取数据，队列为空时会阻塞
        if data is None:  # 收到结束信号
            q.put(None)  # 转发结束信号给其他消费者
            break
        print(f"[{name}] 消费数据: {data}，剩余: {q.qsize()}")
        time.sleep(random.uniform(0.2, 0.6))
        q.task_done()  # 标记任务完成
    print(f"[{name}] 消费结束")


# 7. 线程本地存储：每个线程的私有数据
thread_local = threading.local()  # 线程本地存储对象


def local_data_worker(name):
    """使用线程本地存储的线程"""
    thread_local.value = name  # 为当前线程设置私有值
    time.sleep(random.uniform(0.1, 0.3))
    # 每个线程只能访问自己的value
    print(f"[{name}] 线程本地值: {thread_local.value}")


# 8. 线程池：使用ThreadPoolExecutor管理线程
def pool_task(task_id):
    """线程池中的任务函数"""
    start = time.time()
    time.sleep(random.uniform(0.2, 0.8))
    end = time.time()
    return f"任务{task_id}完成，耗时: {end - start:.2f}s，线程: {threading.current_thread().name}"


def main():
    global shared_counter
    shared_counter = 0

    print("===== 1. 基本线程创建与启动 =====")
    # 创建线程实例
    thread1 = MyThread("继承线程", 1)
    thread2 = threading.Thread(
        target=thread_func,
        args=("目标线程", 1.5),
        daemon=False  # 非守护线程（默认）：主线程会等待其完成
    )

    # 启动线程
    thread1.start()
    thread2.start()

    # 等待线程完成（join的使用）
    thread1.join()
    thread2.join()
    print("基本线程演示结束\n")

    print("===== 2. 线程同步（锁机制） =====")
    resource = SharedResource()
    sync_threads = []
    for i in range(5):
        t = threading.Thread(
            target=resource.increment,
            args=(f"同步线程{i}",)
        )
        sync_threads.append(t)
        t.start()

    # 等待所有同步线程完成
    for t in sync_threads:
        t.join()
    print(f"最终计数器值: {resource.count}")
    print("线程同步演示结束\n")

    print("===== 3. 线程通信（Event） =====")
    event = threading.Event()
    waiter1 = threading.Thread(target=event_waiter, args=(event, "等待线程1"))
    waiter2 = threading.Thread(target=event_waiter, args=(event, "等待线程2"))

    waiter1.start()
    waiter2.start()
    time.sleep(2)  # 主线程延迟后触发事件
    print("主线程触发事件")
    event.set()  # 设置事件，唤醒所有等待线程

    waiter1.join()
    waiter2.join()
    print("Event通信演示结束\n")

    print("===== 4. 线程通信（Condition） =====")
    cond = threading.Condition()
    condition_threads = [
        threading.Thread(target=condition_worker, args=(cond, f"条件线程{i}", i + 1))
        for i in range(3)
    ]

    for t in condition_threads:
        t.start()

    # 主线程修改共享变量并通知等待线程
    time.sleep(1)
    with cond:
        shared_counter = 2
        print(f"主线程更新shared_counter={shared_counter}，通知所有等待线程")
        cond.notify_all()  # 唤醒所有等待线程

    for t in condition_threads:
        t.join()
    print("Condition通信演示结束\n")

    print("===== 5. 线程通信（Queue） =====")
    q = Queue(maxsize=2)  # 最大容量为2的队列
    producer = threading.Thread(target=queue_producer, args=(q, "生产者"))
    consumer1 = threading.Thread(target=queue_consumer, args=(q, "消费者1"))
    consumer2 = threading.Thread(target=queue_consumer, args=(q, "消费者2"))

    producer.start()
    consumer1.start()
    consumer2.start()

    producer.join()
    consumer1.join()
    consumer2.join()
    print("Queue通信演示结束\n")

    print("===== 6. 线程本地存储 =====")
    local_threads = [
        threading.Thread(target=local_data_worker, args=(f"本地线程{i}",))
        for i in range(3)
    ]

    for t in local_threads:
        t.start()
    for t in local_threads:
        t.join()
    print("线程本地存储演示结束\n")

    print("===== 7. 线程池（ThreadPoolExecutor） =====")
    with ThreadPoolExecutor(max_workers=3, thread_name_prefix="池线程") as executor:
        # 提交单个任务
        future = executor.submit(pool_task, 0)
        print(future.result())  # 获取单个任务结果

        # 批量提交任务
        tasks = list(range(1, 5))
        results = executor.map(pool_task, tasks)  # 按顺序返回结果
        for res in results:
            print(res)

    print("线程池演示结束\n")

    print("===== 8. 守护线程演示 =====")

    def daemon_worker():
        while True:
            print("守护线程运行中...")
            time.sleep(0.5)

    daemon_thread = threading.Thread(target=daemon_worker, daemon=True)
    daemon_thread.start()
    print("主线程休眠2秒后结束（守护线程会被强制终止）")
    time.sleep(2)
    print("主线程结束，守护线程将随之终止")


if __name__ == "__main__":
    main()

1. 线程创建方式
   • 继承threading.Thread类并重写run()方法
   • 直接传入目标函数target创建线程
   • 线程命名与参数传递
2. 线程生命周期管理
   • start()：启动线程
   • join()：等待线程完成
   • 守护线程（daemon=True）：随主线程结束而终止
3. 线程同步（解决资源竞争）
   • Lock：基础互斥锁，确保共享资源原子操作
   • RLock：可重入锁（适合递归场景）
   • with语句自动管理锁的获取与释放
4. 线程间通信
   • Event：简单信号传递（触发 / 等待机制）
   • Condition：复杂条件等待（基于谓词的唤醒）
   • Queue：线程安全的消息队列（生产者 - 消费者模型）
5. 线程私有数据
   • threading.local()：存储线程独有数据，避免共享冲突
6. 线程池
   • ThreadPoolExecutor：高效管理线程资源
   • submit()：提交单个任务
   • map()：批量处理任务并获取结果
7. 异常处理
   • 线程内部的异常捕获与处理
   • 错误信息的正确输出
8. 线程属性
   • 线程名称（name）
   • 线程标识（ident）
   • 当前线程获取（threading.current_thread()）

重写 run

1. 重写run方法的作用

Thread类是 Python 线程的基类，它内部有一个run方法，这个方法是线程的 “入口点”—— 当线程启动时，本质上就是执行run方法里的代码。

但Thread类自带的run方法是一个空实现（没有实际逻辑），因此需要通过继承Thread类并重写run方法，把我们需要线程执行的任务（比如循环、计算、IO 操作等）写在重写的run方法里。

例如代码中的MyThread类：

class MyThread(threading.Thread):
    def __init__(self, name, delay):
        super().__init__(name=name)
        self.delay = delay

    # 重写run方法，定义线程要执行的任务
    def run(self):
        print(f"[{self.name}] 开始执行 (继承Thread类)")
        time.sleep(self.delay)  # 模拟任务执行
        print(f"[{self.name}] 执行结束")

这里重写的run方法明确了线程的任务：打印开始信息 → 休眠指定时间 → 打印结束信息。

2. run方法会被自动调用（无需手动调用）

重写后的run方法不需要我们手动调用，而是通过线程的start()方法间接触发：

# 创建线程实例
thread1 = MyThread("继承线程", 1)
# 启动线程：调用start()，会自动执行重写的run()方法
thread1.start()   #run 是 threading.Thread 类中定义的 “线程执行逻辑的核心方法”—— 当调用线程的 start() 方法时，最终会执行 run 方法来完成线程的 “业务活动”。

• 当调用start()时，Python 会创建一个新的线程，并在新线程中自动执行run方法里的逻辑。
• 注意：如果直接调用run()（如thread1.run()），不会创建新线程，只是普通的函数调用（在当前线程中执行），这就失去了 “多线程” 的意义。

总结

• 重写run方法：目的是给线程 “分配任务”，定义线程要执行的具体逻辑。
• 调用方式：通过start()启动线程，start()会自动在新线程中调用重写后的run方法（无需手动调用run）

为什么是thread1.start()？

start() 方法的核心作用：启动新线程并触发 run()

threading.Thread 类的 start() 方法不是简单地 “调用 run()”，而是做了两件关键事：

1. 向操作系统申请创建一个新的线程（底层调用操作系统的线程 API，如 Linux 的pthread_create）；
2. 在新线程中自动调用 run() 方法（确保 run() 中的逻辑在独立线程中执行）

类比c++ thread

在 Python 中 “继承 Thread 类并重写 run 方法”，对应 C++11 std::thread 的 “函数对象（Functor）” 用法（也叫 “仿函数” 或 “可调用对象”）。以下是详细对比和解释：

1. 核心逻辑：“类封装执行逻辑”

Python 中重写 run 的本质是将线程的执行逻辑封装在类的成员函数中；C++11 中 “函数对象” 的本质是将线程的执行逻辑封装在类的 operator() 重载方法中。

2. 代码结构对比

Python 侧（继承 Thread 并重写 run）

import threading
class MyThread(threading.Thread):
    def __init__(self, name, delay):
        super().__init__(name=name)
        self.delay = delay
    def run(self):  # 重写 run 方法，定义线程任务
        print(f"[{self.name}] 执行任务...")
        time.sleep(self.delay)

# 使用：创建线程并启动
t = MyThread("线程A", 1)
t.start()  # 自动调用重写的 run 方法

C++11 侧（函数对象 / Functor）

#include <thread>
#include <iostream>
using namespace std;

class MyFunctor {
public:
    MyFunctor(const string& name, int delay) 
        : name_(name), delay_(delay) {}
    
    void operator()() {  // 重载 operator()，定义线程任务
        cout << "[" << name_ << "] 执行任务..." << endl;
        this_thread::sleep_for(chrono::seconds(delay_));
    }
private:
    string name_;
    int delay_;
};

// 使用：创建函数对象并传入 std::thread
int main() {
    MyFunctor func("线程A", 1);
    thread t(func);  // 将函数对象传入 thread，自动调用 operator()
    t.join();
    return 0;
}

3. 行为与设计意图的一致性

• 封装性：两者都通过 “类” 封装线程的状态（如 Python 的 self.delay、C++ 的 delay_）**和**执行逻辑（run 或 operator()），适合复杂场景（如需要维护线程私有的成员变量）。
• 自动调用：Python 中 start() 会自动调用 run；C++ 中 std::thread 会自动调用函数对象的 operator()，无需手动调用。

4. 其他 C++11 线程用法对比

C++11 std::thread 还支持函数指针、Lambda、成员函数等方式，但这些与 Python “重写 run” 的设计思路不同：

• 函数指针 / Lambda：逻辑是 “无状态” 的（或依赖外部变量捕获），无法像类一样自然封装线程私有的状态。
• 成员函数：需额外传递对象指针（如 thread(&Class::func, &obj)），更适合 “复用已有类的方法”，而非 “为线程定制逻辑”

thread 使用

在 start_thread 函数中，args 这种传参方式是 Python 多线程中传递 “位置参数” 的标准写法，核心作用是：将线程目标函数（如 log_writer、data_cleaner）需要的参数 “打包” 成一个元组（tuple），传递给线程，线程执行时会自动将元组中的元素 “解包” 为目标函数的独立位置参数

二、结合代码实例详解

以你的三个线程为例，逐个分析 args 传参的对应关系：

1. 日志线程：log_thread = start_thread(log_writer, args=(log_queue,))

• 目标函数 log_writer 的定义：def log_writer(log_queue):（需要 1 个位置参数 log_queue）。

• args=(log_queue,) 是一个单元素元组（注意末尾的逗号 ,，否则会被视为普通变量而非元组）。

• 线程执行时，会自动将元组解包，等价于调用 log_writer(log_queue)（参数正确传递）。

2. 数据清洗线程：cleaner_thread = start_thread(data_cleaner, args=(raw_data,))

• 目标函数 data_cleaner 的定义：def data_cleaner(raw_data_list):（需要 1 个位置参数 raw_data_list）。

• args=(raw_data,) 中，raw_data 是提前定义的列表（raw_data = [" hello ", "", "world ", "python "]）。

• 线程执行时，等价于调用 data_cleaner(raw_data)（参数正确传递）。

3. 告警线程：alert_thread = start_thread(alert_sender, args=(alert_threshold,))

• 目标函数 alert_sender 的定义：def alert_sender(alert_threshold):（需要 1 个位置参数 alert_threshold）。
• args=(alert_threshold,) 中，alert_threshold 是提前定义的整数（alert_threshold = 5）。
• 线程执行时，等价于调用 alert_sender(alert_threshold)（参数正确传递

multiple_tasks.py · beihangya/python - 码云 - 开源中国

threading.Event

event_set_and_clear.py · beihangya/python - 码云 - 开源中国

threading.Event 是 Python 多线程编程中用于线程间同步的核心工具

threading.Event 解决的是 “线程间如何等待某个事件发生” 的问题，例如：

• 主线程等待所有工作线程初始化完成后再继续执行；

• 后台线程等待 “停止信号” 以优雅退出；

• 多个线程等待某个资源准备就绪后再同时开始工作

• threading.Event 是「线程间同步的信号工具」，用于控制线程的启停（比如让线程等待信号、接收停止指令），本身不占用线程资源，也不是线程。

• 真正的线程是通过 threading.Thread 创建的实例（即 t = threading.Thread(...)）

set

threading.Event 类是 Python 多线程同步的核心工具，其核心功能依赖于一个内部布尔标志位（默认值为 False）。set() 和 clear() 方法是控制这个标志位的核心接口，直接决定了线程的 “阻塞” 与 “唤醒” 状态

set() 方法：发送 “事件触发” 信号 :

set ，将 Event 内部的标志位设置为 True，并唤醒所有正在通过 wait() 方法等待该事件的线程。

详细行为：

1. 标志位变更：调用 set() 后，Event 的内部标志位从 False 变为 True（此状态会一直保持，直到被 clear() 重置）。
2. 唤醒等待线程：所有因调用 event.wait() 而处于阻塞状态的线程，会被立即唤醒并继续执行（wait() 方法返回 True）。
3. 对后续 wait() 的影响：如果标志位已为 True（即 set() 已调用且未被 clear()），此时新调用 event.wait() 的线程不会阻塞，会立即返回 True（因为标志位已是触发状态）

clear

clear() 方法：清除 “事件触发” 信号

`

clear , 将 Event 内部的标志位重置为 False，只会让 “后续调用 wait() 方法的线程” 进入阻塞状态（直到再次调用 set()）

对于已经被 set() 唤醒、正在执行的线程，clear() 完全没有影响（不会让它们重新阻塞）

详细行为：

1. 标志位变更：调用 clear() 后，Event 的内部标志位从 True 变回 False（此状态会一直保持，直到被 set() 再次触发）。

2. 对后续 wait() 的影响 ,标志位为 False 时，新调用 event.wait() 的线程会进入阻塞状态，暂停执行，直到：

   • 其他线程调用 set()（标志位变为 True），此时阻塞的线程被唤醒；

   • 若 wait(timeout=xxx) 设置了超时时间，超时后线程会自动唤醒（返回 False）。

3. 不影响已唤醒的线程：clear() 只影响之后调用 wait() 的线程，对已经被 set() 唤醒并正在执行的线程无影响。

wait

让线程进入 “阻塞状态”，暂时让出 CPU 资源，直到以下两种情况之一发生才会被唤醒：

1. 其他线程通过 notify()/notify_all() 方法发送 “唤醒信号”；
2. 等待超时（若指定了 timeout 参数）

import threading
import time

# 创建一个事件（初始为未设置状态）
event = threading.Event()

def worker():
    print("线程开始等待事件...")
    # 调用 wait() 阻塞，直到 event 被 set() 或超时
    # timeout=5 表示最多等5秒，超时后返回 False
    success = event.wait(timeout=5)
    if success:
        print("事件已触发，线程继续执行！")
    else:
        print("等待超时，线程继续执行！")

# 启动工作线程
t = threading.Thread(target=worker)
t.start()

# 主线程延迟3秒后触发事件
time.sleep(3)
event.set()  # 设置事件，唤醒等待的线程

重写Event ,增加特定功能

class MyEvent(threading.Event):  # ← 继承threading.Event
    def __init__(self):
        super(MyEvent, self).__init__()  # 调用父类初始化
        self.data = None  # ← 新增data属性

    def clear(self):  # ← 重写clear方法
        """清除事件信号和数据"""
        super(MyEvent, self).clear()  # 调用父类clear
        self.data = None  # ← 额外清除数据

    def set(self, data=None):  # ← 重写set方法
        """设置事件信号并携带数据"""
        self.data = data  # ← 保存数据
        super(MyEvent, self).set()  # 调用父类set

        threading.Event
              ↑
              │ 继承
              │
          MyEvent
              │
    ┌─────────┴─────────┐
    │                    │
重写set()         重写clear()
新增data属性      清除data属性

自定义 MyEvent：

event = MyEvent()
event.set(data={'pic': img, 'barcode': 'ABC123'})  # 设置信号+数据
result = event.wait()                               # 等待信号
data = event.data                                   # ✅ 获取携带的数据

为了在线程间传递数据和信号，定义了自定义事件类：

MyEvent.py · beihangya/python - 码云 - 开源中国