ROS2 执行器（Executor）与回调组（Callback Group）

ROS2 作为分布式机器人开发框架其核心事件驱动模型依赖执行器Executor驱动回调函数执行而回调组Callback Group则是控制回调并发策略的核心机制。一、概念1.1 执行器ExecutorROS2 中节点的回调函数订阅、定时器、服务/客户端、动作等不会自动执行必须由 Executor 接管并驱动。Executor 作用监听事件通过rcl_wait监听底层 ROS 中间件的事件如话题消息到达、服务请求就绪事件处理从“等待集wait set”中取出就绪事件调度回调根据回调组的并发规则在 Executor 线程中执行回调函数循环执行持续上述流程直到被显式停止或节点销毁。1.2 回调组Callback Group每个回调函数必须归属一个回调组显式/隐式回调组定义了组内回调的并发执行规则是 Executor 调度的依据。其作用是将不同特性的回调如快速响应/耗时计算、是否共享资源分组避免单回调阻塞整个节点。二、执行器Executor2.1 执行器的类型ROS2 提供 3 种 Executor均继承自rclcpp::Executor基类类型核心特性适用场景SingleThreadedExecutor单线程串行执行所有回调节点默认spin()即使用此类型简单场景、无并发需求、调试方便无线程安全问题MultiThreadedExecutor多线程并行执行回调可配置线程数默认 CPU 核心数回调有阻塞/耗时操作、需高响应性的场景StaticSingleThreadedExecutor静态单线程执行器提前注册所有节点/回调运行时无动态注册开销性能敏感、回调固定的场景如嵌入式设备2.2 Executor APIAPI功能add_node(node)将节点注册到 Executor多节点管理核心remove_node(node)从 Executor 移除节点spin()阻塞式循环持续执行回调直到stop()或节点销毁spin_once(timeout)等待timeout时长执行一个就绪回调无就绪则超时退出spin_some()非阻塞执行所有已就绪的回调无等待stop()终止 Executor 的循环需手动调用否则spin()一直阻塞2.3 特性多节点管理一个 Executor 可注册多个节点统一调度所有节点的回调提升资源利用率线程模型MultiThreadedExecutor的线程数通过set_num_threads(n)配置线程数不宜超过 CPU 核心数避免线程切换开销阻塞特性spin()是阻塞调用若需非阻塞执行需将spin()放在独立线程中。三、回调组Callback Group3.1 回调组的类型回调组的类型由rclcpp::CallbackGroupType定义决定组内回调的并发规则类型核心规则线程安全适用场景MutuallyExclusive默认同一组内回调串行执行不同组可并行Executor 多线程时内置互斥锁无需手动加锁共享资源的回调、快速响应回调Reentrant同一组内回调可并行执行Executor 多线程时无内置锁需用户保证线程安全无共享资源的耗时回调、独立计算任务3.2 回调组的绑定方式隐式绑定创建回调如订阅者时不指定回调组自动加入节点的默认MutuallyExclusive回调组显式绑定手动创建回调组在创建订阅/定时器时通过callback_group参数指定所属组。3.3 特性生命周期回调组与节点绑定节点销毁则回调组自动销毁组间并发MultiThreadedExecutor下不同MutuallyExclusive组的回调可并行执行组内并发仅Reentrant组支持组内并行MutuallyExclusive组内绝对串行。四、Executor 与 Callback Group 的协同规则两者的协同是 ROS2 回调调度的核心规则总结Executor 决定最大并行度线程数回调组决定组内并发规则单线程 Executor 下无论回调组类型如何所有回调均串行执行Reentrant组也无法并行多线程 Executor 下MutuallyExclusive组组内串行组间并行Reentrant组组内/组间均并行需手动保证线程安全默认行为节点spin()等价于SingleThreadedExecutor 默认MutuallyExclusive回调组。五、C 代码示例示例创建不同回调组、绑定订阅/定时器、单/多线程 Executor 对比、Reentrant 组线程安全处理。5.1 完整代码#includerclcpp/rclcpp.hpp#includestd_msgs/msg/string.hpp#includemutex#includechronousingnamespacestd::chrono_literals;classExecutorCallbackGroupDemo:publicrclcpp::Node{public:ExecutorCallbackGroupDemo():Node(executor_cg_demo),counter_(0){// 1. 创建不同类型的回调组 // 互斥型回调组默认cg_mutex_this-create_callback_group(rclcpp::CallbackGroupType::MutuallyExclusive);// 可重入型回调组cg_reentrant_this-create_callback_group(rclcpp::CallbackGroupType::Reentrant);// 2. 绑定订阅者到不同回调组 // 订阅者1绑定到互斥组处理快速消息sub1_this-create_subscriptionstd_msgs::msg::String(topic_mutex,10,std::bind(ExecutorCallbackGroupDemo::callback_mutex,this,std::placeholders::_1),rclcpp::SubscriptionOptions().callback_group(cg_mutex_)// 显式指定回调组);// 订阅者2绑定到可重入组处理耗时任务sub2_this-create_subscriptionstd_msgs::msg::String(topic_reentrant,10,std::bind(ExecutorCallbackGroupDemo::callback_reentrant,this,std::placeholders::_1),rclcpp::SubscriptionOptions().callback_group(cg_reentrant_));// 3. 绑定定时器到不同回调组 // 定时器1绑定到互斥组1Hz快速任务timer1_this-create_wall_timer(1s,std::bind(ExecutorCallbackGroupDemo::timer_callback_mutex,this),cg_mutex_// 直接指定回调组简化写法);// 定时器2绑定到可重入组2Hz耗时任务timer2_this-create_wall_timer(500ms,std::bind(ExecutorCallbackGroupDemo::timer_callback_reentrant,this),cg_reentrant_);}// ---------- 互斥组回调串行无需手动加锁 ----------voidcallback_mutex(conststd_msgs::msg::String::SharedPtr msg){RCLCPP_INFO(this-get_logger(),[Mutex Group] Sub1 received: %s,msg-data.c_str());std::this_thread::sleep_for(500ms);// 模拟短耗时}voidtimer_callback_mutex(){RCLCPP_INFO(this-get_logger(),[Mutex Group] Timer1 (1Hz) counter: %d,counter_);}// ---------- 可重入组回调并行需手动加锁 ----------voidcallback_reentrant(conststd_msgs::msg::String::SharedPtr msg){std::lock_guardstd::mutexlock(mtx_);// 保护共享资源RCLCPP_INFO(this-get_logger(),[Reentrant Group] Sub2 received: %s (thread: %ld),msg-data.c_str(),std::this_thread::get_id());std::this_thread::sleep_for(1s);// 模拟长耗时}voidtimer_callback_reentrant(){std::lock_guardstd::mutexlock(mtx_);// 保护共享资源RCLCPP_INFO(this-get_logger(),[Reentrant Group] Timer2 (2Hz) counter: %d (thread: %ld),counter_,std::this_thread::get_id());}private:rclcpp::CallbackGroup::SharedPtr cg_mutex_;// 互斥型回调组rclcpp::CallbackGroup::SharedPtr cg_reentrant_;// 可重入型回调组rclcpp::Subscriptionstd_msgs::msg::String::SharedPtr sub1_,sub2_;rclcpp::TimerBase::SharedPtr timer1_,timer2_;intcounter_;// 共享计数器需线程安全std::mutex mtx_;// 保护共享资源的互斥锁};intmain(intargc,char*argv[]){rclcpp::init(argc,argv);autonodestd::make_sharedExecutorCallbackGroupDemo();// 场景1单线程Executor所有回调串行 // rclcpp::executors::SingleThreadedExecutor executor;// executor.add_node(node);// RCLCPP_INFO(node-get_logger(), SingleThreadedExecutor started...);// executor.spin();// 场景2多线程Executor并行执行不同组回调 rclcpp::executors::MultiThreadedExecutor executor;executor.set_num_threads(4);// 设置4个线程executor.add_node(node);RCLCPP_INFO(node-get_logger(),MultiThreadedExecutor (4 threads) started...);executor.spin();rclcpp::shutdown();return0;}5.2 CMakeLists.txt 配置cmake_minimum_required(VERSION 3.8) project(executor_cg_demo) if(CMAKE_COMPILER_IS_GNUCXX OR CMAKE_CXX_COMPILER_ID MATCHES Clang) add_compile_options(-Wall -Wextra -Wpedantic) endif() find_package(ament_cmake REQUIRED) find_package(rclcpp REQUIRED) find_package(std_msgs REQUIRED) add_executable(executor_cg_demo src/executor_cg_demo.cpp) ament_target_dependencies(executor_cg_demo rclcpp std_msgs) install(TARGETS executor_cg_demo DESTINATION lib/${PROJECT_NAME}) if(BUILD_TESTING) find_package(ament_lint_auto REQUIRED) ament_lint_auto_find_test_dependencies() endif() ament_package()5.3 代码解析与运行效果回调组创建手动创建MutuallyExclusive和Reentrant两组回调组回调绑定订阅者/定时器分别绑定到不同组演示显式绑定方式线程安全处理Reentrant组的回调访问共享变量counter_时通过std::mutex加锁Executor 切换单线程 Executor所有回调串行执行timer22Hz会被sub21s 耗时阻塞实际频率远低于 2Hz多线程 ExecutorMutuallyExclusive组内串行Reentrant组内并行timer2和sub2可同时执行控制台会打印不同线程 ID。六、常见坑点与实践6.1 常见坑点单线程 Executor 阻塞若回调中有sleep()/IO 阻塞所有其他回调会被挂起解决方案改用多线程 Executor 分组Reentrant 组线程安全未加锁访问共享资源会导致数据错乱如counter_计数错误忘记注册节点手动创建 Executor 时未调用add_node()会导致回调不执行线程数过度配置MultiThreadedExecutor线程数超过 CPU 核心数会增加线程切换开销重复 spin() 调用同一线程多次调用spin()会导致永久阻塞需在独立线程中调用。6.2 实践分组策略快速响应回调如传感器数据接收放入MutuallyExclusive组避免被耗时回调阻塞耗时回调如路径规划、图像处理放入独立MutuallyExclusive组或Reentrant组无共享资源时共享资源的回调放入同一MutuallyExclusive组利用内置锁避免手动加锁Executor 选择简单场景使用默认spin()SingleThreadedExecutor复杂场景使用MultiThreadedExecutor线程数设为 CPU 核心数如 4/8性能敏感场景使用StaticSingleThreadedExecutor非阻塞 spin若需节点同时处理其他逻辑将spin()放入独立线程std::threadspin_thread([executor](){executor.spin();});// 主线程处理其他逻辑spin_thread.join();七、相关补充spin_once()/spin_some() 用法spin_once(100ms)等待 100ms若有就绪回调则执行一个否则退出spin_some()立即执行所有就绪回调无阻塞适合循环中轮询跨节点 Executor一个 Executor 可注册多个节点示例autonode1std::make_sharedNode1();autonode2std::make_sharedNode2();executor.add_node(node1);executor.add_node(node2);executor.spin();Executor 停止调用executor.stop()可终止spin()循环需在其他线程调用默认回调组节点的get_default_callback_group()可获取默认组用于批量绑定回调动态调整线程数MultiThreadedExecutor运行时可通过set_num_threads()动态调整线程数。ROS2 的 Executor 与 Callback Group 是控制回调执行的核心掌握两者的协同规则是编写高性能 ROS2 节点的关键Executor 决定“用多少线程执行回调”Callback Group 决定“哪些回调可以并行执行”单线程 Executor 保证简单性多线程 Executor 提升响应性回调组则是平衡并发与线程安全的核心手段。