AimRT 从零到一：官方示例精讲 —— 六、pb_chn示例.md

pb_chn示例

官方仓库：pb_chn

这个官方示例展示了如何基于 protobuf 协议和 local 后端实现 channel 通信。主要包括四种场景：

1. 基础示例：分别用两个模块（Publisher 和 Subscriber）发布和订阅 protobuf 消息，使用 local channel 通信。
2. 单 Pkg 集成：与基础示例类似，但将发布和订阅模块集成到同一个 Pkg 运行。
3. Publisher App 模式：以 App 模式直接运行 Publisher，同时以 Module 方式运行 Subscriber。
4. Subscriber App 模式：以 App 模式直接运行 Subscriber，同时以 Module 方式运行 Publisher。

需要先阅读一下官方文档的相关内容：Channel 章节

这里我们没有复现​​benchmark​​两个模块，只关注通信相关内容。

配置文件(configuration_protobuf_channel.yaml)

依据官方示例项目结构自行编写YAML配置文件：

# ============== 基础信息 ==============
base_info:project_name: Protobuf_channel # 项目名称build_mode_tags: ["EXAMPLE", "SIMULATION", "TEST_CAMERA"] # 构建模式标签aimrt_import_options: # AimRT 框架构建选项AIMRT_BUILD_WITH_PROTOBUF: "ON" # 启用 Protobuf 支持（必须开启）AIMRT_BUILD_RUNTIME: "ON" # 启用模块化运行支持（必须开启）AIMRT_BUILD_TESTS: "OFF" # 关闭测试模块构建AIMRT_BUILD_EXAMPLES: "OFF" # 关闭示例构建（实际跑示例需打开）AIMRT_BUILD_DOCUMENT: "OFF" # 关闭文档构建AIMRT_BUILD_CLI_TOOLS: "OFF" # 关闭命令行工具AIMRT_USE_LOCAL_PROTOC_COMPILER: "OFF" # 使用内置 Protoc 编译器AIMRT_BUILD_WITH_ROS2: "OFF" # 关闭 ROS2 支持AIMRT_BUILD_NET_PLUGIN: "OFF" # 关闭网络插件AIMRT_BUILD_ROS2_PLUGIN: "OFF" # 关闭 ROS2 插件# ============== 协议配置 ==============
protocols:- name: my_prototype: protobufoptions:xxx: xxx # 预留可扩展选项# 以下为注释掉的备用协议示例：# - name: my_ros2_proto#   type: ros2#   options:#     zzz: zzz#   build_mode_tag: ["EXAMPLE"]## - name: example_proto#   type: protobuf#   build_mode_tag: ["EXAMPLE"] # 仅在 EXAMPLE 模式启用# ============== 模块定义 ==============
modules:- name: normal_publisher_module # 定期发布 ExampleEventMsg 消息- name: normal_subscriber_module # 订阅并处理 ExampleEventMsg 消息# ============== 包组合配置 ==============
pkgs:- name: protobuf_channel_pkg # 单包整合示例modules:- name: normal_publisher_module- name: normal_subscriber_module- name: protobuf_channel_pub_pkg # 发布模块包（仅发布者）modules:- name: normal_publisher_module- name: protobuf_channel_sub_pkg # 订阅模块包（仅订阅者）modules:- name: normal_subscriber_module# ============== 部署方案 ==============
deploy_modes:- name: local_deploy # 本地部署方案deploy_ins:- name: local_ins_protobuf_channel # 双包部署（发布+订阅分开）pkgs:- name: protobuf_channel_pub_pkg- name: protobuf_channel_sub_pkg- name: local_ins_protobuf_channel_single_pkg # 单包部署（发布+订阅合一）pkgs:- name: protobuf_channel_pkg## ========================================================
## 官方工具似乎不支持 App 模式，因此相关文件需要手动配置。
## 发布配置主要影响后续生成的 cfg 文件和启动脚本。
## 可以将以下内容添加到配置文件中，但仍需手动调整。
## ========================================================
# - name: local_ins_protobuf_channel_publisher_app # app发布
#   pkgs:
#     - name: protobuf_channel_sub_pkg # 订阅模块包，用于订阅app发布的消息# - name: local_ins_protobuf_channel_subscriber_app # app订阅
#   pkgs:
#     - name: protobuf_channel_pub_pkg # 发布模块包，用于向app发布消息

protocols目录

这是配置文件中 协议 部分生成的目录，管理通信协议相关的文件

protocols/
└── my_proto├── CMakeLists.txt└── my_proto.proto

my_proto.proto文件

用来定义通信的消息类型

syntax = "proto3";package Protobuf_channel.my_proto;	// 命名空间：Protobuf_channel::my_proto// 在这里定义您的消息
message ExampleEventMsg {string msg = 1;int32 num = 2;
}

CMakeLists.txt文件

AimRT框架封装了从Protobuf生成c++代码的操作，只需要使用提供的CMake 方法即可。相关内容见官方文档：Protobuf封装

# Get the current folder name
string(REGEX REPLACE ".*/\(.*\)" "\\1" CUR_DIR ${CMAKE_CURRENT_SOURCE_DIR})# Get namespace
get_namespace(CUR_SUPERIOR_NAMESPACE)
string(REPLACE "::" "_" CUR_SUPERIOR_NAMESPACE_UNDERLINE ${CUR_SUPERIOR_NAMESPACE})# Set target name
set(CUR_TARGET_NAME ${CUR_SUPERIOR_NAMESPACE_UNDERLINE}_${CUR_DIR})
set(CUR_TARGET_ALIAS_NAME ${CUR_SUPERIOR_NAMESPACE}::${CUR_DIR})# Add target
add_protobuf_gencode_target_for_proto_path(TARGET_NAME ${CUR_TARGET_NAME}_pb_gencodePROTO_PATH ${CMAKE_CURRENT_SOURCE_DIR}GENCODE_PATH ${CMAKE_CURRENT_BINARY_DIR})
add_library(${CUR_TARGET_ALIAS_NAME}_pb_gencode ALIAS ${CUR_TARGET_NAME}_pb_gencode)# Set installation
install(DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR}DESTINATION "share"FILES_MATCHINGPATTERN "*.proto")set_property(TARGET ${CUR_TARGET_NAME}_pb_gencode PROPERTY EXPORT_NAME ${CUR_TARGET_ALIAS_NAME}_pb_gencode)
install(TARGETS ${CUR_TARGET_NAME}_pb_gencodeEXPORT ${INSTALL_CONFIG_NAME}ARCHIVE DESTINATION libFILE_SET HEADERSDESTINATION include/${CUR_TARGET_NAME}_pb_gencode)

最终生成的Target需要链接到模块，可以在Cmake文件最后添加下面的内容，查看Target名称

# 打印最终Target名称
message(STATUS "最终Target名称 - CUR_TARGET_NAME: ${CUR_TARGET_NAME}_pb_gencode")

module目录

normal_publisher_module

链接生成的消息文件(CMakeLists.txt​)

# 略# Set link libraries of target
target_link_libraries(${CUR_TARGET_NAME}PRIVATE yaml-cpp::yaml-cppPUBLIC aimrt::interface::aimrt_module_cpp_interfacePUBLIC Protobuf_channel_my_proto_pb_gencode)  # 在这里进行链接

模块定义(normal_publisher_module.h​)

#pragma once
#include <atomic>
#include <future>#include "aimrt_module_cpp_interface/module_base.h"namespace Protobuf_channel::normal_publisher_module {
using namespace aimrt;/*** @brief 普通发布者模块类** 该模块用于定期发布Protobuf格式的消息到指定主题*/
class NormalPublisherModule : public aimrt::ModuleBase {public:NormalPublisherModule() = default;~NormalPublisherModule() override = default;// 获取模块基本信息ModuleInfo Info() const override {return ModuleInfo{.name = "NormalPublisherModule"};}bool Initialize(aimrt::CoreRef core) override;bool Start() override;void Shutdown() override;private:// 获取日志记录器auto GetLogger() { return core_.GetLogger(); }// 主工作循环void MainLoop();private:aimrt::CoreRef core_;                    // 核心系统引用aimrt::executor::ExecutorRef executor_;  // 任务执行器引用std::atomic_bool run_flag_ = false;  // 运行状态标志std::promise<void> stop_sig_;        // 停止信号量std::string topic_name_ = "test_topic";   // 发布主题名称double channel_frq_ = 0.5;                // 发布频率(Hz)aimrt::channel::PublisherRef publisher_;  // 消息发布器引用
};}  // namespace Protobuf_channel::normal_publisher_module

• 定期发布Protobuf格式的消息到指定主题

模块实现(normal_publisher_module.cc​)

初始化阶段

bool NormalPublisherModule::Initialize(aimrt::CoreRef core) {core_ = core;  // 保存 CoreRef，后续模块操作需要用到try {// === Step 1: 从配置文件读取 topic 和发布频率 ===auto file_path = core_.GetConfigurator().GetConfigFilePath();if (!file_path.empty()) {YAML::Node cfg_node = YAML::LoadFile(std::string(file_path));topic_name_ = cfg_node["topic_name"].as<std::string>();channel_frq_ = cfg_node["channel_frq"].as<double>();}// === Step 2: 获取执行器 ExecutorRef（用来安排任务执行） ===executor_ = core_.GetExecutorManager().GetExecutor("work_thread_pool");AIMRT_CHECK_ERROR_THROW(executor_ && executor_.SupportTimerSchedule(),"获取执行器'work_thread_pool'失败");// ➔ Note: Executor 是 AimRT// 中安排异步任务的组件，类似“线程池”或“定时器调度器”// === Step 3: 获取 PublisherRef（即通道的发布句柄） ===publisher_ = core_.GetChannelHandle().GetPublisher(topic_name_);AIMRT_CHECK_ERROR_THROW(publisher_, "获取主题'{}'的发布者失败",topic_name_);// ➔ Note: PublisherRef 是发布消息用的关键对象，相当于“话题写入器”// === Step 4: 注册消息类型（必须要做！）===bool ret = aimrt::channel::RegisterPublishType<Protobuf_channel::my_proto::ExampleEventMsg>(publisher_);AIMRT_CHECK_ERROR_THROW(ret, "注册发布类型失败");// ➔ Note: 必须告诉系统：这个 Publisher 上发送的是 ExampleEventMsg 类型的// Protobuf 消息。} catch (const std::exception& e) {AIMRT_ERROR("初始化失败, 错误信息: {}", e.what());return false;}AIMRT_INFO("模块初始化成功");return true;
}

• 获取了配置文件中的相关配置

• 获取了执行器executor_​

• 获取了消息发布者publisher_​

• 对消息类型进行注册

运行阶段

bool NormalPublisherModule::Start() {try {run_flag_ = true;  // 设置运行标志，控制主循环// === 启动主循环任务 ===executor_.Execute(std::bind(&NormalPublisherModule::MainLoop, this));// ➔ Note: 这里没有直接启动线程，而是通过 executor 异步调度执行// MainLoop（即委托给了 AimRT 的调度器管理）} catch (const std::exception& e) {AIMRT_ERROR("模块启动失败, 错误信息: {}", e.what());return false;}AIMRT_INFO("模块启动成功");return true;
}void NormalPublisherModule::MainLoop() {try {AIMRT_INFO("主工作循环开始运行");// === 创建 PublisherProxy，简化消息发布过程 ===aimrt::channel::PublisherProxy<Protobuf_channel::my_proto::ExampleEventMsg>publisher_proxy(publisher_);// ➔ Note: 绑定 publisher_，后面发布时只需要传 msg，不用每次带 PublisherRef// ➔ Note: Proxy 还能管理 Context，更高级uint32_t count = 0;while (run_flag_) {// === 按配置的发布频率休眠 ===std::this_thread::sleep_for(std::chrono::milliseconds(static_cast<uint32_t>(1000 / channel_frq_)));count++;AIMRT_INFO("发布循环计数: {}", count);// === 构造并填充消息 ===Protobuf_channel::my_proto::ExampleEventMsg msg;msg.set_msg("当前计数: " + std::to_string(count));  // 设置文本字段msg.set_num(count);                                 // 设置数字字段// === 发布消息 ===AIMRT_INFO("发布新消息, 内容: {}", aimrt::Pb2CompactJson(msg));publisher_proxy.Publish(msg);// ➔ Note: Publish是异步的，实际发送取决于通道后端的实现// ➔ Note: 如果后端阻塞，可能导致 Publish 耗时不稳定，但一般很快返回}AIMRT_INFO("主工作循环正常退出");} catch (const std::exception& e) {AIMRT_ERROR("主工作循环异常退出, 错误信息: {}", e.what());}// 通知 Shutdown 主线程，主循环已经结束stop_sig_.set_value();
}

• 通过执行器绑定MainLoop​函数

• ​MainLoop​函数首先创建了一个发布代理publisher_proxy​，用来简化消息发布操作（绑定 publisher_，后面发布时只需要传 msg，不用每次带 PublisherRef）

• 构造并填充消息，进行发布

停止阶段

void NormalPublisherModule::Shutdown() {try {if (run_flag_) {run_flag_ = false;              // 通知主循环退出stop_sig_.get_future().wait();  // 等待 MainLoop 退出信号}} catch (const std::exception& e) {AIMRT_ERROR("模块关闭失败, 错误信息: {}", e.what());return;}AIMRT_INFO("模块已正常关闭");
}

• 通过信号量通知进行优雅退出

normal_subscriber_module

链接生成的消息文件(CMakeLists.txt​)

# 略# Set link libraries of target
target_link_libraries(${CUR_TARGET_NAME}PRIVATE yaml-cpp::yaml-cppPUBLIC aimrt::interface::aimrt_module_cpp_interfacePUBLIC Protobuf_channel_my_proto_pb_gencode)	# 链接生成的消息文件

模块定义(normal_subscriber_module.h​)

#pragma once#include "aimrt_module_cpp_interface/module_base.h"
#include "my_proto.pb.h"  // 自定义Protobuf消息头文件namespace Protobuf_channel::normal_subscriber_module {
using namespace aimrt;/*** @brief 普通订阅者模块类** 该模块用于订阅并处理Protobuf格式的消息*/
class NormalSubscriberModule : public aimrt::ModuleBase {public:NormalSubscriberModule() = default;~NormalSubscriberModule() override = default;// 获取模块基本信息ModuleInfo Info() const override {return ModuleInfo{.name = "NormalSubscriberModule"};}bool Initialize(aimrt::CoreRef core) override;  // 初始化模块bool Start() override;                          // 启动模块void Shutdown() override;                       // 关闭模块private:// 获取日志记录器auto GetLogger() { return core_.GetLogger(); }/*** @brief 事件处理回调函数* @param ctx 消息上下文* @param data 接收到的消息数据*/void EventHandle(aimrt::channel::ContextRefctx,  // 上下文信息是由AimRT系统框架传递的，在编译期确定const std::shared_ptr<const Protobuf_channel::my_proto::ExampleEventMsg>&data);private:aimrt::CoreRef core_;                       // 核心系统引用std::string topic_name_ = "test_topic";     // 订阅主题名称aimrt::channel::SubscriberRef subscriber_;  // 消息订阅者引用
};}  // namespace Protobuf_channel::normal_subscriber_module

模块实现(normal_subscriber_module.cc​)

初始化阶段

bool NormalSubscriberModule::Initialize(aimrt::CoreRef core) {core_ = core;  // 保存 CoreRef，后续操作模块需要用到try {// === Step 1: 读取配置文件，拿到订阅的话题名 ===auto file_path = core_.GetConfigurator().GetConfigFilePath();if (!file_path.empty()) {YAML::Node cfg_node = YAML::LoadFile(std::string(file_path));topic_name_ = cfg_node["topic_name"].as<std::string>();}// === Step 2: 获取SubscriberRef，准备订阅主题 ===subscriber_ = core_.GetChannelHandle().GetSubscriber(topic_name_);AIMRT_CHECK_ERROR_THROW(subscriber_, "获取主题'{}'的订阅者失败",topic_name_);// === Step 3: 注册消息订阅回调 ===bool ret =aimrt::channel::Subscribe<Protobuf_channel::my_proto::ExampleEventMsg>(subscriber_,std::bind(&NormalSubscriberModule::EventHandle, this,std::placeholders::_1, std::placeholders::_2));AIMRT_CHECK_ERROR_THROW(ret, "订阅消息失败");// ➔ Note: 这里使用的是【函数风格】【带 Context// 参数】【智能指针回调】的订阅形式// ➔ Note: 必须在 Initialize阶段完成订阅，运行中不允许动态订阅} catch (const std::exception& e) {AIMRT_ERROR("初始化失败，错误信息: {}", e.what());return false;}AIMRT_INFO("模块初始化成功");return true;
}void NormalSubscriberModule::EventHandle(aimrt::channel::ContextRef ctx,const std::shared_ptr<const Protobuf_channel::my_proto::ExampleEventMsg>&data) {// === Step 4: 处理收到的新消息 ===AIMRT_INFO("收到新消息，上下文: {}, 数据内容: {}", ctx.ToString(),aimrt::Pb2CompactJson(*data));// ➔ Note: ContextRef// 提供了消息链路、来源、发送时间等上下文信息，可以根据需要提取// ➔ Note: data 是 Protobuf 消息的【const智能指针】，直接解引用使用即可
}

• 获取配置文件中的相关配置
• 获取订阅者subscriber_​
• 注册消息订阅回调EventHandle​函数
• 注意点：可以看到回调函数会接收一个​​ctx​​上下文参数，而我们编写的发布者并没有看到这个数据。
• 这个上下文参数是由AimRT框架自动传入的，也可以只保留​​data​​一个参数。这一过程是在编译期确定的

运行阶段

bool NormalSubscriberModule::Start() {// 当前 Start 阶段无特定动作，AimRT 框架会自动管理订阅回调的调度return true;
}

停止阶段

void NormalSubscriberModule::Shutdown() {// 当前 Shutdown 阶段无特定动作，SubscriberRef 生命周期随模块销毁而结束
}

启动Pkg模式通信示例

protobuf channel

一个最基本的、基于 protobuf 协议与 local 后端的 channel 示例，演示内容包括：

• 如何使用 protobuf 协议作为 channel 传输协议；
• 如何基于 Module 方式使用 Executor、Channel publish 和 subscribe 功能；
• 如何使用 local 类型的 channel 后端；
• 如何以 Pkg 模式集成 Module 并启动；

对应配置文件(cfg/local_deploy_local_ins_protobuf_channel_cfg.yaml​)

aimrt:log:core_lvl: INFO # Trace/Debug/Info/Warn/Error/Fatal/Offbackends:- type: consoleexecutor:executors:- name: work_thread_pooltype: asio_threadoptions:thread_num: 4# 消息通道相关配置官方文档：https://docs.aimrt.org/v0.10.0/tutorials/cfg/channel.html#aimrt-channelchannel:backends:- type: local # local类型的 Channel 后端是 AimRT 官方提供的一种 Channel 后端，用于将消息发布到同进程中的其他模块，它会自动判断发布端和订阅端是否在同一个Pkg内，从而采用各种方式进行性能的优化options:subscriber_use_inline_executor: false # 如果配置为true，则直接使用发布端的执行器来执行订阅端的回调，会阻塞发布端的 Publish 方法直到所有的订阅端回调都执行完成subscriber_executor: work_thread_pool # 仅在subscriber_use_inline_executor为false时生效，后端会将订阅端的回调都投递进此执行器中异步执行，发布端的 Publish 方法会立即返回pub_topics_options:- topic_name: "(.*)"enable_backends: [local]sub_topics_options:- topic_name: "(.*)"enable_backends: [local]module:pkgs:- path: ./libprotobuf_channel_pub_pkg.so- path: ./libprotobuf_channel_sub_pkg.somodules:- name: NormalPublisherModulelog_lvl: INFO- name: NormalSubscriberModulelog_lvl: INFO# 模块自定义配置
NormalPublisherModule:topic_name: test_topic # topic名称channel_frq: 0.5 # 0.5HzNormalSubscriberModule:topic_name: test_topic

消息通道相关配置官方文档：配置文件-通道

说明：

• 此示例创建了以下两个模块：

  • ​NormalPublisherModule​：会基于 work_thread_pool​ 执行器，以配置的频率、向配置的 topic 中发布 ExampleEventMsg​ 类型的消息；
  • ​NormalSubscriberModule​：会订阅配置的 topic 下的 ExampleEventMsg​ 类型的消息；

• 此示例将 NormalPublisherModule​ 和 NormalSubscriberModule​ 分别集成到 pb_chn_pub_pkg​ 和 pb_chn_sub_pkg​ 两个 Pkg 中，并在配置文件中加载这两个 Pkg 到一个 AimRT 进程中；

• 此示例使用 local 类型的 channel 后端进行通信；

以pkg模式启动

启动命令为：

./aimrt_main --cfg_file_path=./cfg/local_deploy_local_ins_protobuf_channel_cfg.yaml

如果是通过aimrt_cli​工具生成代码框架，则执行start_local_deploy_local_ins_protobuf_channel.sh​脚本

protobuf channel single pkg

一个最基本的、基于 protobuf 协议与 local 后端的 channel 示例，演示内容包括：

• 如何使用 protobuf 协议作为 channel 传输协议；
• 如何基于 Module 方式使用 Executor、Channel publish 和 subscribe 功能；
• 如何使用 local 类型的 channel 后端；
• 如何以 Pkg 模式集成 Module 并启动；

对应配置文件(cfg/local_deploy_local_ins_protobuf_channel_single_pkg_cfg.yaml​)

aimrt:log:core_lvl: INFO # Trace/Debug/Info/Warn/Error/Fatal/Offbackends:- type: consoleexecutor:executors:- name: work_thread_pooltype: asio_threadoptions:thread_num: 4channel:backends:- type: localoptions:subscriber_use_inline_executor: falsesubscriber_executor: work_thread_poolpub_topics_options:- topic_name: "(.*)"enable_backends: [local]sub_topics_options:- topic_name: "(.*)"enable_backends: [local]module:pkgs:- path: ./libprotobuf_channel_pkg.sodisable_modules: []modules:- name: NormalPublisherModulelog_lvl: INFO- name: NormalSubscriberModulelog_lvl: INFO# Module custom configuration
NormalPublisherModule:topic_name: test_topicchannel_frq: 0.5NormalSubscriberModule:topic_name: test_topic

说明：

• 此示例与 protobuf channel 示例基本一致，唯一的区别是将 NormalPublisherModule​ 和 NormalSubscriberModule​ 集成到 protobuf_channel_pkg​ 一个 Pkg 中；

以pkg模式启动

启动命令为：

./aimrt_main --cfg_file_path=./cfg/local_deploy_local_ins_protobuf_channel_single_pkg_cfg.yaml

如果是通过aimrt_cli​工具生成代码框架，则执行start_local_deploy_local_ins_protobuf_channel_single_pkg.sh​脚本

App目录

normal_pb_chn_publisher_app

链接Protobuf生成的消息文件(CMakeLists.txt​)

# 略target_link_libraries(${CUR_TARGET_NAME}PRIVATE gflags::gflagsaimrt::runtime::coreaimrt::interface::aimrt_module_protobuf_interfaceProtobuf_channel_my_proto_pb_gencode)

主函数(main.cpp​)

#include <csignal>   // 信号处理
#include <iostream>  // 标准输入输出// 包含AIMRT框架相关头文件
#include "aimrt_module_cpp_interface/core.h"
#include "aimrt_module_protobuf_interface/channel/protobuf_channel.h"
#include "aimrt_module_protobuf_interface/util/protobuf_tools.h"
#include "core/aimrt_core.h"
#include "my_proto.pb.h"  // Protobuf生成的消息定义using namespace aimrt::runtime::core;  // 使用AIMRT核心命名空间bool run_flag = true;  // 全局运行标志// 信号处理函数
void SignalHandler(int sig) {if (sig == SIGINT || sig == SIGTERM) {  // 处理中断和终止信号run_flag = false;                     // 设置运行标志为false以优雅退出return;}raise(sig);  // 其他信号原样抛出
};// 主函数
int32_t main(int32_t argc, char** argv) {// 注册信号处理函数signal(SIGINT, SignalHandler);signal(SIGTERM, SignalHandler);std::cout << "AimRT 启动." << std::endl;try {AimRTCore core;  // 创建AIMRT核心对象// 初始化配置AimRTCore::Options options;if (argc > 1)options.cfg_file_path = argv[1];  // 从命令行参数获取配置文件路径core.Initialize(options);  // 初始化核心// 创建模块aimrt::CoreRef module_handle(core.GetModuleManager().CreateModule("NormalPublisherModule"));  // 创建发布者模块std::string topic_name = "test_topic";  // 默认主题名double channel_frq = 0.5;               // 默认发布频率(Hz)// 读取配置文件auto file_path = module_handle.GetConfigurator().GetConfigFilePath();if (!file_path.empty()) {YAML::Node cfg_node =YAML::LoadFile(std::string(file_path));             // 加载YAML配置topic_name = cfg_node["topic_name"].as<std::string>();  // 获取主题名配置channel_frq = cfg_node["channel_frq"].as<double>();  // 获取发布频率配置}// 注册发布类型auto publisher = module_handle.GetChannelHandle().GetPublisher(topic_name);  // 获取发布者// 检查发布者是否有效(失败会抛出异常)AIMRT_HL_CHECK_ERROR_THROW(module_handle.GetLogger(), publisher,"获取主题'{}'的发布者失败", topic_name);// 创建发布者代理(用于发布Protobuf消息)aimrt::channel::PublisherProxy<Protobuf_channel::my_proto::ExampleEventMsg>publisher_proxy(publisher);bool ret = publisher_proxy.RegisterPublishType();  // 注册发布类型AIMRT_HL_CHECK_ERROR_THROW(module_handle.GetLogger(), ret,"注册发布类型失败");// 启动核心(异步)auto fu = core.AsyncStart();// 发布消息循环uint32_t count = 0;while (run_flag) {// 按配置频率休眠std::this_thread::sleep_for(std::chrono::milliseconds(static_cast<uint32_t>(1000 / channel_frq)));count++;AIMRT_HL_INFO(module_handle.GetLogger(),"循环计数 : {} -------------------------", count);// 构造并发布Protobuf消息Protobuf_channel::my_proto::ExampleEventMsg msg;msg.set_msg("计数: " + std::to_string(count));  // 设置消息内容msg.set_num(count);                             // 设置消息编号// 记录日志并发布AIMRT_HL_INFO(module_handle.GetLogger(), "发布新的Protobuf事件, 数据: {}",aimrt::Pb2CompactJson(msg));  // 将Protobuf转为JSON格式日志publisher_proxy.Publish(msg);               // 发布消息}// 优雅关闭core.Shutdown();  // 关闭核心fu.wait();        // 等待异步操作完成} catch (const std::exception& e) {std::cout << "AimRT 运行异常并退出: " << e.what() << std::endl;return -1;  // 异常退出}std::cout << "AimRT 退出." << std::endl;  // 正常退出日志return 0;
}

• 基于 App 模式创建模块的方式使用 Channel publish 功能

• 创建 NormalPublisherModule 模块，获取 CoreRef 句柄，以配置的频率、向配置的 topic 中发布 ExampleEventMsg 类型的消息

normal_pb_chn_subscriber_app

链接Protobuf生成的消息文件(CMakeLists.txt​)

target_link_libraries(${CUR_TARGET_NAME}PRIVATE gflags::gflagsaimrt::runtime::coreaimrt::interface::aimrt_module_protobuf_interfaceProtobuf_channel_my_proto_pb_gencode)

主函数(main.cpp​)

#include <csignal>
#include <iostream>#include "aimrt_module_cpp_interface/core.h"
#include "aimrt_module_protobuf_interface/channel/protobuf_channel.h"
#include "aimrt_module_protobuf_interface/util/protobuf_tools.h"
#include "core/aimrt_core.h"
#include "my_proto.pb.h"using namespace aimrt::runtime::core;// 全局核心指针，用于信号处理
AimRTCore* global_core_ptr = nullptr;/*** @brief 信号处理函数* @param sig 接收到的信号*/
void SignalHandler(int sig) {// 如果是SIGINT(中断)或SIGTERM(终止)信号，且全局核心指针有效，则优雅关闭if (global_core_ptr && (sig == SIGINT || sig == SIGTERM)) {global_core_ptr->Shutdown();return;}// 其他信号继续传递raise(sig);
};/*** @brief 主函数*/
int32_t main(int32_t argc, char** argv) {// 注册信号处理函数signal(SIGINT, SignalHandler);signal(SIGTERM, SignalHandler);std::cout << "AimRT 启动中..." << std::endl;try {// 创建核心对象AimRTCore core;global_core_ptr = &core;// 初始化配置AimRTCore::Options options;if (argc > 1) {options.cfg_file_path = argv[1];  // 从命令行参数获取配置文件路径}core.Initialize(options);// 创建模块aimrt::CoreRef module_handle(core.GetModuleManager().CreateModule("NormalSubscriberModule"));std::string topic_name = "test_topic";  // 默认主题名// 从配置文件读取主题名auto file_path = module_handle.GetConfigurator().GetConfigFilePath();if (!file_path.empty()) {YAML::Node cfg_node = YAML::LoadFile(std::string(file_path));topic_name = cfg_node["topic_name"].as<std::string>();}// 获取订阅者auto subscriber =module_handle.GetChannelHandle().GetSubscriber(topic_name);AIMRT_HL_CHECK_ERROR_THROW(module_handle.GetLogger(), subscriber,"获取主题 '{}' 的订阅者失败", topic_name);// 订阅事件bool ret =aimrt::channel::Subscribe<Protobuf_channel::my_proto::ExampleEventMsg>(subscriber,[module_handle](const std::shared_ptr<const Protobuf_channel::my_proto::ExampleEventMsg>& data) {// 收到消息时的回调函数AIMRT_HL_INFO(module_handle.GetLogger(),"收到新的PB协议事件消息, 内容: {}",aimrt::Pb2CompactJson(*data));});AIMRT_HL_CHECK_ERROR_THROW(module_handle.GetLogger(), ret, "订阅失败");// 启动核心core.Start();// 关闭核心core.Shutdown();global_core_ptr = nullptr;  // 清除全局指针} catch (const std::exception& e) {std::cout << "AimRT 运行时发生异常并退出: " << e.what() << std::endl;return -1;}std::cout << "AimRT 正常退出" << std::endl;return 0;
}

• 基于 App 模式创建模块的方式使用 Channel subscribe 功能
• 创建 NormalSubscriberModule 模块，获取 CoreRef 句柄，会订阅配置的 topic 下的 ExampleEventMsg 类型的消息

启动APP模式通信示例

protobuf channel publisher app

一个基于 protobuf 协议与 local 后端的 channel 示例，演示内容包括：

• 如何使用 protobuf 协议作为 channel 传输协议；
• 如何基于 App 模式创建模块的方式使用 Channel publish 功能；
• 如何基于 Module 方式使用 Channel subscribe 功能；
• 如何使用 local 类型的 channel 后端；
• 如何以 App 模式启动；

配置文件(cfg/local_deploy_local_ins_protobuf_channel_publisher_app_cfg.yaml​)

aimrt:log:core_lvl: INFO # Trace/Debug/Info/Warn/Error/Fatal/Offbackends:- type: consoleexecutor:executors:- name: work_thread_pooltype: asio_threadoptions:thread_num: 4channel:backends:- type: localoptions:subscriber_use_inline_executor: falsesubscriber_executor: work_thread_poolpub_topics_options:- topic_name: "(.*)"enable_backends: [local]sub_topics_options:- topic_name: "(.*)"enable_backends: [local]module:pkgs:- path: ./libprotobuf_channel_sub_pkg.so # 订阅方使用的是之前以module模式编写的包enable_modules: []modules:- name: NormalPublisherModule # 启动发布模块 (App模式编写)log_lvl: INFO- name: NormalSubscriberModule # 启动订阅模块 (pkg模式编写)log_lvl: INFO# Module custom configuration
NormalPublisherModule:topic_name: test_topicchannel_frq: 0.5NormalSubscriberModule:topic_name: test_topic

以app模式启动

启动命令为：

./normal_pb_chn_publisher_app ./cfg/local_deploy_local_ins_protobuf_channel_publisher_app_cfg.yaml

**注意：**​aim_cli​工具不支持生成App模式代码框架，因此不会有相应的启动脚本，需要自行编写。

• 可以在生成代码框架时，先使用pkg模式进行设置，之后对生成的配置文件、启动脚本进行修改即可

编写一个app模式的启动脚本start_local_deploy_local_ins_protobuf_channel_publisher_app.sh​:

#!/bin/bash./normal_pb_chn_publisher_app ./cfg/local_deploy_local_ins_protobuf_channel_publisher_app_cfg.yaml

protobuf channel subscriber app

一个基于 protobuf 协议与 local 后端的 channel 示例，演示内容包括：

• 如何使用 protobuf 协议作为 channel 传输协议；
• 如何基于 App 模式创建模块的方式使用 Channel subscribe 功能；
• 如何基于 Module 方式使用 Channel publish 功能；
• 如何使用 local 类型的 channel 后端；
• 如何以 App 模式启动；

配置文件(cfg/local_deploy_local_ins_protobuf_channel_subscriber_app_cfg.yaml​)

aimrt:log:core_lvl: INFO # Trace/Debug/Info/Warn/Error/Fatal/Offbackends:- type: consoleexecutor:executors:- name: work_thread_pooltype: asio_threadoptions:thread_num: 4channel:backends:- type: localoptions:subscriber_use_inline_executor: falsesubscriber_executor: work_thread_poolpub_topics_options:- topic_name: "(.*)"enable_backends: [local]sub_topics_options:- topic_name: "(.*)"enable_backends: [local]module:pkgs:- path: ./libprotobuf_channel_pub_pkg.soenable_modules: [NormalPublisherModule]modules:- name: NormalPublisherModulelog_lvl: INFO- name: NormalSubscriberModulelog_lvl: INFO# Module custom configuration
NormalPublisherModule:topic_name: test_topicchannel_frq: 0.5NormalSubscriberModule:topic_name: test_topic

以app模式启动

启动命令为：

./normal_pb_chn_subscriber_app ./cfg/local_deploy_local_ins_protobuf_channel_subscriber_app_cfg.yaml

**注意：**​aim_cli​工具不支持生成App模式代码框架，因此不会有相应的启动脚本，需要自行编写。

• 可以在生成代码框架时，先使用pkg模式进行设置，之后对生成的配置文件、启动脚本进行修改即可

编写一个app模式的启动脚本start_local_deploy_local_ins_protobuf_channel_publisher_app.sh​:

#!/bin/bash./normal_pb_chn_subscriber_app ./cfg/local_deploy_local_ins_protobuf_channel_subscriber_app_cfg.yaml

知识点

发布消息的流程

✅ 1. 读取配置参数（如 topic 名称、发布频率）

• 作用：动态配置 topic 名、发布频率等运行参数。

• 接口示例：

  auto file_path = core_.GetConfigurator().GetConfigFilePath();
  YAML::Node cfg_node = YAML::LoadFile(std::string(file_path));
  topic_name_ = cfg_node["topic_name"].as<std::string>();
  channel_frq_ = cfg_node["channel_frq"].as<double>();
  

---

✅ 2. 获取执行器 ExecutorRef（用于任务调度）

• 作用：用于安排主发布任务的异步执行。

• 接口：

  executor_ = core_.GetExecutorManager().GetExecutor("work_thread_pool");
  

• 验证是否支持定时调度：

  AIMRT_CHECK_ERROR_THROW(executor_ && executor_.SupportTimerSchedule(), ...);
  

---

✅ 3. 获取发布者 PublisherRef（指定 topic）

• 作用：绑定发布主题，用于后续发送消息。

• 接口：

  publisher_ = core_.GetChannelHandle().GetPublisher(topic_name_);
  

---

✅ 4. 注册发布消息类型（绑定 Protobuf 类型）

• 作用：向系统注册将要发送的 Protobuf 消息类型。

• 接口：

  bool ret = aimrt::channel::RegisterPublishType<Protobuf_channel::my_proto::ExampleEventMsg>(publisher_);
  AIMRT_CHECK_ERROR_THROW(ret, "注册发布类型失败");
  

---

✅ 5. 创建 PublisherProxy（可选，简化发布）

• 作用：封装 PublisherRef，便于发布消息，自动处理上下文。

• 接口：

  aimrt::channel::PublisherProxy<Protobuf_channel::my_proto::ExampleEventMsg> publisher_proxy(publisher_);
  

---

✅ 6. 构造并填充消息对象

• 作用：设置消息字段（如文本、数值）。

• 接口：

  Protobuf_channel::my_proto::ExampleEventMsg msg;
  msg.set_msg("当前计数: " + std::to_string(count));
  msg.set_num(count);
  

---

✅ 7. 发布消息

• 作用：将消息写入通道，供订阅者消费。

• 接口：

  publisher_proxy.Publish(msg);
  

‍

🗂️ 总结表（含具体接口）

步骤	描述	接口/代码片段
1	获取配置参数	​cfg_node["topic_name"].as[std::string](std::string)()​ cfg_node["channel_frq"].as<double>()​
2	获取执行器	​core_.GetExecutorManager().GetExecutor("work_thread_pool")​
3	获取 Publisher	​core_.GetChannelHandle().GetPublisher(topic_name_)​
4	注册消息类型	​RegisterPublishType<ExampleEventMsg>(publisher_)​
5	创建 PublisherProxy	​PublisherProxy<ExampleEventMsg> proxy(publisher_)​
6	构造消息	​msg.set_msg(...)​,msg.set_num(...)​
7	发布消息	​proxy.Publish(msg)​

接收消息的流程

✅ 1. 读取配置参数（如 topic 名称）

• 作用：动态配置接收的 topic 名，便于部署时灵活调整。

• 接口示例：

  auto file_path = core_.GetConfigurator().GetConfigFilePath();
  YAML::Node cfg_node = YAML::LoadFile(std::string(file_path));
  topic_name_ = cfg_node["topic_name"].as<std::string>();
  

---

✅ 2. 获取订阅者 SubscriberRef（绑定到 topic）

• 作用：从通道系统获取订阅者句柄，准备订阅对应的消息通道。

• 接口：

  subscriber_ = core_.GetChannelHandle().GetSubscriber(topic_name_);
  AIMRT_CHECK_ERROR_THROW(subscriber_, "获取主题'{}'的订阅者失败", topic_name_);
  

---

✅ 3. 注册订阅消息类型 + 绑定回调函数（完成订阅）

• 作用：声明要接收的消息类型 + 注册回调函数； 回调函数将在消息到达时被自动触发。

• 接口：

  bool ret = aimrt::channel::Subscribe<Protobuf_channel::my_proto::ExampleEventMsg>(subscriber_,std::bind(&NormalSubscriberModule::EventHandle, this,std::placeholders::_1, std::placeholders::_2));
  AIMRT_CHECK_ERROR_THROW(ret, "订阅消息失败");
  

• 说明：

  • 使用的是【函数风格】接口；
  • 回调形式为【带 Context 参数】+【智能指针消息】；
  • 只能在模块的 ​Initialize()​ ​ 阶段调用订阅函数；
  • 一个 SubscriberRef​ 不能重复订阅相同类型。

---

✅ 4. 实现回调函数，处理接收到的消息

• 作用：处理订阅消息的内容，可访问上下文信息、执行业务逻辑。

• 接口：

  void NormalSubscriberModule::EventHandle(aimrt::channel::ContextRef ctx,const std::shared_ptr<const Protobuf_channel::my_proto::ExampleEventMsg>& data) {AIMRT_INFO("收到新消息，上下文: {}, 数据内容: {}", ctx.ToString(),aimrt::Pb2CompactJson(*data));
  }
  

• 说明：

  • ​ctx​ 提供上下文信息（如消息来源、时间戳）；
  • ​data​ 是 Protobuf 消息的 const 智能指针，可直接使用；
  • 处理逻辑可以轻量，也可以将任务调度到其它执行器中（但执行器不是在订阅处指定的）。

---

🗂️ 总结表（含具体接口）

步骤	描述	接口/代码片段
1	获取配置参数（topic）	​cfg_node["topic_name"].as[std::string](std::string)()​
2	获取 SubscriberRef	​core_.GetChannelHandle().GetSubscriber(topic_name_)​
3	注册消息回调	​Subscribe<ExampleEventMsg>(subscriber_, callback)​
4	实现回调处理函数	​EventHandle(ContextRef, shared_ptr<const Msg>)​

关于订阅方的执行器

❓1. 为什么订阅方在代码中没有显式指定执行器？

在 AimRT 中，订阅方执行器的配置 不在订阅模块代码中指定，而是 通过配置文件的 Channel 后端参数进行集中管理，原因如下：

• 解耦设计理念：AimRT 框架采用高度模块化设计，模块本身只声明对 Topic 的订阅关系，具体使用哪个执行器（或是否同步处理）由 channel.backends.options​ 统一决定；
• 配置驱动、逻辑清晰：通过配置控制订阅执行行为可以更灵活地调整运行时特性，例如从同步切换为异步、调整执行线程数，而无需更改模块实现；
• 利于跨模块优化调度：在同一 Channel 后端中集中处理执行策略，有助于线程资源的统一调度和优化，避免各模块各自为政、造成资源浪费或线程争抢。

因此：订阅方是否使用执行器，由 ​channel.backends[].options​​ 中的 ​subscriber_executor​​ 决定，而非模块代码内部。

🛠️ 2. 相关配置解释

  executor:executors:- name: work_thread_pooltype: asio_threadoptions:thread_num: 4

🔸 定义了一个线程池执行器 work_thread_pool​，基于 asio_thread​ 类型，拥有 4 个线程。

🔸 这个线程池将会执行消息发布循环、消息接收回调的代码

  channel:backends:- type: localoptions:subscriber_use_inline_executor: falsesubscriber_executor: work_thread_pool

🔸 配置了一个 local​ 类型的 Channel 后端：

• ​subscriber_use_inline_executor: false​：表示不使用同步执行模式；
• ​subscriber_executor: work_thread_pool​：所有订阅回调将异步投递到 work_thread_pool​ 执行器中；
• 因此，订阅模块的回调逻辑不会阻塞发布方的 ​Publish()​ ​ 调用。

    pub_topics_options:- topic_name: "(.*)"enable_backends: [local]sub_topics_options:- topic_name: "(.*)"enable_backends: [local]

🔸 通配符 (.*)​ 匹配所有 Topic，使其均由 local​ 后端处理。

🔸 因此可以实现不同话题使用不同后端处理

🔌 3. AimRT 插件机制

AimRT 的 Channel​ 通信机制是高度可扩展的 —— 其后端实现支持插件化架构。除了内置的 local​ 后端之外，官方或第三方还可以实现如下多种通信形式的插件：

• 网络通信（如 TCP、UDP、Http）
• 中间件集成（如 ROS2 DDS）

这些插件均可作为 channel.backends​ 的新类型，通过配置加载并与发布/订阅模块联动，实现模块间的远程、跨进程或跨平台消息传递。

相关内容后续学习插件章节时再进行详细记录