1. 项目概述为什么是Actor模型为什么是C如果你是一个有几年经验的C程序员可能已经熟练掌握了STL容器、多线程同步、RAII这些基础技能也写过一些并发程序。但当你面对一个需要处理海量并发连接、或者需要构建一个高吞吐、低延迟的分布式服务核心时是否感觉传统的“线程锁”模型开始力不从心锁竞争、死锁、数据竞争、线程上下文切换开销这些问题像幽灵一样缠绕着项目让代码变得复杂且脆弱。这正是我选择“从零手写一个Actor框架”作为深度进阶项目的原因。Actor模型提供了一种截然不同的并发编程范式。它不是一个具体的库而是一种设计思想每个Actor都是一个独立的计算实体拥有私有的状态并且只通过异步消息与其他Actor通信。没有共享内存自然也就没有锁。这听起来像是为C这种追求极致性能和控制力的语言量身定做的挑战——我们需要在语言层面手动构建一套安全、高效的消息传递和生命周期管理机制。市面已有成熟的Actor框架如CAFC Actor Framework或Erlang/OTP。但“手写”的意义在于你将被迫深入理解每一个细节消息如何序列化与传递邮箱Mailbox如何实现才能兼顾吞吐与公平Actor的生命周期创建、运行、停止如何安全管理调度器如何高效地利用CPU核心这个过程是对你C综合能力的一次“大考”涉及移动语义、智能指针、类型擦除、模板元编程、无锁数据结构等多个高阶主题。完成它你收获的不仅仅是一个可用的框架更是一种构建复杂、可靠并发系统的底层思维和能力。2. 核心设计定义我们的微型Actor宇宙在动手写代码之前我们必须为这个框架划定清晰的边界并做出关键的设计决策。一个完整的Actor框架是庞大的我们的目标是构建一个核心完备、可扩展的“微型宇宙”确保其概念的正确性和核心组件的可用性。2.1 核心组件定义我们的框架将围绕以下几个核心组件构建Actor基类actor_base所有用户自定义Actor的基类。它定义了一个Actor最基本的接口接收消息的入口。我们将使用一个纯虚函数来处理消息。消息体messageActor之间通信的载体。它必须能够封装任意类型的数据并安全地传递给目标Actor。这里需要用到类型擦除技术。邮箱mailbox每个Actor都有一个私有的邮箱用于缓存接收到的消息。它是一个线程安全的队列是生产者和消费者模型的关键。执行单元execution_unit或调度器scheduler负责从Actor的邮箱中取出消息并调用该Actor的消息处理函数。它可以是一个简单的线程池每个工作线程从一个全局队列中窃取任务Actor消息。Actor引用actor_ref指向Actor的智能指针。它封装了向目标Actor发送消息的逻辑并管理着Actor的生命周期依赖防止Actor在处理消息时被意外销毁。系统actor_system框架的入口点和容器。负责创建Actor、管理调度器、提供全局配置和关闭例程。2.2 关键设计决策消息传递语义移动而非拷贝在C中为了效率我们应优先考虑移动语义。message对象在传递过程中应被移动而非拷贝。这要求我们的邮箱队列如std::queue存储的是可移动的类型或者使用像moodycamel::ConcurrentQueue这样的第三方无锁队列其本身就支持移动语义。Actor生命周期由引用计数管理我们采用std::shared_ptr和std::weak_ptr来管理Actor的生命周期。actor_system持有Actor的std::shared_ptr。actor_ref内部包含一个std::weak_ptr指向目标Actor。当需要发送消息时actor_ref会尝试将weak_ptr提升为shared_ptr。如果提升成功说明Actor还活着则发送消息如果失败则静默丢弃消息。这确保了发送消息的操作不会延长Actor的生命周期也避免了悬空指针。调度策略工作窃取Work-Stealing为了实现良好的负载均衡我们的调度器将采用工作窃取算法。每个工作线程或调度器维护一个本地任务队列存放Actor和message对。当自己的队列为空时可以去其他线程的队列里“窃取”任务。这能有效减少竞争提高多核利用率。我们可以基于std::deque和std::mutex实现一个简易版本或者直接集成moodycamel::ConcurrentQueue作为每个Actor的邮箱并由调度器统一轮询。错误处理让Actor崩溃Let it crash这是从Erlang哲学中借鉴的。一个Actor内部发生的未处理异常不应该导致整个系统崩溃。我们的调度器在调用Actor的消息处理函数时应该用try-catch块包裹。一旦捕获异常就记录日志然后停止该Actor将其标记为失效不再调度。其父Actor或监控者Supervisor可以收到通知并决定是否重启它。在我们的初级版本中可以先实现简单的日志记录和停止。3. 实现详解从消息到调度一步步构建现在让我们进入具体的实现环节。我将按照依赖顺序从内到外构建这些组件。3.1 消息message的实现类型擦除的经典应用消息需要承载任意类型的数据。我们可以使用标准库的std::any但它的性能开销和类型操作可能不是最优的。这里我们实现一个自定义的、基于void*和函数指针的类型擦除容器通常称为“类型安全的void*”。class message { private: // 内部接口基类 struct message_base { virtual ~message_base() default; virtual std::unique_ptrmessage_base clone() const 0; virtual const std::type_info type() const noexcept 0; }; // 内部模板派生类保存实际数据 templatetypename T struct message_impl : public message_base { T data_; explicit message_impl(T data) : data_(std::forwardT(data)) {} explicit message_impl(const T data) : data_(data) {} std::unique_ptrmessage_base clone() const override { return std::make_uniquemessage_implT(data_); } const std::type_info type() const noexcept override { return typeid(T); } }; std::unique_ptrmessage_base content_; public: // 默认构造一个空消息 message() default; // 通用构造支持任意可移动构造的类型 templatetypename T, typename std::enable_if_t!std::is_same_vstd::decay_tT, message explicit message(T data) : content_(std::make_uniquemessage_implstd::decay_tT(std::forwardT(data))) {} // 移动构造/赋值 message(message) noexcept default; message operator(message) noexcept default; // 拷贝操作谨慎使用可能昂贵 message(const message other) : content_(other.content_ ? other.content_-clone() : nullptr) {} message operator(const message other) { if (this ! other) { content_ other.content_ ? other.content_-clone() : nullptr; } return *this; } // 检查是否持有数据 bool empty() const noexcept { return !content_; } // 获取内部数据的类型信息 const std::type_info type() const noexcept { static const std::type_info empty_type typeid(void); return content_ ? content_-type() : empty_type; } // 安全地获取数据使用前必须用 isT() 检查 templatetypename T T* get_if() noexcept { if (type() typeid(T)) { return static_castmessage_implT*(content_.get())-data_; } return nullptr; } templatetypename T const T* get_if() const noexcept { return const_castmessage*(this)-get_ifT(); } // 更安全的获取方式失败则抛出 std::bad_cast templatetypename T T get() { auto ptr get_ifT(); if (!ptr) { throw std::bad_cast(); } return *ptr; } };注意这个message类实现了深拷贝但在Actor框架中消息在发送后即被移动拷贝操作应极少发生。getT()提供了类型安全的访问但使用者必须知道消息的确切类型。在实际框架中我们通常会为每种消息类型定义唯一的type_id并在Actor内部使用if或switch进行模式匹配这比依赖typeid更高效、更灵活。3.2 Actor基类与邮箱mailboxActor基类非常简单它主要提供一个处理消息的虚函数接口和一个邮箱。// 前向声明 class actor_system; class actor_ref; class actor_base : public std::enable_shared_from_thisactor_base { friend class actor_system; friend class actor_ref; protected: // 邮箱使用一个简单的线程安全队列。实际项目中可替换为无锁队列。 using mailbox_type moodycamel::ConcurrentQueuemessage; mailbox_type mailbox_; std::atomicbool stopped_{false}; // 每个Actor必须实现此方法来处理消息 virtual void on_message(const message msg) 0; // 由调度器调用处理邮箱中的一条消息 bool process_one_message() { message msg; if (mailbox_.try_dequeue(msg)) { try { on_message(msg); } catch (const std::exception e) { // 错误处理记录日志并标记停止 std::cerr Actor crashed: e.what() std::endl; stopped_.store(true); } return true; } return false; // 邮箱为空 } public: virtual ~actor_base() default; // 启动、停止等生命周期方法可以由 actor_system 管理 };邮箱的实现直接使用了moodycamel::ConcurrentQueue这是一个高性能的无锁多生产者多消费者队列非常适合作为Actor的邮箱。如果你的项目不希望引入第三方库可以用std::queue搭配std::mutex和std::condition_variable实现一个线程安全队列但性能会有所差距。3.3 Actor引用actor_ref安全的通信句柄actor_ref是外部世界与Actor交互的唯一安全通道。它不拥有Actor只持有弱引用。class actor_ref { std::weak_ptractor_base actor_weak_ptr_; // 可以附加一个指向 actor_system 的指针或引用用于消息路由这里简化处理。 public: actor_ref() default; explicit actor_ref(const std::shared_ptractor_base actor) : actor_weak_ptr_(actor) {} // 检查引用是否仍然有效Actor是否存活 bool expired() const { return actor_weak_ptr_.expired(); } // 模板方法发送消息 templatetypename MsgType void send(MsgType msg) { if (auto actor_ptr actor_weak_ptr_.lock()) { // 将消息打包成通用 message 对象 message wrapped_msg(std::forwardMsgType(msg)); // 投递到该Actor的邮箱 static_castactor_base*(actor_ptr.get())-mailbox_.enqueue(std::move(wrapped_msg)); // 注意这里需要通知调度器有新的消息到达否则Actor可能不会被调度。 // 这通常通过 actor_system 或调度器的一个通知机制来完成。 // 例如scheduler_-notify_new_task(); } // 如果Actor已死则静默丢弃消息Let it crash 哲学的一部分 } };实操心得actor_ref::send是性能关键路径。这里进行了两次动态操作1.weak_ptr::lock()涉及原子操作2. 邮箱的enqueue。确保它们足够高效。另外发送消息后通知调度器是必须的否则框架无法工作。一个常见的优化是“批量通知”或“偷懒通知”即工作线程在空闲时主动扫描所有Actor的邮箱而不是每发一条消息就唤醒一次调度器。3.4 调度器scheduler与执行单元调度器是框架的大脑。我们实现一个简化版本使用固定数量的工作线程每个线程不断尝试从“全局待处理Actor队列”或通过工作窃取获取任务。class scheduler { std::vectorstd::thread workers_; std::atomicbool running_{false}; // 全局任务队列存放有待处理消息的Actor的弱引用。 // 更复杂的实现中每个工作线程会有本地队列。 moodycamel::ConcurrentQueuestd::weak_ptractor_base global_pending_actors_; // 工作线程的主循环 void worker_thread(int thread_id) { while (running_.load(std::memory_order_relaxed)) { std::weak_ptractor_base actor_weak; // 尝试从全局队列获取一个Actor if (global_pending_actors_.try_dequeue(actor_weak)) { if (auto actor actor_weak.lock()) { // 处理该Actor的一条消息 if (!actor-process_one_message()) { // 如果处理完一条消息后邮箱为空则暂时不把它放回队列 // 如果邮箱不为空但process_one_message返回false这通常不会发生。 // 更完善的实现会检查邮箱是否还有消息有则重新入队。 } else { // 如果处理了消息并且Actor未停止且邮箱可能还有消息则重新放入队列 // 这里简化处理只要成功处理一条就认为可能还有重新入队。 // 这可能导致饥饿更好的办法是只在邮箱非空时才重新入队。 global_pending_actors_.enqueue(std::move(actor_weak)); } } // 如果lock失败Actor已死丢弃该任务 } else { // 队列为空休眠一小段时间避免空转 std::this_thread::yield(); // 更优的做法是使用条件变量等待 } } } public: scheduler(size_t thread_count std::thread::hardware_concurrency()) { if (thread_count 0) thread_count 1; running_.store(true); workers_.reserve(thread_count); for (size_t i 0; i thread_count; i) { workers_.emplace_back(scheduler::worker_thread, this, i); } } ~scheduler() { running_.store(false); for (auto t : workers_) { if (t.joinable()) t.join(); } } // 通知调度器有一个Actor有待处理的消息 void schedule(std::weak_ptractor_base actor_weak) { global_pending_actors_.enqueue(std::move(actor_weak)); } };这个调度器非常基础存在明显问题1) 全局队列是竞争热点2) 没有工作窃取3) 调度策略可能导致某些Actor饥饿。但它勾勒出了核心循环取Actor - 处理一条消息 - 根据情况重新调度。3.5 演员系统actor_system粘合一切actor_system是用户使用的入口点负责创建Actor和管理调度器。class actor_system { std::unique_ptrscheduler scheduler_; // 用于存储所有Actor的强引用防止其过早析构可选也可由用户管理 std::vectorstd::shared_ptractor_base actors_registry_; std::mutex registry_mutex_; public: actor_system(size_t scheduler_threads std::thread::hardware_concurrency()) : scheduler_(std::make_uniquescheduler(scheduler_threads)) {} // 创建Actor的模板函数 templatetypename ActorType, typename... Args std::shared_ptrActorType create_actor(Args... args) { static_assert(std::is_base_of_vactor_base, ActorType, ActorType must derive from actor_base); auto actor std::make_sharedActorType(std::forwardArgs(args)...); { std::lock_guardstd::mutex lock(registry_mutex_); actors_registry_.push_back(actor); } // 可以在这里执行Actor的启动初始化逻辑 return actor; } // 获取Actor的引用 templatetypename ActorType actor_ref get_ref(const std::shared_ptrActorType actor) { return actor_ref(actor); } // 发送消息并通知调度器供 actor_ref 内部调用或用户直接调用 void send_message(const actor_ref ref, message msg) { // 这里简化了实际应由 actor_ref 的 send 方法调用 system 的此方法 // 以实现通知调度器的逻辑 if (auto actor_ptr ref.actor_weak_ptr_.lock()) { actor_ptr-mailbox_.enqueue(std::move(msg)); scheduler_-schedule(std::weak_ptractor_base(actor_ptr)); } } // 关闭系统 void shutdown() { scheduler_.reset(); // 销毁调度器等待所有工作线程结束 // 清空Actor注册表 std::lock_guardstd::mutex lock(registry_mutex_); actors_registry_.clear(); } };4. 实战演练构建一个简单的Ping-Pong示例理论说再多不如跑个例子。我们来定义两个ActorPingActor和PongActor让它们互相发送消息。首先定义消息类型。在实际框架中我们通常用struct来定义消息。struct ping_msg { int count; }; struct pong_msg { int count; }; struct stop_msg {};然后实现PingActorclass ping_actor : public actor_base { actor_ref pong_ref_; int max_count_; int current_{0}; void on_message(const message msg) override { // 模式匹配检查消息类型并处理 if (auto* ping msg.get_ifping_msg()) { // 启动信号 std::cout Ping starting, count: ping-count std::endl; max_count_ ping-count; pong_ref_.send(pong_msg{current_}); } else if (auto* pong msg.get_ifpong_msg()) { std::cout Ping received pong, current: pong-count std::endl; current_ pong-count 1; if (current_ max_count_) { std::this_thread::sleep_for(std::chrono::milliseconds(100)); // 模拟工作 pong_ref_.send(pong_msg{current_}); } else { std::cout Ping finished. std::endl; pong_ref_.send(stop_msg{}); // 可以通知系统停止自己 } } else if (msg.get_ifstop_msg()) { std::cout Ping stopping. std::endl; stopped_.store(true); } // 忽略未知消息 } public: explicit ping_actor(actor_ref pong_ref) : pong_ref_(std::move(pong_ref)) {} };接着是PongActorclass pong_actor : public actor_base { actor_ref ping_ref_; void on_message(const message msg) override { if (auto* pong msg.get_ifpong_msg()) { std::cout Pong received ping, sending back, current: pong-count std::endl; std::this_thread::sleep_for(std::chrono::milliseconds(50)); ping_ref_.send(pong_msg{pong-count}); // 原样发回 } else if (msg.get_ifstop_msg()) { std::cout Pong stopping. std::endl; stopped_.store(true); } } public: explicit pong_actor(actor_ref ping_ref) : ping_ref_(std::move(ping_ref)) {} };最后在主函数中驱动整个系统int main() { actor_system system; // 先创建Pong再创建Ping并相互传递引用 auto pong system.create_actorpong_actor(actor_ref{}); // 临时空引用 auto ping system.create_actorping_actor(system.get_ref(pong)); // 为Pong设置正确的Ping引用 // 这里暴露了一个问题Actor构造时可能无法获得完整的依赖关系。 // 更好的模式是Actor在启动后通过消息接收其依赖的引用。 // 我们临时通过一个公共接口非Actor风格来设置仅作演示。 // 假设我们为pong_actor添加一个set_ping_ref方法非线程安全仅演示。 // 更Actor化的方式是让Ping发送一个“初始化”消息给Pong包含自己的引用。 // 启动Ping actor_ref ping_ref system.get_ref(ping); ping_ref.send(ping_msg{10}); // 让Ping-Pong进行10个回合 // 等待一段时间让消息处理完成 std::this_thread::sleep_for(std::chrono::seconds(3)); system.shutdown(); return 0; }运行这个程序你应该能看到Ping和Pong交替打印信息。这个简单的例子演示了消息定义、Actor实现、消息发送和响应的完整闭环。5. 性能调优与进阶思考一个玩具框架和工业级框架的差距往往就在性能和鲁棒性上。完成基础版本后你可以从以下几个方向进行深度优化和扩展1. 邮箱与调度器的极致优化无锁邮箱我们已经使用了moodycamel::ConcurrentQueue这很好。还可以考虑rigtorp::MPMCQueue或其他方案并针对单生产者单消费者SPSC场景进行特化优化因为一个Actor的邮箱通常只有一个生产者多个发送者和一个消费者调度器。工作窃取调度器实现每个工作线程一个本地双端队列。线程优先从自己队列的头部取任务LIFO缓存友好窃取时从其他队列的尾部取任务FIFO减少竞争。这能极大提升吞吐量。任务批处理调度器一次从Actor邮箱中取出多条消息例如32条进行处理可以减少调度开销。这需要修改process_one_message为process_messages(int batch_size)。2. 消息序列化与网络扩展目前的message只在进程内有效。要支持分布式Actor必须实现消息的序列化与反序列化。可以集成protobuf、flatbuffers或msgpack。message类需要增加serialize()和deserialize()方法并且每个消息类型都需要向系统注册序列化器。需要引入network_manager组件负责管理TCP/UDP连接将序列化的消息发送到远程节点并接收、反序列化、投递到本地对应的Actor邮箱。3. 生命周期与监控树Supervision Tree实现supervisorActor它负责监控一组子Actor。当子Actor崩溃抛出异常时监控策略决定如何处理重启子Actor、重启所有子Actor、上报给更高级的监控者或者直接放弃。这需要框架提供Actor链接link或监控monitor的API并在调度器的异常处理环节将崩溃事件通知给其监控者。4. 定时器与未来式Future支持定时器框架应提供schedule_once和schedule_repeated接口允许Actor在指定延迟后给自己或他人发送消息。这需要一个基于堆std::priority_queue的定时器队列由专门的线程或集成到调度器主循环中处理。Future/Promise提供同步或异步获取操作结果的机制。例如actor_ref.sendrequest().then([](response r) { ... })。这需要框架能够关联请求与响应通常通过一个唯一的request_id和每个Actor内部的一个std::unordered_maprequest_id, std::promiseresponse来实现。5. 背压Backpressure机制当一个Actor处理消息的速度远慢于接收速度时其邮箱可能爆满导致内存耗尽。需要实现背压策略当邮箱大小超过阈值时通知发送者减慢发送速率或者丢弃最旧的消息有损或者阻塞发送者无损但可能引发死锁。这是一个复杂但关键的生产级特性。6. 常见陷阱与调试技巧在开发和测试你自己的Actor框架时一定会遇到各种问题。以下是我踩过的一些坑和总结的技巧1. 循环引用导致的内存泄漏这是使用std::shared_ptr管理生命周期时最常见的陷阱。如果两个Actor互相持有对方的shared_ptr引用就会形成循环引用永远无法释放。解决方案严格使用actor_ref内部是weak_ptr作为Actor之间互相引用的类型。actor_system或创建者持有唯一的shared_ptr所有权。确保actor_ref是传递引用的唯一方式。2. 消息顺序保证Actor模型通常不保证消息的全局顺序但保证来自同一个发送者的消息到达同一个接收者的顺序如果使用TCP网络的话。在单机多线程调度下由于工作窃取来自不同发送者的消息处理顺序是不确定的。注意如果你的业务逻辑依赖特定顺序需要在应用层通过序列号或状态机来保证不能依赖框架的默认行为。3. 调试困难由于异步和非线性的执行流传统的断点调试可能让你抓狂。打印日志变得至关重要。技巧为每个Actor分配一个唯一的ID并在每条日志中输出[ActorID:ThreadID]。在message类中添加一个可选的trace_id字段用于追踪一个请求在整个系统中的流转路径。工具考虑集成像gperftools这样的性能分析工具查看调度器的负载是否均衡邮箱队列长度是否健康。4. 阻塞操作会“冻住”调度器如果在Actor的on_message函数中执行了阻塞IO如读取文件、网络请求那么处理该消息的线程会被阻塞无法处理其他Actor的消息严重降低系统吞吐。黄金法则Actor内部绝不能有阻塞操作。必须将阻塞操作异步化例如使用异步IO库如libuv、asio在操作完成后通过回调或发送消息的方式通知Actor。5. 启动与关闭的竞态条件在系统关闭时可能还有消息在飞驰或者Actor正在处理消息。粗暴地销毁所有对象会导致崩溃。优雅关闭actor_system::shutdown()应首先停止接受新消息然后等待所有邮箱为空且所有正在处理的消息完成最后再析构Actor和调度器。可以使用引用计数正在处理的消息数或阶段标志来实现。手写一个Actor框架就像亲手搭建一座并发编程的大厦。从地基消息、邮箱到骨架Actor、引用再到机电系统调度器最后进行精装修监控、网络、定时器。每一步都需要你对C内存模型、并发原语、数据结构和设计模式有深刻的理解。这个过程充满挑战但回报是巨大的你将不再畏惧构建任何复杂的并发系统并且对市面上其他并发框架的原理一目了然。这或许是一个C程序员向架构师迈进的最扎实的一步。
