1. 从零开始：你的第一个ROS2节点

如果你刚接触ROS2，可能会觉得“节点”这个词有点抽象。别担心，我们可以用一个简单的类比来理解：想象一个机器人公司，每个部门（比如视觉处理部、运动规划部、决策控制部）都是一个独立的“节点”。它们各自有明确的职责，通过公司内部的邮件系统（ROS2的通信机制）交换信息，协同完成“让机器人动起来”这个大任务。rclcpp 就是ROS2为C++开发者提供的官方“部门建设与沟通工具包”。

那么，如何用这个工具包创建你的第一个“部门”呢？代码其实非常简洁。我们先来看一个最基础的“Hello World”节点，它不做任何通信，仅仅完成创建和运行。

#include "rclcpp/rclcpp.hpp"

int main(int argc, char **argv)
{
  // 1. 初始化ROS2系统
  rclcpp::init(argc, argv);
  
  // 2. 创建一个名为“my_first_node”的节点
  auto node = std::make_shared<rclcpp::Node>("my_first_node");
  
  // 3. 打印一条日志，证明节点活了
  RCLCPP_INFO(node->get_logger(), "Hello ROS2! My first node is running.");
  
  // 4. 让节点保持运行，等待处理（虽然这里没什么可处理的）
  rclcpp::spin(node);
  
  // 5. 关闭ROS2系统，释放资源
  rclcpp::shutdown();
  return 0;
}

把这段代码保存为 hello_node.cpp。接下来，我们需要把它编译成一个可执行程序。ROS2项目通常使用 CMakeLists.txt 来管理编译。在你的项目目录下，创建一个 CMakeLists.txt 文件，内容如下：

cmake_minimum_required(VERSION 3.8)
project(my_ros2_package)

# 如果没有找到ROS2，就去系统里找
find_package(ament_cmake REQUIRED)
find_package(rclcpp REQUIRED)

# 声明我们要生成一个可执行文件
add_executable(hello_node src/hello_node.cpp)
# 告诉编译器，这个可执行文件需要链接rclcpp库
ament_target_dependencies(hello_node rclcpp)

# 安装指令，方便后续用命令行启动
install(TARGETS hello_node
  DESTINATION lib/${PROJECT_NAME})

然后，在终端里进入你的项目目录，依次执行以下命令来编译和运行：

# 创建一个名为‘build’的文件夹用于存放编译中间文件
mkdir -p build
cd build
# 运行cmake生成编译规则
cmake ..
# 开始编译
make
# 返回项目根目录，并设置ROS2环境（假设你已经安装好了ROS2）
cd ..
source /opt/ros/humble/setup.bash  # 根据你的ROS2版本（如foxy, galactic）调整路径
# 运行你的节点！
ros2 run my_ros2_package hello_node

如果一切顺利，你会在终端看到输出：[INFO] [my_first_node]: Hello ROS2! My first node is running.。恭喜你，你的第一个ROS2节点已经成功启动并运行了！这个节点现在就像一个安静的部门，虽然还没开始跟其他部门通信，但它的基础架构已经搭建完毕。rclcpp::spin(node) 这行代码是关键，它让节点进入一个循环，持续监听外部事件（比如将来要接收的消息或服务请求），保持节点活跃而不退出。在实际开发中，节点的生命周期管理非常重要，我们会在后续章节深入探讨如何优雅地启动和关闭节点，以及如何处理信号中断。

2. 让节点“开口说话”：发布与订阅模型

节点创建好了，但一个不会交流的节点在机器人系统里是“孤独”的。ROS2中最基础、最常用的通信方式就是发布/订阅（Publish/Subscribe）模型。这就像公司里的公告板（Topic），有的部门（Publisher）负责往公告板上贴通知（Message），而关心这个通知的部门（Subscriber）就会去公告板前阅读。这是一种单向、一对多的通信方式，发布者只管发，不关心谁接收；订阅者只管收，不关心谁发的。

让我们动手创建一个“聊天”场景：一个节点每秒发布一次问候，另一个节点实时接收并打印出来。首先，我们创建发布者节点 talker.cpp：

#include "rclcpp/rclcpp.hpp"
#include "std_msgs/msg/string.hpp" // 使用ROS2内置的字符串消息类型
#include <chrono>

using namespace std::chrono_literals;

class TalkerNode : public rclcpp::Node
{
public:
  TalkerNode() : Node("talker")
  {
    // 创建一个发布者，发布到“chatter”这个话题，队列长度设为10
    publisher_ = this->create_publisher<std_msgs::msg::String>("chatter", 10);
    
    // 创建一个定时器，每500毫秒触发一次回调函数
    timer_ = this->create_wall_timer(
      500ms, std::bind(&TalkerNode::timer_callback, this));
    
    RCLCPP_INFO(this->get_logger(), "Talker node started. Publishing to 'chatter' topic.");
  }

private:
  void timer_callback()
  {
    auto message = std_msgs::msg::String();
    message.data = "Hello ROS2! Current time: " + std::to_string(this->now().seconds());
    
    // 发布消息
    publisher_->publish(message);
    RCLCPP_INFO(this->get_logger(), "Publishing: '%s'", message.data.c_str());
  }
  
  rclcpp::Publisher<std_msgs::msg::String>::SharedPtr publisher_;
  rclcpp::TimerBase::SharedPtr timer_;
};

int main(int argc, char **argv)
{
  rclcpp::init(argc, argv);
  auto node = std::make_shared<TalkerNode>();
  rclcpp::spin(node);
  rclcpp::shutdown();
  return 0;
}

这个节点做了几件事：它继承自 rclcpp::Node，这是一种更面向对象、更推荐的组织节点代码的方式。在构造函数里，它创建了一个发布者，指定了消息类型（std_msgs::msg::String）和话题名称（“chatter”）。那个数字 10 是发布队列长度，它就像一个缓冲区，如果消息发布速度超过了网络发送速度，最新的10条消息会被暂存，更旧的消息会被丢弃。然后它创建了一个定时器，周期性地调用 timer_callback 函数来组装并发布消息。

现在，我们需要一个听众。创建订阅者节点 listener.cpp：

#include "rclcpp/rclcpp.hpp"
#include "std_msgs/msg/string.hpp"

class ListenerNode : public rclcpp::Node
{
public:
  ListenerNode() : Node("listener")
  {
    // 创建一个订阅者，订阅“chatter”话题
    // 当有新消息时，自动调用 `topic_callback` 函数
    subscription_ = this->create_subscription<std_msgs::msg::String>(
      "chatter", 10, std::bind(&ListenerNode::topic_callback, this, std::placeholders::_1));
    
    RCLCPP_INFO(this->get_logger(), "Listener node started. Listening to 'chatter' topic.");
  }

private:
  void topic_callback(const std_msgs::msg::String::SharedPtr msg) const
  {
    // 收到消息时的处理逻辑
    RCLCPP_INFO(this->get_logger(), "I heard: '%s'", msg->data.c_str());
  }
  
  rclcpp::Subscription<std_msgs::msg::String>::SharedPtr subscription_;
};

int main(int argc, char **argv)
{
  rclcpp::init(argc, argv);
  auto node = std::make_shared<ListenerNode>();
  rclcpp::spin(node);
  rclcpp::shutdown();
  return 0;
}

订阅者的核心是 create_subscription 函数。你指定要监听的话题（“chatter”）、队列长度（同样是10），以及一个回调函数。一旦有消息发布到“chatter”话题上，ROS2底层中间件就会自动把消息传递过来，并调用你注册的 topic_callback 函数。这里的 std::placeholders::_1 是一个占位符，代表回调函数接收的第一个参数（即消息指针）。

将这两个节点都添加到 CMakeLists.txt 中编译，然后打开两个终端，分别运行 ros2 run my_package talker 和 ros2 run my_package listener。你会看到 talker 在不停地打印发布信息，而 listener 则实时地打印出它听到的内容。这就是ROS2分布式通信的魅力，这两个节点甚至可以运行在网络中不同的电脑上，只要它们连接到同一个ROS2网络（通过配置ROS_DOMAIN_ID），就能无缝通信。

注意：发布/订阅是异步和非阻塞的。发布者调用 publish() 后立刻返回，不会等待订阅者接收。这种设计保证了发布者线程的高效性，但也要求开发者注意消息的时序和丢失问题，对于要求可靠、按序送达的场景，需要设计更上层的协议或使用其他通信方式。

3. 请求与应答：服务与客户端模型

发布/订阅模型很适合流式数据广播，但机器人系统中还有很多场景需要请求-应答（Request-Response） 式的同步交互。比如，你让机器人“去拿一杯水”，机器人需要回答“好的，正在执行”或者“抱歉，水杯不在视野内”。这就是ROS2的服务（Service） 模型。它就像公司里的专项工作接口：一个部门（Service Server）对外提供一个特定的服务（比如“计算两数之和”），其他部门（Service Client）可以提交请求单，并等待该部门处理完成后返回结果单。

让我们实现一个经典的“两数相加”服务。首先，我们需要定义服务接口。ROS2已经为我们准备了很多基础服务类型，在 example_interfaces 包中就有一个现成的 AddTwoInts 服务。我们直接使用它。先看服务端 add_two_ints_server.cpp：

#include "rclcpp/rclcpp.hpp"
#include "example_interfaces/srv/add_two_ints.hpp"

using std::placeholders::_1;
using std::placeholders::_2;

class AddTwoIntsServerNode : public rclcpp::Node
{
public:
  AddTwoIntsServerNode() : Node("add_two_ints_server")
  {
    // 创建一个服务，服务名为“add_two_ints”，绑定处理回调函数
    service_ = this->create_service<example_interfaces::srv::AddTwoInts>(
      "add_two_ints",
      std::bind(&AddTwoIntsServerNode::handle_add_two_ints, this, _1, _2));
    
    RCLCPP_INFO(this->get_logger(), "AddTwoInts service server ready.");
  }

private:
  // 服务处理函数
  void handle_add_two_ints(
    const std::shared_ptr<example_interfaces::srv::AddTwoInts::Request> request,
    std::shared_ptr<example_interfaces::srv::AddTwoInts::Response> response)
  {
    // 从请求中取出a和b
    long long a = request->a;
    long long b = request->b;
    
    // 执行“服务”：计算和
    long long sum = a + b;
    
    // 将结果填入响应中
    response->sum = sum;
    
    RCLCPP_INFO(this->get_logger(), "Incoming request: a=%lld, b=%lld", a, b);
    RCLCPP_INFO(this->get_logger(), "Sending back response: [%lld]", sum);
  }
  
  rclcpp::Service<example_interfaces::srv::AddTwoInts>::SharedPtr service_;
};

int main(int argc, char **argv)
{
  rclcpp::init(argc, argv);
  auto node = std::make_shared<AddTwoIntsServerNode>();
  rclcpp::spin(node);
  rclcpp::shutdown();
  return 0;
}

服务端的关键是 create_service 函数。它指定了服务类型和名称，并绑定了一个处理函数。这个处理函数有两个参数：一个是请求对象的共享指针，一个是响应对象的共享指针。你从请求中读取数据，进行逻辑处理，然后把结果写入响应对象。ROS2框架会自动帮你完成网络传输和响应回复。

现在，我们创建一个客户端来调用这个服务 add_two_ints_client.cpp：

#include "rclcpp/rclcpp.hpp"
#include "example_interfaces/srv/add_two_ints.hpp"
#include <chrono>
#include <cstdlib>

using namespace std::chrono_literals;

class AddTwoIntsClientNode : public rclcpp::Node
{
public:
  AddTwoIntsClientNode() : Node("add_two_ints_client"), count_(0)
  {
    // 创建一个客户端，连接到“add_two_ints”服务
    client_ = this->create_client<example_interfaces::srv::AddTwoInts>("add_two_ints");
    
    // 创建一个定时器，每秒尝试发送一次请求
    timer_ = this->create_wall_timer(
      1s, std::bind(&AddTwoIntsClientNode::send_request, this));
  }

private:
  void send_request()
  {
    // 等待服务端上线
    while (!client_->wait_for_service(1s)) {
      if (!rclcpp::ok()) {
        RCLCPP_ERROR(this->get_logger(), "Interrupted while waiting for service.");
        return;
      }
      RCLCPP_INFO(this->get_logger(), "Service not available, waiting again...");
    }
    
    // 构造请求
    auto request = std::make_shared<example_interfaces::srv::AddTwoInts::Request>();
    request->a = count_ * 2;
    request->b = count_ * 2 + 1;
    count_++;
    
    // 异步发送请求，并指定收到响应后的回调函数
    auto future_result = client_->async_send_request(request);
    
    // 异步等待结果（非阻塞方式，在实际应用中更常见）
    // 这里为了演示简单，我们使用future的wait（会阻塞当前线程）
    // 在实际节点中，更推荐使用 `rclcpp::spin_until_future_complete(node, future_result)`
    std::future_status status = future_result.wait_for(2s);
    if (status == std::future_status::ready) {
      auto response = future_result.get();
      RCLCPP_INFO(this->get_logger(), "Result of %lld + %lld = %lld", 
                  request->a, request->b, response->sum);
    } else {
      RCLCPP_ERROR(this->get_logger(), "Failed to call service add_two_ints");
    }
  }
  
  rclcpp::Client<example_interfaces::srv::AddTwoInts>::SharedPtr client_;
  rclcpp::TimerBase::SharedPtr timer_;
  int count_;
};

int main(int argc, char **argv)
{
  rclcpp::init(argc, argv);
  auto node = std::make_shared<AddTwoIntsClientNode>();
  rclcpp::spin(node);
  rclcpp::shutdown();
  return 0;
}

客户端代码稍微复杂一点，因为它涉及到服务发现和异步调用。create_client 创建了一个客户端对象，但此时服务端可能还没启动。所以客户端需要先 wait_for_service 来等待服务端上线。发送请求时，我们使用 async_send_request 进行异步调用，它立即返回一个 std::shared_future 对象，不会阻塞当前线程。然后我们可以选择在未来的某个时刻（比如在定时器回调里，或者另一个线程里）去检查这个 future 是否已经完成（即收到了响应），并获取结果。这种异步模式避免了客户端在等待服务响应时“卡住”，可以同时处理其他任务，是构建高效响应式节点的关键。

编译并运行这两个节点。先启动服务端，再启动客户端，你会看到客户端每秒发送一组递增的数字给服务端，并打印出服务端返回的和。服务模型非常适合执行一次性的、需要明确结果的任务，比如查询传感器状态、调用一个算法、触发一个动作等。

4. 构建复杂任务：动作服务器与客户端

服务和发布/订阅模型能解决很多问题，但机器人任务中有一类非常典型的需求：长时间运行、可反馈、可取消的任务。比如“移动到目标点”，这个任务可能需要几秒甚至几分钟，我们希望在执行过程中能持续收到“当前已走50%”这样的进度反馈，并且能在中途紧急取消。ROS2的动作（Action） 就是为这种场景设计的。你可以把它理解为“加强版的服务”，它由目标（Goal）、反馈（Feedback）、结果（Result）三部分组成，并且支持在任务执行中持续推送反馈，以及接收取消指令。

我们来实现一个模拟的“倒计时”动作服务器。首先，需要定义动作接口。创建一个名为 action 的文件夹，在里面新建 Countdown.action 文件：

# 目标：从哪个数字开始倒计时
int32 start_number
---
# 结果：最终结束时的信息
bool success
string message
---
# 反馈：每秒反馈当前剩余数字
int32 current_number

然后，我们需要在 CMakeLists.txt 和 package.xml 中配置，让ROS2的构建系统在编译时根据这个 .action 文件自动生成对应的C++头文件。这个过程稍微繁琐，但一旦配置好，后面就可以像使用普通消息类型一样使用自定义的动作类型了。假设我们已经生成了头文件 my_interface/action/countdown.hpp。

接下来创建动作服务器 countdown_action_server.cpp：

#include "rclcpp/rclcpp.hpp"
#include "rclcpp_action/rclcpp_action.hpp"
#include "my_interface/action/countdown.hpp"
#include <thread>
#include <atomic>

using Countdown = my_interface::action::Countdown;
using GoalHandleCountdown = rclcpp_action::ServerGoalHandle<Countdown>;

class CountdownActionServer : public rclcpp::Node
{
public:
  CountdownActionServer() : Node("countdown_action_server")
  {
    // 创建动作服务器
    this->action_server_ = rclcpp_action::create_server<Countdown>(
      this,
      "countdown", // 动作名称
      std::bind(&CountdownActionServer::handle_goal, this, std::placeholders::_1, std::placeholders::_2),
      std::bind(&CountdownActionServer::handle_cancel, this, std::placeholders::_1),
      std::bind(&CountdownActionServer::handle_accepted, this, std::placeholders::_1));
    
    RCLCPP_INFO(this->get_logger(), "Countdown action server started.");
  }

private:
  rclcpp_action::Server<Countdown>::SharedPtr action_server_;
  
  // 1. 处理新目标请求：决定是否接受这个目标
  rclcpp_action::GoalResponse handle_goal(
    const rclcpp_action::GoalUUID & uuid,
    std::shared_ptr<const Countdown::Goal> goal)
  {
    RCLCPP_INFO(this->get_logger(), "Received goal request with start_number %d", goal->start_number);
    // 简单检查：开始数字必须为正数
    if (goal->start_number <= 0) {
      RCLCPP_WARN(this->get_logger(), "Goal rejected: start_number must be positive.");
      return rclcpp_action::GoalResponse::REJECT;
    }
    return rclcpp_action::GoalResponse::ACCEPT_AND_EXECUTE;
  }
  
  // 2. 处理取消请求
  rclcpp_action::CancelResponse handle_cancel(
    const std::shared_ptr<GoalHandleCountdown> goal_handle)
  {
    RCLCPP_INFO(this->get_logger(), "Received request to cancel goal");
    // 这里我们允许取消
    return rclcpp_action::CancelResponse::ACCEPT;
  }
  
  // 3. 目标被接受后，在新线程中执行
  void handle_accepted(const std::shared_ptr<GoalHandleCountdown> goal_handle)
  {
    // 为了避免阻塞动作服务器的回调线程，我们开一个新线程来执行长时间任务
    std::thread{std::bind(&CountdownActionServer::execute, this, std::placeholders::_1), goal_handle}.detach();
  }
  
  // 实际执行倒计时的函数
  void execute(const std::shared_ptr<GoalHandleCountdown> goal_handle)
  {
    RCLCPP_INFO(this->get_logger(), "Executing goal...");
    rclcpp::Rate loop_rate(1); // 1Hz
    const auto goal = goal_handle->get_goal();
    auto feedback = std::make_shared<Countdown::Feedback>();
    auto result = std::make_shared<Countdown::Result>();
    
    int current_number = goal->start_number;
    
    for (int i = current_number; i > 0; --i) {
      // 检查是否被取消
      if (goal_handle->is_canceling()) {
        result->success = false;
        result->message = "Countdown canceled by user.";
        goal_handle->canceled(result);
        RCLCPP_INFO(this->get_logger(), "Goal canceled.");
        return;
      }
      
      // 发布反馈：当前剩余数字
      feedback->current_number = i;
      goal_handle->publish_feedback(feedback);
      RCLCPP_INFO(this->get_logger(), "Feedback: %d", i);
      
      // 模拟耗时操作
      loop_rate.sleep();
    }
    
    // 任务完成，设置结果
    result->success = true;
    result->message = "Countdown finished successfully.";
    goal_handle->succeed(result);
    RCLCPP_INFO(this->get_logger(), "Goal succeeded.");
  }
};

动作服务器的逻辑比服务复杂，因为它要管理任务的生命周期。它需要三个回调函数：handle_goal 决定是否接受一个新任务；handle_cancel 处理取消请求；handle_accepted 在任务被接受后，启动实际执行。关键点在于，执行函数 execute 通常要在一个独立的线程中运行，这样才能在执行过程中持续检查取消信号、发布反馈，而不阻塞主回调线程。反馈通过 goal_handle->publish_feedback 发送，最终结果通过 goal_handle->succeed（或 canceled、abort）发送。

再来看客户端 countdown_action_client.cpp：

#include "rclcpp/rclcpp.hpp"
#include "rclcpp_action/rclcpp_action.hpp"
#include "my_interface/action/countdown.hpp"
#include <future>

using Countdown = my_interface::action::Countdown;

class CountdownActionClient : public rclcpp::Node
{
public:
  CountdownActionClient() : Node("countdown_action_client"), goal_done_(false)
  {
    this->client_ = rclcpp_action::create_client<Countdown>(this, "countdown");
    
    // 定时发送目标
    this->timer_ = this->create_wall_timer(
      std::chrono::seconds(5),
      std::bind(&CountdownActionClient::send_goal, this));
  }

private:
  rclcpp_action::Client<Countdown>::SharedPtr client_;
  rclcpp::TimerBase::SharedPtr timer_;
  bool goal_done_;
  
  void send_goal()
  {
    using namespace std::placeholders;
    
    this->timer_->cancel(); // 只发送一次
    
    if (!this->client_->wait_for_action_server(std::chrono::seconds(10))) {
      RCLCPP_ERROR(this->get_logger(), "Action server not available after waiting");
      this->goal_done_ = true;
      return;
    }
    
    auto goal_msg = Countdown::Goal();
    goal_msg.start_number = 10; // 从10开始倒计时
    
    RCLCPP_INFO(this->get_logger(), "Sending goal: countdown from %d", goal_msg.start_number);
    
    // 设置发送目标时的选项：允许反馈回调
    auto send_goal_options = rclcpp_action::Client<Countdown>::SendGoalOptions();
    send_goal_options.feedback_callback =
      std::bind(&CountdownActionClient::feedback_callback, this, _1, _2);
    send_goal_options.result_callback =
      std::bind(&CountdownActionClient::result_callback, this, _1);
    
    // 异步发送目标
    auto future_goal_handle = this->client_->async_send_goal(goal_msg, send_goal_options);
  }
  
  // 反馈回调
  void feedback_callback(
    rclcpp_action::ClientGoalHandle<Countdown>::SharedPtr,
    const std::shared_ptr<const Countdown::Feedback> feedback)
  {
    RCLCPP_INFO(this->get_logger(), "Feedback received: %d", feedback->current_number);
  }
  
  // 结果回调
  void result_callback(const rclcpp_action::ClientGoalHandle<Countdown>::WrappedResult & result)
  {
    this->goal_done_ = true;
    switch (result.code) {
      case rclcpp_action::ResultCode::SUCCEEDED:
        RCLCPP_INFO(this->get_logger(), "Goal succeeded! Message: %s", result.result->message.c_str());
        break;
      case rclcpp_action::ResultCode::ABORTED:
        RCLCPP_ERROR(this->get_logger(), "Goal was aborted");
        break;
      case rclcpp_action::ResultCode::CANCELED:
        RCLCPP_WARN(this->get_logger(), "Goal was canceled");
        break;
      default:
        RCLCPP_ERROR(this->get_logger(), "Unknown result code");
        break;
    }
  }
};

动作客户端同样采用异步模式。它通过 async_send_goal 发送目标，并可以注册两个重要的回调：feedback_callback 用于实时接收进度反馈；result_callback 在任务最终完成（成功、取消或失败）时被调用。这种设计使得客户端节点可以非常灵活地监控和管理长时间运行的任务。

动作是构建复杂机器人行为（如导航、机械臂抓取）的基石。它把任务执行、状态监控和用户控制优雅地结合在了一起。在实际项目中，我经常用动作来实现机器人的任务队列，每个任务都可以被独立监控和干预，大大提升了系统的可控性和可靠性。

5. 性能优化实战：让你的节点跑得更快更稳

当你掌握了基本的通信方式后，下一个挑战就是如何让节点在资源受限的嵌入式平台或高负载场景下依然高效稳定。ROS2的默认设置为了通用性和易用性，在某些方面做了权衡。这里分享几个我踩过坑后总结的实战优化技巧。

第一，选择合适的服务质量（QoS）策略。 这是ROS2相较于ROS1一个巨大的进步，也是性能调优的第一把钥匙。QoS策略定义了消息传输的可靠性、持久性和历史记录方式。创建发布者、订阅者、客户端、服务时，都可以传入一个 rclcpp::QoS 对象。默认的QoS是“尽力而为”的传输，可能会丢消息。对于关键的控制指令，我们必须使用可靠传输。

// 创建一个使用可靠传输、保留最后一条消息的QoS配置
rclcpp::QoS qos_profile(10); // 队列深度10
qos_profile.reliability(RMW_QOS_POLICY_RELIABILITY_RELIABLE); // 可靠传输
qos_profile.durability(RMW_QOS_POLICY_DURABILITY_TRANSIENT_LOCAL); // 持久化，新订阅者能收到最后一条消息

auto pub = node->create_publisher<std_msgs::msg::String>("cmd_vel", qos_profile);

对于像激光雷达（LiDAR）点云这种数据量巨大、但偶尔丢一帧无伤大雅的数据，则可以使用尽力而为（Best Effort） 传输，能显著降低延迟和CPU占用。

rclcpp::QoS sensor_qos(5); // 小队列
sensor_qos.reliability(RMW_QOS_POLICY_RELIABILITY_BEST_EFFORT);
sensor_qos.durability(RMW_QOS_POLICY_DURABILITY_VOLATILE);
auto sub = node->create_subscription<sensor_msgs::msg::LaserScan>("scan", sensor_qos, callback);

第二，慎用 rclcpp::spin_some 与多线程执行器。 默认的 rclcpp::spin(node) 会阻塞当前线程，直到节点关闭。它使用一个单线程执行器来处理所有回调（定时器、订阅、服务等）。如果某个回调函数执行时间很长（比如一个复杂的图像处理算法），它会阻塞所有其他回调，导致节点“卡住”。解决方案有两种：

  1. 使用 spin_some 或 spin_once 进行手动循环，在循环间隙插入你自己的逻辑或检查。但这需要你自己管理循环频率和资源，比较繁琐。
  2. 使用多线程执行器（MultiThreadedExecutor）。这是更推荐的方式。它允许你指定一个线程池，不同的回调可以被分配到不同的线程中并发执行。
#include "rclcpp/executors/multi_threaded_executor.hpp"

int main(int argc, char **argv)
{
  rclcpp::init(argc, argv);
  auto node1 = std::make_shared<MyNode1>();
  auto node2 = std::make_shared<MyNode2>();
  
  rclcpp::executors::MultiThreadedExecutor executor;
  executor.add_node(node1);
  executor.add_node(node2);
  
  // 使用4个线程的池子来执行回调
  executor.spin();
  
  rclcpp::shutdown();
  return 0;
}

第三，优化消息传递：使用零拷贝与循环池。 在高速数据传输场景（比如视频流），消息的序列化、反序列化和拷贝会成为性能瓶颈。ROS2支持零拷贝（Zero-Copy） 传输，前提是使用共享内存中间件（如Cyclone DDS的共享内存插件）并且发布者和订阅者在同一台机器上。这需要特定的DDS配置，比较复杂。

一个更实用的技巧是复用消息对象，避免在每次回调或发布时都创建新的消息。我们可以预分配一个消息对象，在回调中只更新其字段内容。

// 在类成员中预分配一个消息
std_msgs::msg::String::SharedPtr shared_msg_;

// 在构造函数中初始化
shared_msg_ = std::make_shared<std_msgs::msg::String>();

// 在定时器回调中复用
void timer_callback() {
  shared_msg_->data = "Hello";
  publisher_->publish(*shared_msg_); // 注意这里解引用
}

第四，合理使用组件（Composition）与生命周期节点。 ROS2鼓励将功能拆分为独立的组件（Component），然后通过进程内组合（In-Process Composition）的方式，让多个组件运行在同一个进程中。这能彻底消除进程间通信（IPC）的开销，极大提升性能。你可以把组件想象成动态链接库（.so文件），主程序在运行时加载并管理它们。此外，ROS2还引入了生命周期节点（LifecycleNode），让节点拥有明确的非激活（Unconfigured）、激活（Active）等状态，便于系统管理。在需要严格启动顺序和状态管理的复杂系统中，生命周期节点非常有用。

性能优化没有银弹，需要根据具体场景进行测量和调整。我常用的工具是 ros2 topic hz 查看发布频率，ros2 topic bw 查看带宽，以及系统的 top 或 htop 命令观察CPU和内存占用。在真实机器人上部署前，一定要在类似硬件环境下进行压力测试，找到系统的瓶颈所在。

Logo
DAMO开发者矩阵

DAMO开发者矩阵，由阿里巴巴达摩院和中国互联网协会联合发起，致力于探讨最前沿的技术趋势与应用成果，搭建高质量的交流与分享平台，推动技术创新与产业应用链接，围绕“人工智能与新型计算”构建开放共享的开发者生态。

更多推荐

cover

【颠覆传统 VLN】RSS2024 NaVid 阅读｜RGB 视频 VLM 做连续视觉语言导航，不靠地图里程计

avatar DAMO开发者矩阵

作业：Vlan间路由

此为PC1运行结果，其余pc都可以与另一个pc或端口连接。

avatar DAMO开发者矩阵
cover

世界模型管预测，VLA管动作：从闫维新的技术路径对照到具身智能的三大落地瓶颈，两条路线为什么是互补而不是二选一

avatar DAMO开发者矩阵