多线程编程进阶

需要预先学习：

• 多线程编程基础
• 数据共享问题与锁

---

1. 线程间通信的问题

锁解决了"互斥"（同一时刻只有一个线程进临界区），但没解决"等待"：线程 A 要等线程 B 完成某件事（比如数据准备好了）才能继续，A 应该怎么等？

• 忙等：while (!ready) {} 死循环空转，浪费 CPU，还带来数据竞争
• 轮询：while (!ready) { sleep(1ms); } 浪费少些，但有延迟，且 ready 的读写仍要同步

这两种方案都不优雅。标准答案有两个：条件变量（等待"事件"）和 future/promise（等待"结果"）。本篇文章围绕它们展开，最后实现一个线程池。

---

2. 条件变量 std::condition_variable

条件变量（condition variable）让线程阻塞等待某个条件成立，由其他线程唤醒它。它必须配合 std::unique_lock（和一把互斥量）使用。

核心 API：

函数	行为
wait(lock)	阻塞，并自动解锁；被唤醒后重新加锁
wait(lock, pred)	带谓词版本：谓词为 false 才阻塞（见下文）
notify_one()	唤醒一个正在等待的线程
notify_all()	唤醒所有正在等待的线程

先看标准的使用模式——生产者消费者：

#include <condition_variable>
#include <iostream>
#include <mutex>
#include <queue>
#include <thread>

std::mutex mtx;
std::condition_variable cv;
std::queue<int> data_queue;    // 受 mtx 保护的共享队列
bool done = false;

void producer()
{
    for (int i = 0; i < 5; ++i)
    {
        {
            std::lock_guard<std::mutex> lock(mtx);
            data_queue.push(i);
            std::cout << "生产: " << i << std::endl;
        }
        cv.notify_one();       // 唤醒一个消费者
        std::this_thread::sleep_for(std::chrono::milliseconds(50));
    }
    {
        std::lock_guard<std::mutex> lock(mtx);
        done = true;
    }
    cv.notify_all();           // 通知所有消费者收工
}

void consumer()
{
    while (true)
    {
        std::unique_lock<std::mutex> lock(mtx);
        // 等待条件: 队列非空 或 生产结束
        cv.wait(lock, [&] { return !data_queue.empty() || done; });

        if (!data_queue.empty())
        {
            int value = data_queue.front();
            data_queue.pop();
            lock.unlock();     // 处理数据前先解锁, 让其他线程也能进临界区
            std::cout << "消费: " << value << std::endl;
        }
        else if (done)
        {
            break;             // 队列空且生产结束, 退出
        }
    }
}

int main()
{
    std::thread p(producer);
    std::thread c1(consumer);
    std::thread c2(consumer);
    p.join();
    c1.join();
    c2.join();
    return 0;
}

wait 的关键语义：调用 wait 时，条件变量原子地完成"解锁 + 睡眠"两步——线程带着锁进入睡眠，锁被释放，其他线程才能进入临界区修改条件；被唤醒后，wait 返回前重新加锁。这个"解锁与睡眠不可分割"的设计避免了竞态：否则可能出现"线程正要睡眠，生产者已经 notify，通知丢失，消费者永远睡下去"。

---

2.1 为什么用谓词重载的 wait

直接 wait(lock) 有两个问题：

1. 虚假唤醒（spurious wakeup）：wait 可能在没有任何线程 notify 的情况下返回（这是操作系统实现的现实，标准明确允许）。如果唤醒后不检查条件直接消费，会读到空队列。

2. 多消费者竞争：notify_all 唤醒所有消费者，但只有一个能抢到锁。抢到的消费完队列后，另一个被唤醒的线程醒来发现队列空了——它必须再检查一次条件。

wait(lock, pred) 等价于 while (!pred()) wait(lock)，循环检查条件，从根本上解决了上面两个问题：

cv.wait(lock, [&] { return !data_queue.empty(); });
// 等价于:
while (!(!data_queue.empty()))
{
    cv.wait(lock);
}

注：谓词版本内部就是"检查谓词 → 不满足才 wait"的循环，所以永远应该用谓词重载，裸 wait(lock) 只在极少数明确知道不会被虚假唤醒干扰的场景才用。

---

2.2 notify_one 与 notify_all

• notify_one()：唤醒一个等待线程。适合"一个事件只需一个线程处理"的场景（如队列来了一个任务）。唤醒哪个线程由调度器决定，不确定。
• notify_all()：唤醒所有等待线程。适合"广播"场景（如 done = true 让所有消费者退出）。

选择原则：能确定只有一个消费者关心时用 notify_one（开销小、竞争少）；需要所有线程都醒来检查时用 notify_all。

注意：notify 不要求持锁——先释放锁再 notify 是更好的实践（如上例生产者）。如果持锁 notify，被唤醒的线程立即尝试加锁却拿不到，白白多一次阻塞-唤醒往返。

---

2.3 condition_variable_any

std::condition_variable_any 可以与任何满足 BasicLockable 要求的锁配合（包括 std::shared_lock、自定义锁），而普通 condition_variable 只能用 std::unique_lock<std::mutex>。代价是性能略低（内部多做一层抽象），一般用不上。

---

3. 异步任务：async、future、promise

条件变量适合"等待事件"，而"等待一个计算结果"用 future/promise 模型更直接。三个组件：

• std::async：启动一个异步任务，返回 std::future
• std::future：代表"未来的结果"，用 get() 阻塞等待结果
• std::promise：手动设置结果，通过 get_future() 把结果传给 future

---

3.1 std::async：最简单的异步调用

#include <future>
#include <iostream>

int slow_square(int n)
{
    std::this_thread::sleep_for(std::chrono::milliseconds(100));
    return n * n;
}

int main()
{
    // 启动异步任务, 立即返回 future, 不阻塞
    std::future<int> fut = std::async(std::launch::async, slow_square, 12);

    std::cout << "异步任务已启动, 主线程可以干别的事" << std::endl;

    int result = fut.get();    // 阻塞直到结果就绪
    std::cout << "结果 = " << result << std::endl;   // 144
    return 0;
}

std::async 的三个要点：

1. 返回值是 std::future<T>，T 是任务函数的返回类型
2. fut.get() 只能调用一次，第二次调用抛 std::future_error；wait() 只等待不取结果
3. 任务函数抛出的异常会被捕获并重新抛出在 get() 处——异步任务的错误处理和同步调用一样自然

---

3.2 启动策略 launch::async 与 launch::deferred

策略	行为
std::launch::async	立即在新线程中执行
std::launch::deferred	懒执行：直到 get()/wait() 才在当前线程执行
std::launch::async | deferred	实现自行选择（默认）

auto f1 = std::async(std::launch::async, task);       // 必定新线程, 立即执行
auto f2 = std::async(std::launch::deferred, task);    // 惰性, get() 时当前线程执行
auto f3 = std::async(task);                           // 实现决定, 通常是 async

陷阱：默认策略下，如果任务可能不被 get()，它可能根本不会执行（deferred）。而 std::async 返回的 future 析构时会阻塞等待任务完成（当任务是 launch::async 时）——fut.get() 没调用但 future 析构了，程序会等到任务结束。这是标准规定的行为，初看反直觉：

void task() { std::this_thread::sleep_for(std::chrono::seconds(1)); }

int main()
{
    std::async(std::launch::async, task);   // 临时 future 析构时阻塞 1 秒!
    std::cout << "这句话 1 秒后才打印" << std::endl;
    return 0;
}

所以始终把 async 的返回值存下来，并且明确选择策略，不要依赖默认行为。

---

3.3 std::promise：手动设置结果

std::promise<T> 是 future 的"写入端"：创建 promise 后用 get_future() 得到配套的 future，把 promise 传进线程（用 std::move，见《线程传参》），线程算完后 set_value() 或 set_exception()：

#include <future>
#include <iostream>
#include <thread>

void compute(std::promise<int> prom, int n)
{
    try
    {
        if (n < 0)
        {
            throw std::invalid_argument("负数没有平方根");
        }
        prom.set_value(n * n);              // 设置结果
    }
    catch (...)
    {
        prom.set_exception(std::current_exception());   // 或设置异常
    }
}

int main()
{
    std::promise<int> prom;
    std::future<int> fut = prom.get_future();

    std::thread t(compute, std::move(prom), 7);   // promise 不可拷贝, 移动进线程
    t.join();

    try
    {
        std::cout << "结果 = " << fut.get() << std::endl;   // 49
    }
    catch (const std::exception& e)
    {
        std::cout << "异常: " << e.what() << std::endl;
    }
    return 0;
}

关键语义：

• promise 是 move-only 类型，必须 std::move 进线程
• set_value 后 promise 进入就绪态，future 的 get() 不再阻塞
• promise 析构时若尚未 set_value，future 会收到 std::future_error（broken promise）——注意这个语义，它提示你"承诺的结果永远不会来了"
• future 可以 wait_for(duration) / wait_until(time_point) 限时等待，超时返回 std::future_status::timeout

if (fut.wait_for(std::chrono::milliseconds(100)) == std::future_status::timeout)
{
    std::cout << "100ms 内没算完, 先做别的事" << std::endl;
}

---

3.4 std::packaged_task：打包任务

std::packaged_task 把可调用对象和 future 打包：调用 packaged_task 的 operator() 时执行任务并把结果写入内部的 promise。适合"任务本身要排队/延迟执行，但调用方想拿结果"的场景：

#include <future>
#include <iostream>
#include <thread>

int add(int a, int b) { return a + b; }

int main()
{
    std::packaged_task<int(int, int)> task(add);   // 打包任务
    std::future<int> fut = task.get_future();      // 先拿到结果的 future

    std::thread t(std::move(task), 3, 4);          // 在线程中执行任务
    t.join();

    std::cout << "3 + 4 = " << fut.get() << std::endl;   // 7
    return 0;
}

三者的分工可以这样理解：async 是"开箱即用"（启动 + 拿结果一步到位），promise 是"手动喂结果"（结果来自任何地方，不一定是函数返回值），packaged_task 是"把任务延迟到任意线程执行"（线程池的常用积木）。

---

3.5 std::shared_future：多线程共享结果

std::future 的 get() 是独占的——只能调用一次。如果多个线程都要读同一个结果，用 std::shared_future：调用 fut.share() 从 future 转换得到，之后每个线程都可以 get()（内部共享同一份结果）：

#include <future>
#include <iostream>
#include <thread>
#include <vector>

int main()
{
    std::future<int> fut = std::async(std::launch::async, [] {
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
        return 42;
    });

    std::shared_future<int> shared = fut.share();   // 转成共享 future

    std::vector<std::thread> workers;
    for (int i = 0; i < 3; ++i)
    {
        workers.emplace_back([shared, i] {          // shared_future 可拷贝
            std::cout << "线程 " << i << " 读到结果: " << shared.get() << std::endl;
        });
    }
    for (auto& t : workers)
    {
        t.join();
    }
    return 0;
}

---

4. 线程池

线程池（thread pool）预先创建一批线程，反复复用它们执行任务队列中的任务，避免"每个任务都创建/销毁线程"的开销。核心组件：

1. 任务队列（受锁保护）+ 条件变量
2. 固定数量的工作线程：循环从队列取任务执行
3. 提交接口：把任务包装成 packaged_task 放入队列，返回 future

实现一个通用的线程池：

#include <condition_variable>
#include <functional>
#include <future>
#include <iostream>
#include <mutex>
#include <queue>
#include <thread>
#include <vector>

class ThreadPool
{
public:
    explicit ThreadPool(size_t thread_count)
    {
        for (size_t i = 0; i < thread_count; ++i)
        {
            workers_.emplace_back([this] { worker_loop(); });
        }
    }

    // 提交任务, 返回 future 用于获取结果
    template <typename F, typename... Args>
    auto submit(F&& f, Args&&... args)
        -> std::future<std::invoke_result_t<F, Args...>>
    {
        using ResultType = std::invoke_result_t<F, Args...>;

        // 把任务打包成 packaged_task, 结果写入 future
        auto task = std::make_shared<std::packaged_task<ResultType()>>(
            std::bind(std::forward<F>(f), std::forward<Args>(args)...));

        std::future<ResultType> result = task->get_future();

        {
            std::lock_guard<std::mutex> lock(queue_mtx_);
            tasks_.push([task] { (*task)(); });   // 任务队列中存放 lambda
        }
        cv_.notify_one();                          // 唤醒一个工作线程
        return result;
    }

    ~ThreadPool()
    {
        {
            std::lock_guard<std::mutex> lock(queue_mtx_);
            stop_ = true;
        }
        cv_.notify_all();                          // 唤醒所有线程检查停止标志

        for (auto& worker : workers_)
        {
            worker.join();
        }
    }

private:
    void worker_loop()
    {
        while (true)
        {
            std::function<void()> task;
            {
                std::unique_lock<std::mutex> lock(queue_mtx_);
                // 等待: 队列非空 或 要求停止
                cv_.wait(lock, [this] { return stop_ || !tasks_.empty(); });

                if (stop_ && tasks_.empty())
                {
                    return;                        // 队列清空且要求停止, 退出
                }

                task = std::move(tasks_.front());
                tasks_.pop();
            }
            task();                                // 队列外执行任务
        }
    }

    std::vector<std::thread> workers_;
    std::queue<std::function<void()>> tasks_;
    std::mutex queue_mtx_;
    std::condition_variable cv_;
    bool stop_ = false;
};

使用：

int main()
{
    ThreadPool pool(4);   // 4 个工作线程

    // 提交 8 个任务, 每个任务返回自己的编号的平方
    std::vector<std::future<int>> futures;
    for (int i = 0; i < 8; ++i)
    {
        futures.push_back(pool.submit([i] {
            std::this_thread::sleep_for(std::chrono::milliseconds(20));
            return i * i;
        }));
    }

    int sum = 0;
    for (auto& fut : futures)
    {
        sum += fut.get();   // 逐个取结果
    }
    std::cout << "sum = " << sum << std::endl;   // 0+1+4+9+16+25+36+49 = 140
    return 0;
}

这个实现的关键点：

• 任务类型擦除：队列里存 std::function<void()>，任意签名任务经 lambda 包装后统一入队
• 结果传递：packaged_task 的 future 在 submit 时就返回，任务真正执行时结果才就绪——get() 自然阻塞等待
• 停止协议：析构时置 stop_ = true 并 notify_all，工作线程清空剩余任务后退出
• 任务在锁外执行：执行任务时不持锁，长任务不会阻塞其他线程取任务

注：这只是教学版线程池。生产级线程池还需要考虑：任务异常隔离、动态扩缩容、任务优先级、std::jthread（C++20，支持协作式取消）等。

---

5. 线程优先级

C++ 标准库没有线程优先级接口——优先级是操作系统概念，需要通过平台 API 操作。std::thread::native_handle() 返回线程的原生句柄，可以传给系统 API：

#include <iostream>
#include <thread>

int main()
{
    std::thread t([] { /* 工作 */ });

    // Linux: pthread API (需要 -pthread)
    pthread_t handle = t.native_handle();
    struct sched_param param;
    param.sched_priority = 10;
    pthread_setschedparam(handle, SCHED_OTHER, &param);

    // Windows: Win32 API
    // HANDLE h = t.native_handle();
    // SetThreadPriority(h, THREAD_PRIORITY_ABOVE_NORMAL);

    t.join();
    return 0;
}

关于优先级要注意：

• 提高优先级不等于更快：优先级只影响"竞争 CPU 时的调度顺序"，单核满载时高优先级线程先获得 CPU
• 实时调度策略（SCHED_FIFO/SCHED_RR）需要 root 权限，且可能饿死其他线程，服务器上慎用
• 多数应用不需要手动调优先级：让操作系统按默认策略调度，把精力放在减少持锁时间和合理分片上
• 优先级与锁的交互有陷阱：优先级反转（priority inversion）——低优先级线程持锁，高优先级线程等锁，中优先级线程抢占低优先级线程，高优先级被间接饿死。Linux 的 pthread_mutex 默认不做优先级继承，需要时用 PTHREAD_PRIO_INHERIT 属性

---

6. 线程局部存储 thread_local

thread_local 声明的变量每个线程有一份独立副本，线程之间互不可见、互不干扰。它解决了"全局状态在多线程下不可用"的问题：

#include <iostream>
#include <thread>

thread_local int counter = 0;   // 每个线程独立的 counter

void work(const char* name)
{
    for (int i = 0; i < 3; ++i)
    {
        ++counter;              // 只改自己线程的那一份
    }
    std::cout << name << " 的 counter = " << counter << std::endl;
}

int main()
{
    std::thread t1(work, "线程1");
    std::thread t2(work, "线程2");
    t1.join();
    t2.join();
    std::cout << "主线程的 counter = " << counter << std::endl;
    return 0;
}

线程1 的 counter = 3
线程2 的 counter = 3
主线程的 counter = 0

三个线程各自维护自己的 counter，不需要任何锁——这是 thread_local 最大的价值：把"共享"变成"每线程私有"，从根源上消除竞争。

使用场景：

• 每线程缓存：如随机数引擎（thread_local std::mt19937）、内存分配器的线程缓存
• 每线程上下文：错误码、日志前缀、数据库连接
• 懒初始化单例的每线程版本

注意事项：

• thread_local 变量在线程首次使用时初始化（而不是线程创建时）
• 生命周期：线程结束时析构。detach 的线程结束后其 thread_local 变量也销毁，不要跨线程持有指向他人 thread_local 变量的指针
• thread_local 与 static 可以组合（static thread_local），语义仍是每线程一份
• 每个线程的 thread_local 有独立存储，线程数量多时内存占用按线程数翻倍

典型应用——线程安全的一次性初始化（每线程各做各的）：

thread_local std::string thread_name;

void ensure_name(const std::string& name)
{
    if (thread_name.empty())
    {
        thread_name = name;   // 只影响当前线程
    }
}

---

7. 设施对比总结

设施	解决的问题	关键 API
condition_variable	等待事件 / 条件	wait、notify_one、notify_all
std::async	一步启动异步任务并取结果	std::launch 策略、future::get
std::promise	手动设置异步结果	set_value、set_exception、get_future
std::packaged_task	把任务与结果打包、延迟执行	get_future、operator()
std::shared_future	多个线程共享同一结果	share()、多次 get()
线程池	复用线程，避免反复创建销毁	submit 返回 future
thread_local	每线程独立存储，免锁	thread_local 关键字

选择建议：等一个结果用 async/future；等一个条件/事件用条件变量；任务量大、频繁提交用线程池；每线程独立状态用 thread_local。裸 std::thread 留给"需要完全控制线程生命周期"的场景。

到这一步，多线程的同步与通信手段基本齐全了。还剩下最后一个主题：原子操作——不用锁也能保证正确性的轻量同步手段，以及它背后的硬件原理。

---

需要进一步学习：

• 原子操作

---