ROS 2 Jazzy 从入门到精通：节点、话题与服务全链路实战指南

在机器人开发的世界里，ROS 2 已经成为事实上的标准框架。它就像机器人的"神经系统"，负责连接各个硬件模块和软件组件，让它们协同工作。本文将带你从零开始，完整掌握 ROS 2 最核心的三个概念：节点（Node）、话题（Topic） 和 服务（Service），并通过实战项目巩固所学知识。

一、ROS 2 节点：机器人世界的"最小执行单元"

在 ROS 2 的世界里，程序就是节点。所有的功能，无论是传感器数据采集、路径规划还是电机控制，都以节点的形式存在。节点之间通过通信机制互相交换信息，共同构成一个完整的机器人应用。

1.1 工作空间与包：代码的"家"

在写代码之前，我们需要先了解 ROS 2 的工程组织方式。ROS 2 采用三层结构：

• 工作空间（Workspace）：整个项目的根目录
• 包（Package）：功能模块的容器
• 节点（Node）：最小的可执行单元

这种结构就像一个公司：工作空间是整个公司，包是各个部门，节点是每个员工。每个员工（节点）负责一个具体的任务，部门（包）负责管理相关的员工，公司（工作空间）负责协调所有部门。

创建工作空间

# 在家目录下创建工作空间
cd ~
mkdir ros2_ws
cd ros2_ws
mkdir src

# 即使是空的，也可以构建一次，生成必要的目录结构
colcon build

构建完成后，工作空间会多出三个目录：

• build/：构建过程的中间文件
• log/：构建日志
• install/：最重要，构建完成后的可执行文件会被安装到这里

配置环境变量

# 临时生效（当前终端）
source ~/ros2_ws/install/setup.bash

# 永久生效（所有终端自动加载）
echo "source /opt/ros/jazzy/setup.bash" >> ~/.bashrc
echo "source ~/ros2_ws/install/setup.bash" >> ~/.bashrc
source ~/.bashrc

通俗解释：环境变量就像系统的"通讯录"。source 命令就是告诉系统：“我在这个目录下有一些可执行文件，以后你找命令的时候也来这里看看”。

创建包

这两条命令就是给你搭好了 ROS 2 开发的「脚手架」，让你可以直接在里面写 Python 或 C++ 代码，实现机器人的各种功能，因为有 rclpy和rclcpp的依赖，它天生具备 ROS 2 的核心能力。

# 创建 Python 包
cd ~/ros2_ws/src
ros2 pkg create my_py_pkg --build-type ament_python --dependencies rclpy

# 创建 C++ 包
ros2 pkg create my_cpp_pkg --build-type ament_cmake --dependencies rclcpp

1.2 Python 节点编写：从 Hello World 到定时器

一个最小的 ROS 2 节点只需要四步：初始化 ROS 2、创建节点对象、让节点"活着"（spin）、退出前关闭 ROS 2。

最小节点模板

#!/usr/bin/env python3
import rclpy
from rclpy.node import Node

class MyCustomNode(Node):
    def __init__(self):
        # 调用父类构造函数，设置节点名
        super().__init__("node_name")
        self.get_logger().info("Hello ROS 2!")

def main(args=None):
    # 1. 初始化 ROS 2 通信
    rclpy.init(args=args)
    
    # 2. 创建节点对象
    node = MyCustomNode()
    
    # 3. 让节点"活着"，处理所有回调
    rclpy.spin(node)
    
    # 4. 退出前关闭 ROS 2 通信
    rclpy.shutdown()

if __name__ == "__main__":
    main()

配置可执行入口

在 my_py_pkg/setup.py 的 entry_points 中添加：

entry_points={
    'console_scripts': [
        "test_node = my_py_pkg.my_first_node:main",
    ],
},

构建并运行

cd ~/ros2_ws
colcon build --packages-select my_py_pkg
source ~/.bashrc
ros2 run my_py_pkg test_node

加入定时器：每秒打印一次

class MyCustomNode(Node):
    def __init__(self):
        super().__init__("my_node_name")
        self.counter_ = 0
        
        # 创建定时器，每秒调用一次 print_hello 函数
        self.timer_ = self.create_timer(1.0, self.print_hello)
        
    def print_hello(self):
        self.get_logger().info(f"Hello {self.counter_}")
        self.counter_ += 1

定时器的周期公式：
$$T=\frac{1}{f}$$
其中：

• $T$：定时器周期（秒）
• $f$：定时器频率（赫兹）

例如，create_timer(1.0, callback) 表示频率为 1Hz，周期为 1 秒。

1.3 C++ 节点编写：同样的逻辑，不同的语法

C++ 节点的结构和 Python 完全一致，只是语法不同。

#include "rclcpp/rclcpp.hpp"
#include <chrono>
#include <functional>

class MyCustomNode : public rclcpp::Node
{
public:
  MyCustomNode() : Node("my_node_name"), counter_(0) {
    // 创建定时器，每秒调用一次 print_hello 函数
    timer_ = this->create_wall_timer(
      std::chrono::seconds(1),
      std::bind(&MyCustomNode::print_hello, this)
    );
  }

private:
  void print_hello() {
    RCLCPP_INFO(this->get_logger(), "Hello %d", counter_);
    counter_++;
  }

  int counter_;
  rclcpp::TimerBase::SharedPtr timer_;
};

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

配置 CMakeLists.txt

在 my_cpp_pkg/CMakeLists.txt 中添加：

add_executable(test_node src/my_first_node.cpp)
ament_target_dependencies(test_node rclcpp)
install(TARGETS test_node DESTINATION lib/${PROJECT_NAME}/)

1.4 节点调试工具：给你的节点做"体检"

ROS 2 提供了强大的命令行工具来调试节点：

# 列出所有正在运行的节点
ros2 node list

# 查看节点详情（发布/订阅的话题、提供的服务等）
ros2 node info /my_node_name

# 运行时修改节点名
ros2 run my_py_pkg test_node --ros-args -r __node:=new_node_name

二、ROS 2 话题：节点之间的"无线电台"

节点写好了，接下来的问题是：多个节点怎么互相通信？ROS 2 提供了三种通信方式：话题（Topic）、服务（Service）和动作（Action）。其中话题是最常用、最基础的一种。

2.1 话题机制：发布订阅模型详解

话题通信采用发布-订阅模型，就像我们听收音机：

• 电台（发布者）持续往某个频率（话题名）发送信号
• 收音机（订阅者）调到这个频率就能收到信号
• 一个电台可以有多个听众，一个听众也可以听多个电台

在 ROS 2 中，话题由两部分组成：

• 名字：唯一标识这个话题
• 类型：话题中传输的数据格式

只有名字和类型都相同的发布者和订阅者才能通信。

2.2 发布者编写：每秒发布一个数字

我们来写一个发布者节点，每秒在 /number 话题上发布一个整数。

#!/usr/bin/env python3
import rclpy
from rclpy.node import Node
from example_interfaces.msg import Int64

class NumberPublisherNode(Node):
    def __init__(self):
        super().__init__("number_publisher")
        self.number_ = 2
        
        # 创建发布者：类型 Int64，话题名 "number"，队列大小 10
        self.number_publisher_ = self.create_publisher(Int64, "number", 10)
        
        # 创建定时器，每秒发布一次
        self.number_timer_ = self.create_timer(1.0, self.publish_number)
        
        self.get_logger().info("Number publisher has been started.")
    
    def publish_number(self):
        # 创建消息对象
        msg = Int64()
        # 填充消息字段
        msg.data = self.number_
        # 发布消息
        self.number_publisher_.publish(msg)

def main(args=None):
    rclpy.init(args=args)
    node = NumberPublisherNode()
    rclpy.spin(node)
    rclpy.shutdown()

if __name__ == "__main__":
    main()

配置依赖和可执行入口

在 my_py_pkg/package.xml 中添加依赖：

<depend>example_interfaces</depend>

在 setup.py 的 console_scripts 中添加：

"number_publisher = my_py_pkg.number_publisher:main",

验证发布者

# 运行发布者
ros2 run my_py_pkg number_publisher

# 新开终端，订阅话题查看内容
ros2 topic echo /number

2.3 订阅者编写：收到数字就累加

接下来写一个订阅者节点，订阅 /number 话题，每收到一个数字就加到计数器上并打印。

#!/usr/bin/env python3
import rclpy
from rclpy.node import Node
from example_interfaces.msg import Int64

class NumberCounterNode(Node):
    def __init__(self):
        super().__init__("number_counter")
        self.counter_ = 0
        
        # 创建订阅者：类型 Int64，话题名 "number"，回调函数 callback_number，队列大小 10
        self.number_subscriber_ = self.create_subscription(
            Int64,
            "number",
            self.callback_number,
            10
        )
        
        self.get_logger().info("Number Counter has been started.")
    
    def callback_number(self, msg: Int64):
        # 处理收到的消息
        self.counter_ += msg.data
        self.get_logger().info(f"Counter: {self.counter_}")

def main(args=None):
    rclpy.init(args=args)
    node = NumberCounterNode()
    rclpy.spin(node)
    rclpy.shutdown()

if __name__ == "__main__":
    main()

运行验证

# 终端1：运行发布者
ros2 run my_py_pkg number_publisher

# 终端2：运行订阅者
ros2 run my_py_pkg number_counter

你会看到订阅者每秒打印一次计数器，数值每次增加 2。

2.4 话题调试工具：可视化与命令行

rqt_graph：可视化节点和话题连接

rqt_graph

运行后会弹出一个窗口，显示所有节点和话题的连接关系。这是调试通信问题最直观的工具。

ros2 topic 命令集

# 列出所有话题
ros2 topic list

# 查看话题详情（发布者、订阅者、类型）
ros2 topic info /number

# 查看消息类型的字段
ros2 interface show example_interfaces/msg/Int64

# 订阅话题并打印内容
ros2 topic echo /number

# 手动发布话题（不用写代码）
ros2 topic pub -r 2.0 /number example_interfaces/msg/Int64 "{data: 7}"

其中 ros2 topic pub 命令非常实用，-r 2.0 表示以 2Hz 的频率发布，{data: 7} 是消息内容。

2.5 自定义消息：让数据更有"语义"

虽然 ROS 2 提供了很多现成的消息类型（如 Int64、String），但在实际项目中，我们经常需要自定义更有语义的消息类型。

创建自定义接口包

最佳实践是单独建一个只放接口的包，不要在里面写节点。

cd ~/ros2_ws/src
ros2 pkg create my_robot_interfaces
cd my_robot_interfaces
rm -r src/ include/
mkdir msg

修改 package.xml

在 <buildtool_depend>ament_cmake</buildtool_depend> 后添加：

<build_depend>rosidl_default_generators</build_depend>
<exec_depend>rosidl_default_runtime</exec_depend>
<member_of_group>rosidl_interface_packages</member_of_group>

修改 CMakeLists.txt

在 find_package(ament_cmake REQUIRED) 后、ament_package() 前添加：

find_package(rosidl_default_generators REQUIRED)

rosidl_generate_interfaces(${PROJECT_NAME}
  "msg/HardwareStatus.msg"
)

ament_export_dependencies(rosidl_default_runtime)

编写自定义消息

创建 msg/HardwareStatus.msg：

int64 version
float64 temperature
bool are_motors_ready
string debug_message

构建并验证

cd ~/ros2_ws
colcon build --packages-select my_robot_interfaces
source ~/.bashrc

# 验证消息是否生成成功
ros2 interface show my_robot_interfaces/msg/HardwareStatus

2.6 实战：turtlesim 闭环控制

现在我们用 ROS 2 自带的 turtlesim 模拟器做一个有趣的实验：让乌龟根据自己的位置自动调整速度。

目标：

• 乌龟在屏幕左侧（x < 5.5）时，以低速画圈
• 乌龟在屏幕右侧（x ≥ 5.5）时，以高速画圈

#!/usr/bin/env python3
import rclpy
from rclpy.node import Node
from geometry_msgs.msg import Twist
from turtlesim.msg import Pose

class TurtleControllerNode(Node):
    def __init__(self):
        super().__init__("turtle_controller")
        
        # 创建速度指令发布者
        self.cmd_vel_pub_ = self.create_publisher(Twist, "/turtle1/cmd_vel", 10)
        
        # 创建位置订阅者
        self.pose_sub_ = self.create_subscription(
            Pose,
            "/turtle1/pose",
            self.callback_pose,
            10
        )
        
        self.get_logger().info("Turtle controller has been started.")
    
    def callback_pose(self, pose: Pose):
        cmd = Twist()
        
        # 根据位置调整速度
        if pose.x < 5.5:
            cmd.linear.x = 1.0
            cmd.angular.z = 1.0
        else:
            cmd.linear.x = 2.0
            cmd.angular.z = 2.0
        
        # 发布速度指令
        self.cmd_vel_pub_.publish(cmd)

def main(args=None):
    rclpy.init(args=args)
    node = TurtleControllerNode()
    rclpy.spin(node)
    rclpy.shutdown()

if __name__ == "__main__":
    main()

运行验证

# 终端1：启动 turtlesim
ros2 run turtlesim turtlesim_node

# 终端2：运行控制器
ros2 run my_py_pkg turtle_controller

你会看到乌龟在屏幕上画圈，并且在左右两侧速度明显不同。这就是一个最简单的闭环控制系统：传感器（位置）→ 控制器 → 执行器（电机）。

三、ROS 2 服务：节点之间的"远程调用"

话题适合传输持续的数据流，但有时候我们需要"请求-响应"式的交互，比如"启动电机"、“重置计数器”。这时候就需要用到服务（Service）。

3.1 服务机制：请求响应模型详解

服务通信采用客户端-服务器模型，就像我们访问网站：

• 浏览器（客户端）发送请求（比如"打开百度首页"）
• 服务器收到请求，处理并返回响应（百度首页的 HTML）
• 一次交互结束

在 ROS 2 中，服务由两部分组成：

• 名字：唯一标识这个服务
• 接口：包含请求（Request）和响应（Response）两部分

3.2 自定义服务接口：定义你的"API"

我们继续沿用上一篇的 number_counter 节点，给它加一个功能：允许外部通过服务把计数器重置成某个值。

创建服务接口

在 my_robot_interfaces 包中创建 srv 目录：

cd ~/ros2_ws/src/my_robot_interfaces
mkdir srv

创建 srv/ResetCounter.srv：

int64 reset_value
---
bool success
string message

注意：--- 是分隔符，上面是请求字段，下面是响应字段。

修改 CMakeLists.txt

在 rosidl_generate_interfaces 中添加服务文件：

rosidl_generate_interfaces(${PROJECT_NAME}
  "msg/HardwareStatus.msg"
  "srv/ResetCounter.srv"
)

构建并验证

cd ~/ros2_ws
colcon build --packages-select my_robot_interfaces
source ~/.bashrc

# 验证服务是否生成成功
ros2 interface show my_robot_interfaces/srv/ResetCounter

3.3 服务端编写：处理请求并返回响应

我们在 number_counter 节点中添加服务端，处理重置计数器的请求。

#!/usr/bin/env python3
import rclpy
from rclpy.node import Node
from example_interfaces.msg import Int64
from my_robot_interfaces.srv import ResetCounter

class NumberCounterNode(Node):
    def __init__(self):
        super().__init__("number_counter")
        self.counter_ = 0
        
        # 订阅者
        self.number_subscriber_ = self.create_subscription(
            Int64,
            "number",
            self.callback_number,
            10
        )
        
        # 创建服务端：类型 ResetCounter，服务名 "reset_counter"，回调函数 callback_reset_counter
        self.reset_counter_service_ = self.create_service(
            ResetCounter,
            "reset_counter",
            self.callback_reset_counter
        )
        
        self.get_logger().info("Number Counter has been started.")
    
    def callback_number(self, msg: Int64):
        self.counter_ += msg.data
        self.get_logger().info(f"Counter: {self.counter_}")
    
    def callback_reset_counter(self, request: ResetCounter.Request, response: ResetCounter.Response):
        # 校验请求参数
        if request.reset_value < 0:
            response.success = False
            response.message = "Cannot reset counter to a negative value"
        elif request.reset_value > self.counter_:
            response.success = False
            response.message = "Reset value must be lower than current counter value"
        else:
            self.counter_ = request.reset_value
            self.get_logger().info(f"Reset counter to {self.counter_}")
            response.success = True
            response.message = "Success"
        
        # Python 中必须返回 response
        return response

def main(args=None):
    rclpy.init(args=args)
    node = NumberCounterNode()
    rclpy.spin(node)
    rclpy.shutdown()

if __name__ == "__main__":
    main()

3.4 客户端编写：发送请求并处理结果

接下来写一个客户端节点，调用 reset_counter 服务。

#!/usr/bin/env python3
import rclpy
from rclpy.node import Node
from my_robot_interfaces.srv import ResetCounter

class ResetCounterClientNode(Node):
    def __init__(self):
        super().__init__("reset_counter_client")
        
        # 创建客户端：类型 ResetCounter，服务名 "reset_counter"
        self.client_ = self.create_client(ResetCounter, "reset_counter")
        
        self.get_logger().info("Reset counter client has been started.")
    
    def call_reset_counter(self, value):
        # 等待服务端上线
        while not self.client_.wait_for_service(1.0):
            self.get_logger().warn("Waiting for service...")
        
        # 创建请求对象
        request = ResetCounter.Request()
        request.reset_value = value
        
        # 异步发送请求，添加回调函数处理响应
        future = self.client_.call_async(request)
        future.add_done_callback(self.callback_reset_counter_response)
    
    def callback_reset_counter_response(self, future):
        # 获取响应
        response = future.result()
        self.get_logger().info(f"Success flag: {response.success}")
        self.get_logger().info(f"Message: {response.message}")

def main(args=None):
    rclpy.init(args=args)
    node = ResetCounterClientNode()
    
    # 调用服务
    node.call_reset_counter(20)
    
    rclpy.spin(node)
    rclpy.shutdown()

if __name__ == "__main__":
    main()

运行验证

# 终端1：运行发布者
ros2 run my_py_pkg number_publisher

# 终端2：运行计数器（包含服务端）
ros2 run my_py_pkg number_counter

# 终端3：运行客户端
ros2 run my_py_pkg reset_counter_client

根据当前计数器的值，你会看到不同的响应。如果计数器大于 20，会重置成功；否则会返回失败。

3.5 服务调试工具：命令行调用服务

ROS 2 提供了 ros2 service 命令来调试服务：

# 列出所有服务
ros2 service list

# 查看服务类型
ros2 service type /reset_counter

# 查看服务接口
ros2 interface show my_robot_interfaces/srv/ResetCounter

# 从终端直接调用服务
ros2 service call /reset_counter my_robot_interfaces/srv/ResetCounter "{reset_value: 7}"

3.6 实战：turtlesim 画笔颜色与启停控制

我们继续扩展 turtlesim 控制器，添加两个服务相关的功能：

1. 乌龟在左侧时画笔变绿，右侧时变红
2. 提供一个服务，允许外部控制乌龟的启停

自定义启停服务接口

创建 my_robot_interfaces/srv/ActivateTurtle.srv：

bool activate
---
string message

修改 CMakeLists.txt 并重新构建。

完整控制器代码

#!/usr/bin/env python3
import rclpy
from rclpy.node import Node
from geometry_msgs.msg import Twist
from turtlesim.msg import Pose
from turtlesim.srv import SetPen
from my_robot_interfaces.srv import ActivateTurtle

class TurtleControllerNode(Node):
    def __init__(self):
        super().__init__("turtle_controller")
        self.is_active_ = True
        self.last_pen_color_ = None
        
        # 速度指令发布者
        self.cmd_vel_pub_ = self.create_publisher(Twist, "/turtle1/cmd_vel", 10)
        
        # 位置订阅者
        self.pose_sub_ = self.create_subscription(
            Pose,
            "/turtle1/pose",
            self.callback_pose,
            10
        )
        
        # 画笔颜色服务客户端
        self.set_pen_client_ = self.create_client(SetPen, "/turtle1/set_pen")
        
        # 启停服务端
        self.activate_turtle_service_ = self.create_service(
            ActivateTurtle,
            "activate_turtle",
            self.callback_activate_turtle
        )
        
        self.get_logger().info("Turtle controller has been started.")
    
    def callback_pose(self, pose: Pose):
        if not self.is_active_:
            return
        
        # 根据位置调整速度
        cmd = Twist()
        if pose.x < 5.5:
            cmd.linear.x = 1.0
            cmd.angular.z = 1.0
            self.set_pen_color(0, 255, 0)  # 绿色
        else:
            cmd.linear.x = 2.0
            cmd.angular.z = 2.0
            self.set_pen_color(255, 0, 0)  # 红色
        
        self.cmd_vel_pub_.publish(cmd)
    
    def set_pen_color(self, r, g, b):
        # 只有颜色变化时才调用服务，避免高频调用
        if (r, g, b) == self.last_pen_color_:
            return
        
        while not self.set_pen_client_.wait_for_service(1.0):
            self.get_logger().warn("Waiting for set_pen service...")
        
        request = SetPen.Request()
        request.r = r
        request.g = g
        request.b = b
        request.width = 2
        request.off = 0
        
        self.set_pen_client_.call_async(request)
        self.last_pen_color_ = (r, g, b)
    
    def callback_activate_turtle(self, request: ActivateTurtle.Request, response: ActivateTurtle.Response):
        self.is_active_ = request.activate
        
        if self.is_active_:
            response.message = "Turtle activated"
            self.get_logger().info("Turtle activated")
        else:
            response.message = "Turtle deactivated"
            self.get_logger().info("Turtle deactivated")
            # 发送零速度，让乌龟停下
            cmd = Twist()
            self.cmd_vel_pub_.publish(cmd)
        
        return response

def main(args=None):
    rclpy.init(args=args)
    node = TurtleControllerNode()
    rclpy.spin(node)
    rclpy.shutdown()

if __name__ == "__main__":
    main()

运行验证

# 终端1：启动 turtlesim
ros2 run turtlesim turtlesim_node

# 终端2：运行控制器
ros2 run my_py_pkg turtle_controller

# 终端3：测试启停服务
ros2 service call /activate_turtle my_robot_interfaces/srv/ActivateTurtle "{activate: false}"
ros2 service call /activate_turtle my_robot_interfaces/srv/ActivateTurtle "{activate: true}"

你会看到乌龟在左右两侧画出不同颜色的圈，并且可以通过服务控制它的启停。

四、三种通信方式对比与选择

ROS 2 提供了三种核心通信方式，它们各有优缺点，适用于不同的场景。

特性	话题（Topic）	服务（Service）	动作（Action）
通信模型	发布-订阅	请求-响应	目标-反馈-结果
数据流向	单向	双向	双向（带反馈）
适用场景	持续高频数据流（传感器、控制指令）	按需快速动作（启停、复位、配置）	耗时任务（导航、抓取）
实时性	高	中等	低
可靠性	可配置（QoS）	可靠	可靠
支持取消	不支持	不支持	支持
带进度反馈	不支持	不支持	支持

选择原则：

• 如果是持续的数据流，用话题
• 如果是一次性的请求-响应，用服务
• 如果是耗时的任务，需要进度反馈和取消功能，用动作

五、常用命令速查表

命令	功能
colcon build	构建工作空间
source install/setup.bash	配置环境变量
ros2 pkg create	创建包
ros2 run <pkg> <exec>	运行节点
ros2 node list	列出所有节点
ros2 node info <node>	查看节点详情
ros2 topic list	列出所有话题
ros2 topic echo <topic>	订阅并打印话题内容
ros2 topic pub <topic> <type> <msg>	手动发布话题
ros2 service list	列出所有服务
ros2 service call <service> <type> <req>	手动调用服务
ros2 interface show <type>	查看接口字段
rqt_graph	可视化节点和话题连接

总结

本文完整介绍了 ROS 2 最核心的三个概念：节点、话题和服务。我们从最基础的工作空间创建开始，一步步编写了 Python 和 C++ 节点，实现了话题的发布和订阅，自定义了消息和服务接口，并通过 turtlesim 模拟器进行了实战。

通过本文的学习，你应该已经掌握了 ROS 2 的基本开发流程，能够独立编写简单的 ROS 2 应用。接下来，你可以继续学习 ROS 2 的高级功能，如动作（Action）、参数（Parameter）、启动文件（Launch）和 QoS 配置，进一步提升你的机器人开发能力。