第四章 节点与回调优化：回调分组策略、回调优先级设置、避免回调阻塞的最佳实践

好，咱们进入第四章。这一章我特别想跟你好好聊聊，因为回调——说白了就是ROS2里干活的最小单元。你节点写得再花哨，回调一卡，整个系统就跟着遭殃。我在做机器人底盘控制的时候，就吃过这个亏，一个回调里做了点图像处理，结果电机控制指令延迟了50毫秒，机器人直接撞墙了。嗯，从那以后我对回调优化就特别上心。

4.1 回调分组策略：别把所有鸡蛋放一个篮子里

ROS2里，同一个节点的回调默认都在一个线程池里跑。你想想看，如果传感器回调、控制回调、日志回调全挤在一起，高优先级的控制任务就得排队等低优先级的日志写完。这合理吗？不合理。

我个人习惯的做法是：按实时性要求分组。把回调分成几个组，每组分配独立的线程池。

核心原则：

  • 实时性要求高的回调（如电机控制、IMU数据读取）放一组，用高优先级线程
  • 实时性要求中等的回调（如状态估计、路径规划）放一组
  • 实时性要求低的回调（如日志记录、诊断信息）放一组，用低优先级线程

具体怎么实现？看代码：

// 创建三个回调组
rclcpp::CallbackGroup::SharedPtr high_prio_group =
    create_callback_group(rclcpp::CallbackGroupType::MutuallyExclusive);
rclcpp::CallbackGroup::SharedPtr mid_prio_group =
    create_callback_group(rclcpp::CallbackGroupType::MutuallyExclusive);
rclcpp::CallbackGroup::SharedPtr low_prio_group =
    create_callback_group(rclcpp::CallbackGroupType::MutuallyExclusive);

// 订阅时指定回调组
auto motor_sub = create_subscription<MotorCmd>(
    "motor_cmd", 10,
    std::bind(&MotorController::motor_callback, this, _1),
    high_prio_group);

auto imu_sub = create_subscription<ImuData>(
    "imu_data", 10,
    std::bind(&MotorController::imu_callback, this, _1),
    high_prio_group);

auto log_sub = create_subscription<LogMsg>(
    "log_info", 10,
    std::bind(&MotorController::log_callback, this, _1),
    low_prio_group);

小技巧：我建议把回调组数量控制在3-5个。太多反而增加上下文切换开销，得不偿失。

4.2 回调优先级设置：给关键任务插队权

分组只是第一步。同一个组里的回调，谁先执行？这就涉及到优先级了。

ROS2本身没有直接给回调设置优先级的API，但我们可以通过操作系统线程优先级来实现。说白了，就是给不同回调组的线程设置不同的nice值或实时优先级。

我在项目中遇到过这样一个场景：机器人急停按钮的回调必须优先于所有其他回调。怎么做的？

// 在节点初始化时设置线程优先级
void set_thread_priority(rclcpp::CallbackGroup::SharedPtr group, int priority) {
    auto executor = std::make_shared<rclcpp::executors::MultiThreadedExecutor>();
    executor->add_callback_group(group, this->get_node_base_interface());
    
    // 设置线程调度策略和优先级
    struct sched_param param;
    param.sched_priority = priority;
    pthread_setschedparam(pthread_self(), SCHED_FIFO, &param);
}

注意：设置实时优先级需要root权限。我在调试时吃过这个亏，程序跑起来没效果，查了半天才发现是权限问题。建议在启动脚本里用 sudo 或者设置 cap_sys_nice 能力。

优先级设置建议：

回调类型 推荐优先级 调度策略
安全相关（急停、碰撞检测） 90-99 SCHED_FIFO
实时控制（电机、伺服） 70-89 SCHED_FIFO
传感器数据（IMU、激光雷达） 50-69 SCHED_RR
状态估计、规划 30-49 SCHED_RR
日志、诊断、可视化 0-29 SCHED_OTHER

4.3 避免回调阻塞的最佳实践：别让你的回调睡着

这是最容易被忽视的问题。回调阻塞，说白了就是回调函数里干了不该干的事，导致它迟迟不返回，后面的回调全得等着。

我曾经在一个项目里，同事在回调里直接调用了 std::this_thread::sleep_for(100ms)，说是为了等待传感器数据稳定。结果整个控制循环从100Hz掉到了10Hz，机器人走路一瘸一拐的。嗯，这个坑我记忆犹新。

避免阻塞的核心原则就一条：回调里不要做任何可能阻塞的操作。具体来说：

禁止在回调中做的事：

  • ❌ 调用 sleep()、usleep()、std::this_thread::sleep_for()
  • ❌ 执行阻塞式I/O（文件读写、网络请求）
  • ❌ 等待互斥锁（如果锁可能被长时间持有）
  • ❌ 执行耗时计算（图像处理、大型矩阵运算）
  • ❌ 调用 rclcpp::spin() 或 rclcpp::spin_some()

那正确的做法是什么？把耗时操作扔到独立线程里去。回调只负责接收数据、做最轻量的处理，然后通过消息队列或共享变量把数据传给工作线程。

// 错误示范：回调里做耗时操作
void bad_callback(const ImageMsg::SharedPtr msg) {
    // 直接在回调里做图像处理，阻塞了！
    auto result = heavy_image_processing(*msg);
    publish_result(result);
}

// 正确做法：回调只负责传递数据
void good_callback(const ImageMsg::SharedPtr msg) {
    // 把数据扔到队列里，立刻返回
    image_queue_.push(*msg);
}

// 工作线程在后台处理
void processing_thread() {
    while (rclcpp::ok()) {
        if (image_queue_.try_pop(msg)) {
            auto result = heavy_image_processing(msg);
            publish_result(result);
        }
    }
}

避坑指南：我曾经用 std::queue 做消息队列，结果在多线程环境下出了数据竞争。后来改用 tbb::concurrent_bounded_queue 或者 ROS2自带的 rclcpp::Buffer，就稳多了。记住，回调里访问共享数据一定要加锁或用无锁数据结构。

还有一个常见问题：回调里调用其他节点的服务。服务调用是同步的，会阻塞当前回调。我建议改用异步方式：

// 错误：同步调用服务，阻塞回调
void callback() {
    auto client = create_client<SomeService>("service_name");
    auto result = client->async_send_request(request);  // 还是阻塞了！
    // 实际上 async_send_request 返回 future，需要等待
}

// 正确：使用异步回调
void callback() {
    auto client = create_client<SomeService>("service_name");
    client->async_send_request(
        request,
        [this](rclcpp::Client<SomeService>::SharedFuture future) {
            // 服务响应在另一个回调里处理，不阻塞当前回调
            handle_service_response(future);
        });
}

4.4 实战经验总结

说了这么多，我总结几条最实用的经验：

  1. 回调分组是基础：把不同实时性要求的回调分开，别混在一起。我一般分3组：高（控制）、中（感知）、低（日志）。
  2. 优先级要合理：安全相关的回调给最高优先级，但别给太高，否则可能饿死其他任务。99留给急停就够了。
  3. 回调里只做三件事：接收数据、简单判断、传递数据。其他一切操作都扔到工作线程里。
  4. 监控回调执行时间：我习惯在每个回调开头和结尾记录时间戳，如果发现某个回调执行超过1ms，就重点排查。
  5. 使用executor的统计功能：ROS2的executor可以统计回调执行时间，用 rclcpp::executors::SingleThreadedExecutor 加上 add_callback_group 就能看到每个回调组的负载情况。

最后提醒一句：回调优化不是一劳永逸的事。系统在变，负载在变，你得定期用 ros2 topic hz 和 ros2 topic delay 检查回调的执行情况。我每两周做一次性能审计，确保系统一直跑在最佳状态。

好了，这一章就到这里。下一章咱们聊聊executor的配置和调优，那是ROS2实时系统的核心引擎。到时候我会分享一个我优化过的executor配置，能让你的节点响应速度提升30%以上。

← 上一章 📖 返回目录 下一章 →