18、消息传递与Actor模型：Actor模型概念、CSP与Actor对比、使用channel实现Actor、ZeroMQ与并发

说到并发编程，很多人第一反应就是锁、条件变量、原子操作。嗯，这些确实重要。但我在实际项目中慢慢发现，很多时候我们其实不需要那么“亲密”地共享数据。你想想看，现实世界里的人是怎么协作的？发个消息、等个回复，对吧？这就是消息传递模型的核心理念。

今天我们来聊聊Actor模型和CSP模型。这两个东西听起来高大上，其实说白了就是两种“发消息”的规矩。我当年第一次接触Actor模型是在做一个分布式爬虫项目，那时候被共享状态搞到头秃，后来换成Actor模式，整个世界清净了。

Actor模型概念

Actor模型最早是Carl Hewitt在1973年提出的。它的核心思想就一句话：一切皆Actor，Actor之间只通过消息通信。

每个Actor有三个基本能力：

• 处理收到的消息——你可以理解为Actor有一个“信箱”，别人往里面投信，它一封一封地处理
• 创建新的Actor——一个Actor可以生儿子，儿子还能生孙子
• 向其他Actor发送消息——注意，它只知道对方的地址，不关心对方内部怎么搞

这里有个关键点：Actor内部状态是私有的。没有全局变量，没有共享内存。你想改一个Actor的状态？只能给它发消息，让它自己改。这就从根本上避免了数据竞争。

我在项目中遇到过一个问题：多个线程同时操作一个订单状态，加锁加得死去活来，还经常死锁。后来我把每个订单封装成一个Actor，所有操作都通过消息队列串行化。嗯，死锁问题直接消失了。

CSP与Actor对比

CSP（Communicating Sequential Processes）和Actor经常被放在一起比较。它们都基于消息传递，但哲学不同。

维度	Actor模型	CSP模型
通信方式	异步，发送方不等待	同步（channel操作是阻塞的）
接收方	每个Actor有唯一信箱	多个goroutine可以监听同一个channel
耦合度	发送方知道接收方地址	发送方和接收方通过channel解耦
典型实现	Erlang, Akka, CAF	Go语言的goroutine+channel
错误处理	“让它崩溃”哲学，监督树	通常由调用方处理

说白了，Actor更像“点对点发微信”，你知道发给谁，发完就不管了。CSP更像“在群里发言”，你把消息丢到channel里，谁有空谁处理。

我个人习惯是：如果系统里实体之间有明确的“身份”关系（比如用户、订单、商品），用Actor更自然。如果只是数据流处理（比如管道、流水线），CSP更合适。

使用channel实现Actor

C++标准库没有直接提供Actor框架，但我们可以用channel来模拟。说白了，每个Actor就是一个线程（或协程），加上一个消息队列。

下面是一个简单的实现思路：

#include <queue>
#include <mutex>
#include <condition_variable>
#include <functional>
#include <thread>
#include <any>

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

    Actor() : running_(true) {
        thread_ = std::thread([this] { run(); });
    }

    ~Actor() {
        send([this] { running_ = false; });
        thread_.join();
    }

    void send(Message msg) {
        std::lock_guard<std::mutex> lock(mutex_);
        queue_.push(std::move(msg));
        cv_.notify_one();
    }

private:
    void run() {
        while (running_) {
            Message msg;
            {
                std::unique_lock<std::mutex> lock(mutex_);
                cv_.wait(lock, [this] { return !queue_.empty() || !running_; });
                if (!running_) break;
                msg = std::move(queue_.front());
                queue_.pop();
            }
            msg(); // 处理消息
        }
    }

    std::queue<Message> queue_;
    std::mutex mutex_;
    std::condition_variable cv_;
    std::thread thread_;
    bool running_;
};

你看，核心就是一个线程安全的消息队列。每个Actor在自己的线程里串行处理消息，天然避免了并发问题。

ZeroMQ与并发

ZeroMQ（简称ZMQ）不是Actor框架，但它提供了一套非常优雅的消息传递原语。你可以把它理解为“增强版的socket”，但不需要自己管理连接。

ZMQ支持多种通信模式：

• 请求-回复（REQ/REP）：一问一答，适合RPC
• 发布-订阅（PUB/SUB）：一对多广播，适合事件分发
• 管道（PUSH/PULL）：任务分发，适合流水线
• 配对（PAIR）：一对一双向通信

我在一个实时数据采集项目里用过ZMQ。采集节点用PUB模式往外发数据，多个处理节点用SUB模式订阅。整个系统零配置，节点随便加，爽得很。

下面是一个简单的ZMQ发布-订阅示例：

// 发布端
#include <zmq.hpp>

int main() {
    zmq::context_t context(1);
    zmq::socket_t publisher(context, zmq::socket_type::pub);
    publisher.bind("tcp://*:5556");

    while (true) {
        std::string data = "sensor:temperature:25.3";
        zmq::message_t message(data.data(), data.size());
        publisher.send(message, zmq::send_flags::none);
        std::this_thread::sleep_for(std::chrono::seconds(1));
    }
    return 0;
}

// 订阅端
#include <zmq.hpp>

int main() {
    zmq::context_t context(1);
    zmq::socket_t subscriber(context, zmq::socket_type::sub);
    subscriber.connect("tcp://localhost:5556");
    subscriber.set(zmq::sockopt::subscribe, "sensor"); // 只订阅sensor前缀的消息

    while (true) {
        zmq::message_t message;
        auto result = subscriber.recv(message, zmq::recv_flags::none);
        if (result) {
            std::string data(static_cast<char*>(message.data()), message.size());
            std::cout << "Received: " << data << std::endl;
        }
    }
    return 0;
}

ZMQ的并发模型很有意思：它内部用多个线程处理I/O，但对外暴露的API是线程安全的。你可以在多个线程里共享同一个socket，也可以每个线程创建自己的socket。我个人建议后者——每个线程独享socket，避免不必要的锁竞争。

最后，我们来画一张图，把今天讲的东西串起来：

这张图展示了三种消息传递模型的对比。左边是Actor，每个Actor独立运行，通过消息互相通信。中间是CSP，goroutine通过channel交换数据。下面是ZeroMQ，它提供了多种通信模式，适合分布式场景。

说实话，没有银弹。Actor模型适合有状态、有身份的实体；CSP适合数据流处理；ZeroMQ适合跨进程、跨网络的通信。我在实际项目中经常把它们组合使用——内部用Actor管理业务状态，跨进程用ZeroMQ，数据管道用CSP思想。