这是 ROS 2 实时控制（ros2_control）必学核心库，专门解决实时线程与非实时线程安全通信的问题 —— 比如：

• 控制器 1kHz 硬实时线程(1ms周期）
• ROS 2 话题发布 / 订阅、参数、服务 非实时线程

如果不用它，直接通信会造成实时性断裂、抖动、卡死，机器人控制会失控。

我会用最清晰、最实用的方式讲完所有功能 + 可直接复制的代码。

---

一、realtime_tools 是什么？

官方定位：ROS 2 为硬实时控制设计的工具集，保证实时线程安全。

核心解决问题：

• 实时线程（控制器）不能被阻塞、不能做内存分配
• 非实时线程（ROS 消息、参数）会阻塞、会动态分配内存
• 两者之间必须用无锁、无阻塞的方式通信

---

二、核心功能（全部 4 个，全是高频使用）

1. RealtimeBuffer —— 无锁单生产者单消费者缓冲
2. RealtimePublisher —— 实时线程安全发布话题（最常用）
3. RealtimeSubscription —— 实时线程安全订阅话题
4. RealtimeClock —— 实时线程安全获取时间

全部用于 ros2_control 控制器、实时驱动、实时任务。

---

三、安装

bash

运行

sudo apt install ros-humble-realtime-tools

---

四、功能 1：RealtimePublisher（最常用！）

用途：在1kHz 实时控制线程里安全发布 ROS 2 话题，不破坏实时性。

普通 publisher 在实时线程里用会：

• 动态分配内存
• 锁等待
• 造成控制抖动、失控

RealtimePublisher = 安全版本

完整代码示例

cpp

运行

#include "rclcpp/rclcpp.hpp"
#include "std_msgs/msg/float64.hpp"
#include "realtime_tools/realtime_publisher.h"

using namespace std::chrono_literals;
using Float64 = std_msgs::msg::Float64;

class RealtimePubNode : public rclcpp::Node
{
public:
    RealtimePubNode() : Node("realtime_pub_demo")
    {
        // 1. 创建普通 publisher
        auto pub = this->create_publisher<Float64>("controller_state", 10);

        // 2. 包装成实时安全 publisher
        rt_pub_ = std::make_shared<realtime_tools::RealtimePublisher<Float64>>(pub);

        // 3. 模拟 1kHz 实时控制线程
        timer_ = this->create_wall_timer(
            1ms, std::bind(&RealtimePubNode::realtime_control_loop, this));
    }

private:
    // 实时控制循环（不能阻塞！）
    void realtime_control_loop()
    {
        if (rt_pub_->trylock()) {
            // 安全填充消息
            rt_pub_->msg_.data = 1.234;

            // 发布（无锁、无阻塞、实时安全）
            rt_pub_->unlockAndPublish();
        }

        RCLCPP_INFO_THROTTLE(this->get_logger(), *this->get_clock(), 1000, "实时发布中...");
    }

    std::shared_ptr<realtime_tools::RealtimePublisher<Float64>> rt_pub_;
    rclcpp::TimerBase::SharedPtr timer_;
};

int main(int argc, char * argv[])
{
    rclcpp::init(argc, argv);
    rclcpp::spin(std::make_shared<RealtimePubNode>());
    rclcpp::shutdown();
    return 0;
}

关键 API

cpp

运行

trylock()            // 非阻塞尝试加锁（实时线程必须用）
msg_                 // 消息本体
unlockAndPublish()   // 解锁 + 发布（一步完成）

---

五、功能 2：RealtimeBuffer（无锁缓冲）

用途：实时线程与非实时线程之间安全共享变量（如目标位置、力矩、速度）。

无锁、无阻塞、无内存分配。

示例：实时线程读取目标值

cpp

运行

#include "realtime_tools/realtime_buffer.h"

// 定义缓冲
realtime_tools::RealtimeBuffer<double> target_pos_buffer_;

// 非实时线程（订阅话题、参数）写入
void set_target_pos(double pos) {
    target_pos_buffer_.writeFromNonRT(pos);
}

// 实时线程（控制循环）读取
double get_target_pos() {
    return *target_pos_buffer_.readFromRT();
}

两个核心方法：

• writeFromNonRT()：非实时线程写
• readFromRT()：实时线程读

---

六、功能 3：RealtimeSubscription

用途：实时线程安全读取订阅的话题，不阻塞、不破坏实时性。

cpp

运行

#include "realtime_tools/realtime_subscription.h"

// 创建实时订阅
rt_sub_ = std::make_shared<realtime_tools::RealtimeSubscription<Float64>>(
    this->create_subscription<Float64>("target", 10,
        [this](const std::shared_ptr<Float64> msg) {
            // 非实时回调
        })
);

// 实时线程中安全取最新消息
auto msg = rt_sub_->getLatestMsg();
if (msg) {
    double target = msg->data;
}

---

七、功能 4：RealtimeClock

用途：实时线程安全获取 ROS 2 时间。

cpp

运行

#include "realtime_tools/realtime_clock.h"

realtime_tools::RealtimeClock rt_clock_;

// 实时线程获取时间
rclcpp::Time now = rt_clock_.now();

---

八、在 ros2_control 自定义控制器里的标准用法（工业级）

cpp

运行

// 头文件
#include "controller_interface/controller_interface.hpp"
#include "realtime_tools/realtime_publisher.h"
#include "std_msgs/msg/float64.hpp"

using Float64 = std_msgs::msg::Float64;

class MyController : public controller_interface::ControllerInterface
{
public:
    controller_interface::return_type init(const std::string &) override
    {
        // 普通 publisher
        auto pub = node_->create_publisher<Float64>("effort_cmd", 10);

        // 实时 publisher
        rt_pub_ = std::make_shared<realtime_tools::RealtimePublisher<Float64>>(pub);

        return controller_interface::return_type::OK;
    }

    controller_interface::return_type update() override
    {
        // 1kHz 实时控制循环

        double command = 1.0;

        // 安全发布
        if (rt_pub_->trylock()) {
            rt_pub_->msg_.data = command;
            rt_pub_->unlockAndPublish();
        }

        return controller_interface::return_type::OK;
    }

private:
    std::shared_ptr<realtime_tools::RealtimePublisher<Float64>> rt_pub_;
};

---

九、依赖配置（必须加）

package.xml

xml

<depend>realtime_tools</depend>

CMakeLists.txt

cmake

find_package(realtime_tools REQUIRED)

ament_target_dependencies(你的节点
  realtime_tools
)

---

十、超清晰总结

realtime_tools 是 ROS 2 实时控制的安全通信基石：

1. RealtimePublisher → 实时线程安全发消息
2. RealtimeBuffer → 实时 / 非实时安全共享变量
3. RealtimeSubscription → 实时线程安全读消息
4. RealtimeClock → 实时线程安全读时间

一句话记忆：实时线程里所有 ROS 2 通信，必须用 realtime_tools！