第二章：ROS2核心概念——节点、话题、服务、动作、参数

各位嵌入式同仁，欢迎来到第二章。说实话，ROS2 的概念体系，是很多初学者第一道坎。我记得刚接触 ROS1 时，被节点、话题、服务这几个词绕得晕头转向。后来做项目多了，才慢慢摸清它们的脾气。

这一章，我会把 ROS2 的五个核心概念——节点、话题、服务、动作、参数，掰开揉碎了讲。每个概念我都会结合嵌入式场景，说说我踩过的坑。

2.1 节点：ROS2 的最小执行单元

节点是什么？说白了，就是一个独立的可执行进程。每个节点负责一件事。比如，一个节点读传感器，另一个节点控制电机，第三个节点做决策。

我习惯把节点想象成一个「小工人」。每个工人只干自己的活，不操心别人。这样系统解耦，好维护。

关键点：ROS2 是分布式架构。节点可以运行在同一块芯片上，也可以分布在多台机器上。节点之间通过「通信总线」交换数据。

创建节点很简单。用 Python 的话，几行代码搞定：

import rclpy
from rclpy.node import Node

class MyNode(Node):
    def __init__(self):
        super().__init__('my_node')
        self.get_logger().info('节点已启动！')

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

嗯，这里要注意：rclpy.spin() 会让节点一直运行，处理回调。如果你在嵌入式板子上跑，记得控制好 CPU 占用。

2.2 话题：异步通信的「广播」模式

话题是 ROS2 最常用的通信方式。一个节点发布消息，其他节点订阅。发布者和订阅者互不知道对方存在。

我在项目中遇到过一个问题：两个传感器节点同时往同一个话题发数据，结果数据打架了。后来我加了时间戳，才解决。

话题通信是异步的。发布者只管发，不管有没有人收。订阅者只管收，不管谁发的。这种模式适合传感器数据、状态信息等周期性数据。

举个例子，发布话题：

class PublisherNode(Node):
    def __init__(self):
        super().__init__('publisher')
        self.pub = self.create_publisher(String, 'chatter', 10)
        self.timer = self.create_timer(1.0, self.timer_callback)

    def timer_callback(self):
        msg = String()
        msg.data = 'Hello, ROS2!'
        self.pub.publish(msg)

订阅话题：

class SubscriberNode(Node):
    def __init__(self):
        super().__init__('subscriber')
        self.sub = self.create_subscription(String, 'chatter', self.callback, 10)

    def callback(self, msg):
        self.get_logger().info(f'收到: {msg.data}')

小技巧：话题的 QoS（服务质量）设置很重要。我建议对传感器数据用「可靠」模式，对控制指令用「尽力而为」模式，减少延迟。

2.3 服务：同步通信的「请求-应答」模式

服务跟话题不一样。服务是同步的。客户端发请求，服务端处理完返回结果。客户端必须等。

你想想看，什么时候用服务？比如，你让机器人「打开摄像头」，它打开后告诉你「已打开」。这种一次性的、需要确认的操作，用服务最合适。

我曾经犯过一个错：用话题去实现「打开摄像头」的功能。结果发布者发了指令，不知道摄像头到底开了没。后来改成服务，问题解决。

定义服务接口（.srv 文件）：

int64 a
int64 b
---
int64 sum

服务端代码：

class ServiceNode(Node):
    def __init__(self):
        super().__init__('service_node')
        self.srv = self.create_service(AddTwoInts, 'add_two_ints', self.add_callback)

    def add_callback(self, request, response):
        response.sum = request.a + request.b
        return response

客户端代码：

class ClientNode(Node):
    def __init__(self):
        super().__init__('client_node')
        self.cli = self.create_client(AddTwoInts, 'add_two_ints')
        while not self.cli.wait_for_service(timeout_sec=1.0):
            self.get_logger().info('等待服务...')
        req = AddTwoInts.Request()
        req.a = 3
        req.b = 5
        future = self.cli.call_async(req)
        rclpy.spin_until_future_complete(self, future)
        self.get_logger().info(f'结果: {future.result().sum}')

注意：服务调用是阻塞的。在嵌入式系统里，如果服务处理时间太长，会卡住整个节点。我建议用异步调用（call_async），或者把耗时操作放到单独线程。

2.4 动作：带反馈的「长任务」模式

动作是 ROS2 里最复杂的概念，但也是最强大的。它适合执行时间长的任务，比如「导航到某个点」、「抓取物体」。

动作有三个阶段：

  • 目标：客户端发送目标
  • 反馈：服务端不断返回进度
  • 结果：任务完成，返回最终结果

我记得第一次用动作时，觉得它跟服务差不多。后来做机器人底盘控制才发现，动作的「可取消」特性太重要了。用户随时可以喊停。

定义动作接口（.action 文件）：

# 目标
int32 target_x
int32 target_y
---
# 结果
bool success
string message
---
# 反馈
int32 current_x
int32 current_y
float32 progress

动作服务端示例：

class MoveActionServer(Node):
    def __init__(self):
        super().__init__('move_action_server')
        self._action_server = ActionServer(
            self,
            Move,
            'move_robot',
            self.execute_callback)

    def execute_callback(self, goal_handle):
        feedback_msg = Move.Feedback()
        # 模拟移动过程
        for i in range(10):
            feedback_msg.current_x = i
            feedback_msg.current_y = i
            feedback_msg.progress = i / 10.0
            goal_handle.publish_feedback(feedback_msg)
            time.sleep(0.5)
        goal_handle.succeed()
        result = Move.Result()
        result.success = True
        result.message = '到达目标'
        return result

核心区别：话题是「广播」，服务是「问答」，动作是「任务」。选哪个，取决于你的场景。

2.5 参数：节点的「配置开关」

参数是节点的配置项。比如，摄像头帧率、PID 系数、串口波特率。参数可以在启动时设置，也可以在运行时动态修改。

我习惯把参数看作「旋钮」。你可以在系统运行时拧一拧，看看效果。不用重新编译代码。

声明参数：

self.declare_parameter('camera_fps', 30)
self.declare_parameter('pid_kp', 1.0)
self.declare_parameter('serial_baud', 115200)

读取参数：

fps = self.get_parameter('camera_fps').value
kp = self.get_parameter('pid_kp').value

动态修改参数（命令行）：

ros2 param set /my_node camera_fps 60

建议：把硬件相关的配置（如 GPIO 引脚号、I2C 地址）都做成参数。这样换硬件时，改参数就行，不用改代码。

2.6 五大概念的关系图

说了这么多，它们之间到底是什么关系？我画个简单的图帮你理解：

┌─────────────────────────────────────────────────────┐
│                     ROS2 系统                        │
│                                                     │
│  ┌──────────┐   话题    ┌──────────┐               │
│  │ 传感器节点 │ ────────→ │ 控制节点  │               │
│  └──────────┘           └──────────┘               │
│       │                                              │
│       │ 服务                                         │
│       ↓                                              │
│  ┌──────────┐   动作    ┌──────────┐               │
│  │ 执行器节点 │ ←───────→ │ 决策节点  │               │
│  └──────────┘           └──────────┘               │
│       │                                              │
│       │ 参数                                         │
│       ↓                                              │
│  ┌──────────┐                                       │
│  │ 参数服务器 │                                       │
│  └──────────┘                                       │
└─────────────────────────────────────────────────────┘

简单总结：

  • 节点是工人
  • 话题是广播喇叭
  • 服务是电话问答
  • 动作是带进度条的任务
  • 参数是旋钮开关

在实际项目中，这五种通信方式经常混合使用。比如，传感器数据用话题传，控制指令用服务发，导航任务用动作执行，系统配置用参数调。

好了，这一章就到这里。下一章我们会深入话题的 QoS 设置，那是很多嵌入式工程师容易忽略的地方。到时候我会分享一个因为 QoS 设置不当导致丢数据的真实案例。

记住：理解这些概念，是 ROS2 系统设计的第一步。别急，慢慢来。

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