一、从单线程到多线程：并发编程的进阶之路

说实话，很多C++开发者写了好几年代码，对并发编程的理解还停留在“开个线程跑任务”的阶段。我自己刚入行时也是这样——直到有一次线上服务因为锁竞争导致性能雪崩，才真正意识到并发编程的水有多深。

今天我们要聊的这几个主题，说白了就是并发编程的“进阶装备”。线程池、读写锁、future/promise、async/packaged_task，还有无锁编程——这些工具用好了，你的程序性能能上一个台阶；用不好，调试起来能让你怀疑人生。

核心要点：本章内容围绕“如何高效、安全地管理并发任务”展开。我们从线程池这个最实用的工具开始，逐步深入到同步机制和异步编程模型，最后简单聊聊无锁编程这个“高阶玩法”。

二、线程池：别再手动创建线程了

2.1 为什么需要线程池？

你想想看，每次来一个任务就 new 一个线程，任务结束就销毁——这效率能高吗？线程的创建和销毁是有开销的，频繁操作会导致性能抖动。我在一个实时数据处理项目中就踩过这个坑：每秒上千个请求，每个请求都开新线程，结果系统大部分时间都在“生娃”和“埋娃”。

线程池的核心思想很简单：提前创建一批线程，任务来了直接分配，任务结束线程不销毁，等着接下一个活。这就像餐馆里提前招好服务员，客人来了直接服务，而不是现去人才市场招人。

2.2 一个简单的线程池实现

下面这个实现我用了很多年，虽然简单但足够实用。注意看核心组件：

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

class ThreadPool {
public:
    ThreadPool(size_t threads) : stop(false) {
        for(size_t i = 0; i < threads; ++i) {
            workers.emplace_back([this] {
                while(true) {
                    std::function<void()> task;
                    {
                        std::unique_lock<std::mutex> lock(this->queue_mutex);
                        this->condition.wait(lock, [this] {
                            return this->stop || !this->tasks.empty();
                        });
                        if(this->stop && this->tasks.empty())
                            return;
                        task = std::move(this->tasks.front());
                        this->tasks.pop();
                    }
                    task();
                }
            });
        }
    }

    template<class F, class... Args>
    auto enqueue(F&& f, Args&&... args) 
        -> std::future<typename std::result_of<F(Args...)>::type> {
        using return_type = typename std::result_of<F(Args...)>::type;
        
        auto task = std::make_shared<std::packaged_task<return_type()>>(
            std::bind(std::forward<F>(f), std::forward<Args>(args)...)
        );
        
        std::future<return_type> res = task->get_future();
        {
            std::unique_lock<std::mutex> lock(queue_mutex);
            if(stop)
                throw std::runtime_error("enqueue on stopped ThreadPool");
            tasks.emplace([task](){ (*task)(); });
        }
        condition.notify_one();
        return res;
    }

    ~ThreadPool() {
        {
            std::unique_lock<std::mutex> lock(queue_mutex);
            stop = true;
        }
        condition.notify_all();
        for(std::thread &worker : workers)
            worker.join();
    }

private:
    std::vector<std::thread> workers;
    std::queue<std::function<void()>> tasks;
    std::mutex queue_mutex;
    std::condition_variable condition;
    bool stop;
};

使用建议：线程数量一般设为 std::thread::hardware_concurrency() 的返回值。如果是CPU密集型任务，就设成这个数；如果是IO密集型，可以适当多设一些。

2.3 使用示例

int main() {
    ThreadPool pool(4);
    
    // 提交任务，获取future
    auto result = pool.enqueue([](int a, int b) {
        return a + b;
    }, 10, 20);
    
    std::cout << "Result: " << result.get() << std::endl;
    
    // 批量提交
    std::vector<std::future<int>> results;
    for(int i = 0; i < 8; ++i) {
        results.emplace_back(
            pool.enqueue([i] {
                return i * i;
            })
        );
    }
    
    for(auto &result : results)
        std::cout << result.get() << ' ';
    std::cout << std::endl;
    
    return 0;
}

三、读写锁：读多写少的场景利器

3.1 shared_mutex 是什么？

传统的 std::mutex 是“排他锁”——不管读还是写，一次只能一个线程访问。但实际场景中，读操作往往远多于写操作。比如一个配置表，99%的时间都在读，只有1%的时间在更新。这时候用普通互斥锁，读线程之间互相阻塞，性能就浪费了。

读写锁（C++17 的 std::shared_mutex）解决了这个问题：多个线程可以同时读，但写的时候独占。说白了就是“读读不互斥，读写互斥，写写互斥”。

3.2 实战用法

#include <shared_mutex>
#include <unordered_map>
#include <string>

class ConfigManager {
public:
    std::string get(const std::string& key) {
        std::shared_lock<std::shared_mutex> lock(mutex_);  // 共享锁
        auto it = config_.find(key);
        return it != config_.end() ? it->second : "";
    }
    
    void set(const std::string& key, const std::string& value) {
        std::unique_lock<std::shared_mutex> lock(mutex_);  // 独占锁
        config_[key] = value;
    }
    
private:
    std::unordered_map<std::string, std::string> config_;
    mutable std::shared_mutex mutex_;
};

注意：我曾经在一个项目中看到有人用 shared_mutex 保护一个只有几十个元素的 vector，结果性能反而更差了。读写锁本身有开销，只有在读操作远多于写操作（比如读:写 > 10:1）时才有优势。小数据量直接用普通 mutex 反而更快。

四、future 与 promise：线程间的“快递员”

4.1 基本概念

future 和 promise 是 C++11 引入的一对“好基友”。它们的作用是在线程之间传递数据——一个线程通过 promise 设置值，另一个线程通过 future 获取值。这就像你网购：promise 是卖家发货，future 是快递单号，你可以拿着单号随时查物流（等待结果）。

4.2 典型用法

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

void worker(std::promise<int> prom) {
    // 模拟耗时计算
    std::this_thread::sleep_for(std::chrono::seconds(2));
    prom.set_value(42);  // 设置结果
}

int main() {
    std::promise<int> prom;
    std::future<int> fut = prom.get_future();
    
    std::thread t(worker, std::move(prom));
    
    // 主线程可以干别的事
    std::cout << "Waiting for result..." << std::endl;
    
    int result = fut.get();  // 阻塞等待结果
    std::cout << "Result: " << result << std::endl;
    
    t.join();
    return 0;
}

关键点：future.get() 只能调用一次！多次调用会抛出异常。如果你需要多个线程等待同一个结果，用 std::shared_future。

五、async 与 packaged_task：更优雅的异步编程

5.1 std::async：一键异步

std::async 是 future/promise 的“高级封装”。你只需要告诉它“我要执行这个函数”，它自动帮你创建线程（或者在线程池中执行），然后返回一个 future。

#include <future>
#include <iostream>

int compute(int x) {
    return x * x;
}

int main() {
    // 默认策略：可能异步，也可能同步
    auto fut1 = std::async(compute, 10);
    
    // 强制异步执行
    auto fut2 = std::async(std::launch::async, compute, 20);
    
    // 延迟执行（直到调用 get 才执行）
    auto fut3 = std::async(std::launch::deferred, compute, 30);
    
    std::cout << fut1.get() << " " 
              << fut2.get() << " " 
              << fut3.get() << std::endl;
    
    return 0;
}

我的经验：std::async 默认策略（std::launch::async | std::launch::deferred）有个坑——它可能不创建新线程，而是在调用 get() 时同步执行。如果你明确需要异步，请指定 std::launch::async。

5.2 packaged_task：把函数打包成“任务包”

packaged_task 比 async 更底层一些。它把函数和 future 打包在一起，你可以把这个“任务包”交给线程池或者某个线程去执行。

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

int main() {
    // 创建一个打包任务
    std::packaged_task<int(int, int)> task([](int a, int b) {
        return a + b;
    });
    
    // 获取 future
    std::future<int> fut = task.get_future();
    
    // 在另一个线程执行
    std::thread t(std::move(task), 10, 20);
    
    std::cout << "Result: " << fut.get() << std::endl;
    
    t.join();
    return 0;
}

六、无锁编程简介：高性能的“双刃剑”

6.1 为什么需要无锁？

锁虽然好用，但有两个问题：一是可能死锁，二是高并发下锁竞争会导致性能下降。无锁编程（Lock-Free）通过原子操作和 CAS（Compare-And-Swap）来实现线程安全，避免了锁的开销。

6.2 一个简单的无锁栈

#include <atomic>

template<typename T>
class LockFreeStack {
private:
    struct Node {
        T data;
        Node* next;
        Node(const T& d) : data(d), next(nullptr) {}
    };
    
    std::atomic<Node*> head_;
    
public:
    LockFreeStack() : head_(nullptr) {}
    
    void push(const T& value) {
        Node* new_node = new Node(value);
        new_node->next = head_.load(std::memory_order_relaxed);
        while(!head_.compare_exchange_weak(
            new_node->next, 
            new_node,
            std::memory_order_release,
            std::memory_order_relaxed
        ));
    }
    
    bool pop(T& value) {
        Node* old_head = head_.load(std::memory_order_relaxed);
        while(old_head != nullptr) {
            if(head_.compare_exchange_weak(
                old_head,
                old_head->next,
                std::memory_order_acquire,
                std::memory_order_relaxed
            )) {
                value = old_head->data;
                delete old_head;
                return true;
            }
        }
        return false;
    }
};

警告：无锁编程非常容易出错。我见过一个团队花了两周时间调试一个无锁队列的 ABA 问题。除非你是并发编程专家，否则建议优先使用锁和标准库提供的并发容器。无锁编程适合的场景：性能要求极高、且你能证明锁是瓶颈的时候。

七、知识体系总览

下面这张图总结了本章的核心知识点和它们之间的关系：

并发编程进阶知识体系 并发编程进阶 线程池 • 复用线程，减少创建开销 • 控制并发数量 • 任务队列 + 工作线程 读写锁 (shared_mutex) • 读读不互斥 • 读写互斥，写写互斥 • 适合读多写少场景 Future / Promise • 线程间传递结果 • promise设置值 • future获取值 Async / PackagedTask • async：一键异步 • packaged_task：可移动的任务包 无锁编程 • 原子操作 + CAS • 避免锁竞争 • 难度高，易出错

八、总结与避坑指南

聊了这么多，最后分享几个我踩过的坑：

  • 线程池大小不是越大越好——我曾经把线程池设成100个线程，结果CPU上下文切换开销比任务执行时间还长。对于CPU密集型任务，线程数等于核心数就够了。
  • shared_mutex 不是万能药——如果写操作频繁，读写锁可能比普通互斥锁还慢。我在一个实时日志系统中就吃过这个亏。
  • future.get() 会阻塞当前线程——如果你在UI线程中调用，界面会卡死。考虑用 then() 或者回调的方式处理结果。
  • 无锁编程慎用——除非你能用性能分析工具证明锁是瓶颈，否则别碰。我见过太多“优化”成无锁后反而更慢的例子。

记住一句话：先保证正确，再追求性能。并发编程的调试难度是指数级上升的，能用简单方案就别炫技。

公众号:蓝海资料掘金营，微信deep3321

← 上一章 📖 返回目录 下一章 →