14、并发模式：Actor 模型实现，线程池设计，无锁编程基础（std::atomic）

并发编程，说白了就是让程序同时干多件事。但这件事，做起来比说起来难十倍。我见过太多项目，一上多线程就崩，一崩就查好几天。嗯，今天咱们就来聊聊三种实用的并发模式：Actor 模型、线程池、还有无锁编程。

14.1 Actor 模型：让每个对象自己干活

Actor 模型的核心思想很简单：每个 Actor 是一个独立的活动实体，它有自己的邮箱，别人只能通过发消息跟它通信。没有共享内存，没有锁，每个 Actor 内部是单线程的。

我在项目中遇到过这样一个场景：游戏服务器里，每个玩家是一个 Actor。玩家移动、攻击、聊天，都通过消息队列发给对应的 Actor 处理。这样设计的好处是，你永远不会出现两个线程同时修改玩家数据的情况。

来看一个简单的 Actor 实现：

#include <queue>
#include <mutex>
#include <thread>
#include <functional>
#include <memory>

class Actor {
public:
    using Message = std::function<void()>;

    void send(Message msg) {
        std::lock_guard<std::mutex> lock(m_mutex);
        m_queue.push(std::move(msg));
        m_cv.notify_one();
    }

    void start() {
        m_thread = std::thread([this] {
            while (m_running) {
                Message msg;
                {
                    std::unique_lock<std::mutex> lock(m_mutex);
                    m_cv.wait(lock, [this] {
                        return !m_queue.empty() || !m_running;
                    });
                    if (!m_running) break;
                    msg = std::move(m_queue.front());
                    m_queue.pop();
                }
                msg();  // 单线程执行，天然安全
            }
        });
    }

    void stop() {
        m_running = false;
        m_cv.notify_all();
        if (m_thread.joinable()) m_thread.join();
    }

private:
    std::queue<Message> m_queue;
    std::mutex m_mutex;
    std::condition_variable m_cv;
    std::thread m_thread;
    std::atomic<bool> m_running{true};
};

你想想看，每个 Actor 内部只有一个线程在跑消息循环。这就意味着，你不需要在 Actor 内部加任何锁。所有并发问题，都变成了消息顺序问题。

14.2 线程池设计：别让线程频繁创建销毁

线程的创建和销毁是有代价的。我记得有一次，一个后台服务每秒钟要处理上千个请求，每个请求都 new 一个线程。结果 CPU 全花在线程切换上了，真正干活的时间不到 10%。

线程池就是解决这个问题的。它预先创建一批线程，任务来了直接丢进去执行。线程用完了也不销毁，等着下一个任务。

一个工业级线程池需要考虑这些：

• 任务队列：线程安全的队列，支持多生产者多消费者
• 工作线程：固定数量，避免频繁创建
• 动态调整：任务多时增加线程，空闲时减少
• 优雅关闭：等待所有任务完成再退出

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

    template<class F>
    void enqueue(F&& f) {
        {
            std::lock_guard<std::mutex> lock(m_mutex);
            m_tasks.emplace(std::forward<F>(f));
        }
        m_condition.notify_one();
    }

    ~ThreadPool() {
        {
            std::lock_guard<std::mutex> lock(m_mutex);
            m_stop = true;
        }
        m_condition.notify_all();
        for (auto& worker : m_workers) {
            worker.join();
        }
    }

private:
    std::vector<std::thread> m_workers;
    std::queue<std::function<void()>> m_tasks;
    std::mutex m_mutex;
    std::condition_variable m_condition;
    std::atomic<bool> m_stop;
};

14.3 无锁编程基础：std::atomic 的正确用法

无锁编程，说白了就是不用 mutex 也能保证线程安全。听起来很酷，但坑特别多。我曾经在一个高并发计数器上用了自旋锁，结果死锁了三天才找到原因。

std::atomic 是 C++11 引入的原子操作库。它保证了对变量的读写是原子的，不会被线程调度打断。

操作	说明	示例
load	原子读取	val = atomic_var.load()
store	原子写入	atomic_var.store(42)
exchange	原子交换	old = atomic_var.exchange(10)
compare_exchange_weak	CAS 操作	atomic_var.compare_exchange_weak(expected, desired)
fetch_add	原子加法	old = atomic_var.fetch_add(1)

来看一个无锁队列的简单实现：

template<typename T>
class LockFreeQueue {
private:
    struct Node {
        T data;
        std::atomic<Node*> next;
        Node(const T& val) : data(val), next(nullptr) {}
    };

    std::atomic<Node*> head;
    std::atomic<Node*> tail;

public:
    LockFreeQueue() {
        Node* dummy = new Node(T());
        head.store(dummy);
        tail.store(dummy);
    }

    void push(const T& val) {
        Node* new_node = new Node(val);
        while (true) {
            Node* last = tail.load();
            Node* next = last->next.load();
            if (last == tail.load()) {
                if (next == nullptr) {
                    if (last->next.compare_exchange_weak(next, new_node)) {
                        tail.compare_exchange_weak(last, new_node);
                        return;
                    }
                } else {
                    tail.compare_exchange_weak(last, next);
                }
            }
        }
    }

    bool pop(T& result) {
        while (true) {
            Node* first = head.load();
            Node* last = tail.load();
            Node* next = first->next.load();
            if (first == head.load()) {
                if (first == last) {
                    if (next == nullptr) return false;
                    tail.compare_exchange_weak(last, next);
                } else {
                    result = next->data;
                    if (head.compare_exchange_weak(first, next)) {
                        delete first;
                        return true;
                    }
                }
            }
        }
    }
};

为什么会用 weak 而不是 strong？因为 weak 在某些平台上可能出现「伪失败」，但性能更好。在循环里用 weak 是安全的，因为失败了就重试。

14.4 三种模式的对比与选择

这三种模式各有适用场景。我一般这样选：

• Actor 模型：适合有明确实体边界的场景，比如玩家、设备、会话。每个实体独立，消息驱动。
• 线程池：适合任务并行的场景，比如计算密集型任务、IO 回调。任务之间没有共享状态。
• 无锁编程：适合高频、短小的操作，比如计数器、队列、标志位。但复杂度高，不建议新手大面积使用。

你想想看，如果每个玩家都用一个线程池任务来处理，那玩家之间的交互怎么办？消息怎么传递？这时候 Actor 模型就更合适。反过来，如果只是计算一堆数字，用 Actor 模型反而增加了不必要的消息开销。

嗯，这三种模式不是互斥的。实际项目中，我经常把 Actor 模型和线程池结合起来用：每个 Actor 内部用一个线程池来执行任务，Actor 之间通过消息通信。这样既有 Actor 的隔离性，又有线程池的高效性。

最后说一句：并发编程没有银弹。理解每种模式的优缺点，根据场景选择，才是正道。