第九章：高级并发与原子操作

并发编程，说白了就是让程序同时干多件事。但多件事一起干，就容易打架。我做了这么多年系统编程，见过太多因为并发问题导致的诡异崩溃——有时候一周才复现一次，查起来简直要命。

这一章，我们来聊聊C++里真正实用的并发工具。不是那种教科书式的理论，而是你写生产代码时真正用得上的东西。

9.1 std::atomic与内存序：别再乱加锁了

很多人一提到线程安全，第一反应就是加锁。其实很多时候，一个std::atomic<int>就能搞定，性能比互斥锁高一个数量级。

先看个最简单的例子：

std::atomic<int> counter{0};

void worker() {
    for (int i = 0; i < 10000; ++i) {
        counter.fetch_add(1, std::memory_order_relaxed);
    }
}

// 主线程
std::thread t1(worker), t2(worker);
t1.join(); t2.join();
std::cout << counter.load(); // 输出20000，稳稳的

嗯，这里要注意：memory_order_relaxed只保证原子性，不保证顺序。什么意思？

我举个例子你就明白了。假设你有两个原子变量：

std::atomic<bool> ready{false};
std::atomic<int> data{0};

// 线程A
data.store(42, std::memory_order_relaxed);
ready.store(true, std::memory_order_relaxed);

// 线程B
if (ready.load(std::memory_order_relaxed)) {
    // 这里读data，可能是0！因为relaxed不保证顺序
    std::cout << data.load(std::memory_order_relaxed);
}

为什么会这样？因为CPU和编译器可能会重排指令。线程A里ready.store可能先于data.store执行，线程B看到ready为true时，data可能还没写进去。

我在项目中遇到过这种坑。当时写一个无锁队列，用了relaxed序，结果压测时偶尔读到脏数据。排查了两天才发现是内存序的问题。

常用的内存序就三种：

内存序	含义	性能	使用场景
relaxed	只保证原子性，不保证顺序	最快	计数器、统计量
acquire/release	配对使用，保证因果关系	中等	生产者-消费者
seq_cst	全局顺序一致	最慢	默认选项，通用场景

9.2 无锁数据结构基础

无锁编程，听起来高大上，其实核心思想就一句话：用原子操作代替锁。

但说起来容易，做起来难。我最早写无锁栈的时候，改了七八版才跑对。这里分享一个最经典的无锁栈实现：

template<typename T>
class LockFreeStack {
    struct Node {
        T data;
        Node* next;
    };
    std::atomic<Node*> head{nullptr};

public:
    void push(const T& val) {
        Node* new_node = new Node{val, nullptr};
        new_node->next = head.load(std::memory_order_relaxed);
        // CAS循环：如果head没变，就替换；否则重试
        while (!head.compare_exchange_weak(
            new_node->next,
            new_node,
            std::memory_order_release,
            std::memory_order_relaxed
        )) {
            // 失败时，new_node->next已被更新为最新的head
        }
    }

    bool pop(T& result) {
        Node* old_head = head.load(std::memory_order_relaxed);
        while (old_head && !head.compare_exchange_weak(
            old_head,
            old_head->next,
            std::memory_order_acquire,
            std::memory_order_relaxed
        )) {
            // 重试
        }
        if (!old_head) return false;
        result = old_head->data;
        delete old_head;  // 注意：这里可能有ABA问题
        return true;
    }
};

9.3 线程池实现：别再频繁创建线程了

你想想看，每次来一个任务就new thread，这得多浪费？线程创建销毁的开销可不小。线程池就是预先创建一批线程，让它们等着干活。

我写过一个生产级的线程池，核心代码大概这样：

class ThreadPool {
    std::vector<std::thread> workers;
    std::queue<std::function<void()>> tasks;
    std::mutex queue_mutex;
    std::condition_variable cv;
    bool stop{false};

public:
    ThreadPool(size_t threads) {
        for (size_t i = 0; i < threads; ++i) {
            workers.emplace_back([this] {
                while (true) {
                    std::function<void()> task;
                    {
                        std::unique_lock<std::mutex> lock(queue_mutex);
                        cv.wait(lock, [this] {
                            return stop || !tasks.empty();
                        });
                        if (stop && tasks.empty()) return;
                        task = std::move(tasks.front());
                        tasks.pop();
                    }
                    task();  // 执行任务时释放锁
                }
            });
        }
    }

    template<typename F>
    auto enqueue(F&& f) -> std::future<decltype(f())> {
        auto task = std::make_shared<std::packaged_task<decltype(f())()>>(
            std::forward<F>(f)
        );
        std::future<decltype(f())> res = task->get_future();
        {
            std::lock_guard<std::mutex> lock(queue_mutex);
            tasks.emplace([task](){ (*task)(); });
        }
        cv.notify_one();
        return res;
    }
};

9.4 std::future与std::async：异步任务的优雅写法

说实话，手动管理线程挺累的。C++11提供的std::async和std::future，让你用同步的写法写异步代码。

// 简单用法：异步执行一个函数
std::future<int> result = std::async(std::launch::async, []() {
    std::this_thread::sleep_for(std::chrono::seconds(2));
    return 42;
});

// 干点别的事...
std::cout << "等待结果...\n";

// 获取结果（会阻塞直到完成）
int value = result.get();
std::cout << "结果是: " << value << "\n";

这里有三个启动策略：

策略	行为	什么时候用
std::launch::async	立即在新线程执行	确定要异步执行
std::launch::deferred	延迟到get()时执行	懒加载场景
默认（两者都有）	由实现决定	不推荐，行为不确定

更复杂的场景可以用std::packaged_task和std::promise：

// promise/future配对，用于线程间传递结果
std::promise<int> prom;
std::future<int>> fut = prom.get_future();

std::thread t([&prom] {
    int result = do_heavy_work();
    prom.set_value(result);  // 线程安全地传递结果
});

int final_result = fut.get();  // 主线程等待
t.join();

嗯，这里要注意：promise和future是move-only的，不能拷贝。而且每个promise只能调用一次set_value或set_exception。

最后说一句：并发编程没有银弹。原子操作快，但容易出错；锁慢，但逻辑清晰。我的原则是：先用锁写对，再考虑用原子操作优化。性能瓶颈在哪里，profile了再说。

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