写在前面

先上连接：AimRT官网
这里记录总结了我们在选择机器人软件架构时的思考过程。如果你用过 ROS，这些对比会帮你快速理解 AimRT 的设计理念。如果你没用过 ROS 也没关系，我们会从实际问题出发。

---

先说说我们遇到的问题

需求其实挺常见的：

• 一个足式机器人，要跑深度学习控制算法
• 深度学习模型在服务器上（PyTorch），机器人本体在工业电脑上
• 模型要 50Hz 交互，网络可能有线可能无线
• 开发在笔记本跑仿真，部署要能直接切到真机
• 团队有人写 C++（控制），有人写 Python（深度学习）
• 测试和部署的同学也得能轻松上手

这不是什么特别的需求，很多机器人项目都有类似场景。但当我们开始思考技术方案时，发现传统的做法都有些不够优雅。

---

机器人软件的核心问题：模块间怎么交流？

想想看，不管你用什么框架，机器人软件本质就是一堆模块在干活：

传感器 → 状态估计 → 路径规划 → 运动控制 → 电机驱动

每个模块从上游拿数据，自己算点东西，再把结果扔给下游。

所以核心问题就一个：这些模块之间怎么传数据？

ROS 的答案：Topic 和 Service

如果你用过 ROS，下面这些代码应该很眼熟：

# ROS 1 / ROS 2 的发布者
pub = rospy.Publisher('imu_data', Imu, queue_size=10)
pub.publish(imu_msg)

# ROS 1 / ROS 2 的订阅者
def callback(msg):
    print(msg.data)
rospy.Subscriber('imu_data', Imu, callback)

# ROS 1 / ROS 2 的服务
response = service_client.call(request)

ROS 用 Topic（话题）和 Service（服务）解决通信问题，配合 Message（消息文件）定义数据格式。这套设计很成熟，经过了大量项目验证。

AimRT 的答案：换个名字，概念相通

AimRT 采用了类似的思路，只是换了个名字：

ROS 概念	AimRT 概念	用途
Topic	Channel	发布-订阅，数据流
Service	RPC	请求-响应，服务调用
.msg 文件	Protobuf	消息定义

为啥要换名字？因为 AimRT 的定位不只是机器人框架——Channel/RPC 是分布式系统、微服务领域的通用概念，用这套术语能和更广阔的技术生态对话。

实际代码长这样：

// AimRT 的 Channel（类似 ROS Topic）
publisher_ = core.GetChannelHandle().GetPublisher("imu_data");
aimrt::channel::Publish(publisher_, imu_msg);

// AimRT 的 RPC（类似 ROS Service）
auto status = co_await proxy->Plan(ctx, request, response);

如果你熟悉 ROS，看到这些代码应该会觉得：“诶，就是换了个函数名嘛。” 对，上手很快，概念是通的。

---

但有个关键区别：底层是可以换的

ROS 的方式

你的代码 → ROS Topic/Service → DDS → 网络
                                 ↑
                            (绑定的，不可更换)

ROS 2 底层绑定了 DDS（Data Distribution Service）。你可以换不同的 DDS 实现（FastDDS、CycloneDDS 之类的），但整个架构就是走 DDS，这是定死的。

AimRT 的方式：插件化

你的代码 → Channel/RPC → Backend (可配置) → 网络
                            ↓
                    ├─ local (进程内)
                    ├─ zenoh (高性能网络)
                    ├─ tcp (传统 TCP)
                    ├─ mqtt (IoT 专用)
                    ├─ grpc (云服务)
                    └─ ros2 (兼容 ROS)

重点来了：Backend 由配置文件决定，代码完全不需要知道用的是哪个。

举个例子，你写了个状态发布模块：

class StateModule : public aimrt::ModuleBase {
  bool Initialize(aimrt::CoreRef core) {
    publisher_ = core.GetChannelHandle().GetPublisher("robot_state");
    aimrt::channel::RegisterPublishType<RobotState>(publisher_);
    return true;
  }
  
  void PublishState() {
    auto msg = std::make_shared<RobotState>();
    // 填充数据...
    
    aimrt::channel::Publish(publisher_, *msg);
    // 消息发出去了，但用什么方式发的？
  }
};

看注意最后一行注释：代码里看不出用什么方式发送消息。这是由配置文件决定的：

# 场景 1：单机开发 - 所有模块在一个进程
channel:
  backends:
    - type: local  # 进程内直接函数调用，延迟 < 100ns

# 场景 2：分布式部署 - 模块在不同进程/机器
channel:
  backends:
    - type: zenoh  # 网络传输，自动服务发现

# 场景 3：和 ROS 2 系统集成
channel:
  backends:
    - type: ros2  # 兼容 ROS 2 的 Topic

# 场景 4：同时支持进程内和网络
channel:
  backends:
    - type: local
    - type: zenoh
  pub_topics_options:
    - topic_name: "(.*)"
      enable_backends: [local, zenoh]  # 同时启用

最后一种配置特别妙：框架会自动判断订阅者在哪，同进程的走 local（纳秒级），跨机器的走 zenoh（网络），开发者完全不需要关心，代码一个字不用改，就改配置。 这就是"接口与实现分离"在实际工程里的体现。开发、测试、部署，三个环境三份配置，代码只写一次。

消息定义：Protobuf vs ROS msg

用过 ROS 的话，.msg 文件肯定写过不少：

# ROS 的 Imu.msg
Header header
geometry_msgs/Vector3 angular_velocity
geometry_msgs/Vector3 linear_acceleration

AimRT 用的是 Protobuf（Protocol Buffers），Google 开发的序列化协议：

// imu.proto
syntax = "proto3";

message ImuMsg {
  double timestamp = 1;
  Vector3 angular_velocity = 2;
  Vector3 linear_acceleration = 3;
}

message Vector3 {
  double x = 1;
  double y = 2;
  double z = 3;
}

为啥 AimRT 选 Protobuf？几个原因：

1. 跨语言支持更广：官方支持十几种语言，C++、Python、Java、Go 都能用
2. 性能确实好：序列化快，生成的代码也高效
3. 工业界在用：gRPC、TensorFlow 这些大项目都基于它，生态成熟
4. 向后兼容性好：可以安全地加新字段，老代码不会崩

对我们来说，最实在的好处是 C++ 和 Python 两边都好用，而且类型自动对齐。

---

Module、Pkg、配置：真·高度解耦

这块是我觉得 AimRT 设计最巧妙的地方——它把"写代码"和"运行系统"彻底分开了。

传统方式长啥样

// 你需要写 main 函数
int main(int argc, char** argv) {
  ros::init(argc, argv, "my_node");
  ros::NodeHandle nh;
  
  // 手动创建对象
  SensorDriver sensor;
  StateEstimator estimator;
  Controller controller;
  
  // 手动启动线程
  std::thread sensor_thread([&]() { sensor.run(); });
  std::thread control_thread([&]() { controller.run(); });
  
  ros::spin();
}

你得自己管对象创建、线程调度、启动顺序。改个模块？main 函数跟着一起改，非常麻烦，协作起来也容易出问题。

AimRT 怎么做？分三步走

第一步：写 Module逻辑

在 AimRT 里，Module 是逻辑上内聚的功能单元，生命周期很清晰：

class MyModule : public aimrt::ModuleBase {
 public:
  // 1. 框架会查询模块信息
  ModuleInfo Info() const override {
    return ModuleInfo{.name = "MyModule"};
  }
  
  // 2. 初始化阶段：注册 Publisher/Subscriber/Service
  bool Initialize(aimrt::CoreRef core) override {
    // 读取配置
    // 注册通信接口
    // 获取执行器
    return true;
  }
  
  // 3. 启动阶段：开始工作
  bool Start() override {
    // 启动循环
    // 开始接收数据
    return true;
  }
  
  // 4. 关闭阶段：清理资源
  void Shutdown() override {
    // 停止循环
    // 释放资源
  }
};

没有 main 函数，没有线程管理，没有启动逻辑， 你只管写业务代码。

第二步：打包成 Pkg

Pkg 是一个或多个 Module 的集合，是部署的单元，编译后成一个 .so 文件：

// pkg_main.cc
#include "aimrt_pkg_c_interface/pkg_macro.h"
#include "my_module.h"

static std::tuple<std::string_view, std::function<aimrt::ModuleBase*()>> 
aimrt_module_register_array[]{
    {"MyModule", []() { return new MyModule(); }}
};

AIMRT_PKG_MAIN(aimrt_module_register_array)

编译出来的 .so 文件可以：

• 复用到不同项目
• 独立管理版本
• 直接二进制发布（不需要源码）
• 符号隔离（不同 Pkg 可以用不同版本的库，互不影响）

第三步：写配置

代码写完了，怎么跑？配置文件决定：

aimrt:
  executor:
    executors:
      - name: work_thread_pool
        type: asio_thread
        options:
          thread_num: 4  # 框架自动管理线程池
  
  channel:
    backends:
      - type: local  # 框架自动选择通信方式
  
  module:
    pkgs:
      - path: ./libsensor_pkg.so
        enable_modules: [SensorModule]
      - path: ./libstate_pkg.so
        enable_modules: [StateEstimateModule]
      - path: ./libcontrol_pkg.so
        enable_modules: [ControlModule]

运行（永远就这一行）

./aimrt_main --cfg_file_path config.yaml

搞定。

这意味着什么？

• 不用写 main 函数（框架提供了 aimrt_main）
• 不用管线程（配置文件里声明执行器）
• 不用手动 new 对象（配置文件列 Pkg，框架自动加载）
• 不用操心启动顺序（框架按 Initialize → Start 统一调度）

工程师只管写 Module 的业务逻辑就行。框架负责把各个模块组装起来、跑起来。

这才是真正的"高度解耦"——你写你的算法，我管我的部署，井水不犯河水。

补充：AimRT 支持两种玩法：

App 模式：自己写 main，链接运行时库，编译成一个 exe

• 适合小工具、Demo
• 简单直接，但模块间会有依赖冲突

Pkg 模式（我们用的）：用框架的 aimrt_main，动态加载 .so

• 适合中大型项目
• 每个人写自己的 Module，编译成独立的 .so
• 符号隔离：张三的 Pkg 用 Eigen 3.3，李四的用 Eigen 3.4，互不影响
• 二进制发布：核心算法可以只发 .so，不暴露源码

在Pkg模式下，团队协作方便且可靠，每个人管好自己的一亩三分地，最后配置文件一组装，系统就跑起来了。

---

执行器：不用自己管线程了

机器人软件通常要跑好几个线程：控制循环、传感器读取、路径规划、日志记录…

传统做法？自己 std::thread：

// 传统做法
std::thread control_thread([this]() {
  while(running) {
    this->ControlLoop();
    std::this_thread::sleep_for(std::chrono::milliseconds(20));
  }
});

std::thread sensor_thread([this]() {
  while(running) {
    this->ReadSensors();
  }
});

// 要记得 join...
control_thread.join();
sensor_thread.join();

问题来了：

• 10 个模块就要 10 个线程？CPU 都在做上下文切换
• 很多线程其实大部分时间在睡觉，资源浪费
• 哪个线程卡住了？调试起来一团乱

AimRT 的做法：执行器统一管理

配置文件里声明：

executor:
  executors:
    # 控制线程池
    - name: control_pool
      type: asio_thread
      options:
        thread_num: 2
    
    # 计算线程池
    - name: compute_pool
      type: asio_thread
      options:
        thread_num: 8
    
    # 定时器
    - name: timer
      type: time_wheel
      options:
        bind_executor: control_pool

在代码里使用：

bool Initialize(aimrt::CoreRef core) {
  // 获取执行器（不需要自己创建线程）
  control_exec_ = core.GetExecutorManager().GetExecutor("control_pool");
  compute_exec_ = core.GetExecutorManager().GetExecutor("compute_pool");
  
  return true;
}

bool Start() {
  // 投递任务到执行器
  control_exec_.Execute([this]() { this->ControlLoop(); });
  compute_exec_.Execute([this]() { this->HeavyComputation(); });
  
  return true;
}

好处很明显：

1. 统一管理：框架管所有线程，日志里能看每个执行器的负载情况
2. 灵活调整：需要更多线程？改配置文件，代码不动
3. 资源复用：多个 Module 共享一个线程池，资源利用率高
4. 确定性调度：实时控制用单线程执行器，保证顺序执行不抖动

---

C++20：用上了新特性

AimRT 基于 C++20 开发，用上了一些很实用的新特性。

协程：异步代码终于能看了

以前写异步 RPC 是这样的：

// 传统回调方式
void CallService() {
  proxy->PlanAsync(request, [this](Response rsp) {
    // 回调 1
    proxy->ExecuteAsync(rsp, [this](Result result) {
      // 回调 2（嵌套）
      proxy->CheckAsync(result, [this](Status status) {
        // 回调 3（地狱）
        // ...
      });
    });
  });
}

现在用协程（C++20）：

co::Task<void> CallService() {
  // 看起来像同步代码，实际是异步的
  auto rsp = co_await proxy->Plan(ctx, request);
  auto result = co_await proxy->Execute(ctx, rsp);
  auto status = co_await proxy->Check(ctx, result);
  
  // 清清爽爽，一目了然
}

代码可读性瞬间上来了。特别是需要连续调几个 RPC 的时候，协程简直救命。

其他一些细节

// 用 std::chrono 管时间（不用自己算毫秒了）
executor.ExecuteAfter(std::chrono::milliseconds(100), task);

// 用 std::string_view 减少拷贝
std::string_view Name() const noexcept override { return "MyModule"; }

// 智能指针管生命周期
auto msg = std::make_shared<RobotState>();

都是 Modern C++ 的标准做法，代码安全性和效率都有保障。

---

日志系统：配置很灵活

调机器人，日志是命根子，AimRT 的日志系统用起来挺顺手。

log:
  core_lvl: Info  # 全局日志级别
  backends:
    # 控制台输出
    - type: console
      options:
        pattern: "[%c][%l][%n]%v"  # 时间、级别、模块名、消息
    
    # 滚动文件
    - type: rotate_file
      options:
        path: ./log
        filename: robot.log
        max_file_size_m: 100   # 单文件最大 100MB
        max_file_num: 10       # 保留 10 个文件
  
  module:
    modules:
      - name: ControlModule
        log_lvl: Debug  # 单独设置某个模块的日志级别
      - name: SensorModule
        log_lvl: Warn

实用的点：

• 每个模块可以单独设日志级别（调试时只看你关心的那个模块）
• 同时输出到控制台和文件
• 自动滚动（不会把磁盘塞爆）
• 格式自己定（开发要详细，部署要简洁）

代码里用起来也方便：

AIMRT_INFO("Robot initialized, version: {}", version);
AIMRT_WARN("Battery low: {}%", battery_level);
AIMRT_ERROR("Sensor timeout: {}", sensor_name);
AIMRT_DEBUG("Joint position: {}", position);  // Debug 级别才输出

支持格式化输出（像 Python 的 f-string），比 printf 好用，比 std::cout 高效。

---

和 ROS 2 能一起用吗？能

首先说明：AimRT 不是来替代 ROS 2 的。

ROS 2 的生态很丰富（导航、SLAM、感知这些都有成熟的包），放着不用太浪费。好在 AimRT 提供了 ros2_plugin，可以和 ROS 2 互通：

channel:
  backends:
    - type: local   # 内部用 local
    - type: ros2    # 对外用 ROS 2 Topic

这样的话：

• 核心控制用 AimRT（要精确控制）
• 外围感知用 ROS 2 包（成熟方案拿来就用）
• 两边通过 Channel 互通

不是非此即彼，而是各取所长。

---

总结一下

选择技术框架是个工程决策，需要权衡很多因素：学习成本、生态成熟度、性能要求、团队技能栈等等。

我们选 AimRT，主要是因为：

• 模块化设计很清爽：Module/Pkg 分层明确
• 配置驱动：改部署方式不用动代码，部署灵活性高
• 通信灵活：可插拔的后端系统
• 现代化：C++20、协程、配置化日志与执行器
• 靠谱：智元做背书，更新很及时

---

这篇主要讲了 AimRT 的核心概念和设计思路，后面可能会写怎么选通信插件、怎么对接深度学习模型、性能调优这些实战内容。