2、ROS2核心概念详解：节点、话题、服务、动作与参数

好，咱们直接进入正题。ROS2 里最核心的五个概念，说白了就是：节点、话题、服务、动作、参数。这五个东西你搞明白了，整个 ROS2 的骨架你就搭起来了。我刚开始学的时候，也绕了不少弯路，今天我把这些坑都给你指出来。

2.1 节点(Node)的创建与运行

节点是什么？你可以把它想象成一个独立的小程序，专门干一件事。比如一个节点负责读激光雷达数据，另一个节点负责做路径规划。它们之间互相通信，但各自独立运行。

创建一个节点，在 Python 里其实很简单。我习惯用面向对象的方式写，这样代码好维护。

import rclpy
from rclpy.node import Node

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

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

if __name__ == '__main__':
    main()

嗯，这里要注意一点：rclpy.spin() 这个函数会让节点一直运行，直到你按 Ctrl+C 停止。我在项目中遇到过有人忘了写这行，结果节点启动后立刻退出了，查了半天才发现。

2.2 话题(Topic)的发布与订阅

话题是 ROS2 里最常用的通信方式。它就像是一个广播频道：发布者往频道里发消息，订阅者接收消息。一个话题可以有多个发布者和多个订阅者，非常灵活。

举个例子，你要发布一个字符串消息：

class Talker(Node):
    def __init__(self):
        super().__init__('talker')
        self.publisher = self.create_publisher(String, 'chatter', 10)
        timer_period = 0.5  # 每秒发两次
        self.timer = self.create_timer(timer_period, self.timer_callback)

    def timer_callback(self):
        msg = String()
        msg.data = 'Hello, ROS2!'
        self.publisher.publish(msg)
        self.get_logger().info(f'发布了: {msg.data}')

订阅端怎么写呢？

class Listener(Node):
    def __init__(self):
        super().__init__('listener')
        self.subscription = self.create_subscription(
            String, 'chatter', self.listener_callback, 10)

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

你想想看，为什么要有队列大小（上面代码里的 10）？说白了就是防止消息堆积。如果订阅者处理得慢，队列满了，旧消息就会被丢掉。我在项目中遇到过传感器数据延迟的问题，后来发现就是队列设得太小了。

2.3 服务(Service)的客户端与服务器

话题是单向的，服务是双向的。服务就像打电话：客户端发起请求，服务器处理并返回响应。适合那些需要即时反馈的场景，比如让机器人抓取一个物体，你得知道它抓没抓到。

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

int64 a
int64 b
---
int64 sum

服务器端代码：

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

    def add_callback(self, request, response):
        response.sum = request.a + request.b
        self.get_logger().info(f'收到请求: {request.a} + {request.b} = {response.sum}')
        return response

客户端代码：

class AddClient(Node):
    def __init__(self):
        super().__init__('add_client')
        self.cli = self.create_client(AddTwoInts, 'add_two_ints')
        while not self.cli.wait_for_service(timeout_sec=1.0):
            self.get_logger().info('等待服务上线...')
        self.req = AddTwoInts.Request()

    def send_request(self, a, b):
        self.req.a = a
        self.req.b = b
        self.future = self.cli.call_async(self.req)

我个人习惯在客户端加一个等待服务的循环，不然服务还没启动，客户端就发请求，会报错。这个坑我踩过好几次。

2.4 动作(Action)的客户端与服务器

动作是服务的升级版。服务只能返回最终结果，动作可以实时反馈进度。比如让机器人走到某个位置，你可以实时知道它走到哪了。

动作有三个关键部分：目标(goal)、反馈(feedback)、结果(result)。

服务器端示例：

class MoveRobotServer(Node):
    def __init__(self):
        super().__init__('move_robot_server')
        self.action_server = ActionServer(
            self, MoveRobot, 'move_robot', self.execute_callback)

    async def execute_callback(self, goal_handle):
        feedback_msg = MoveRobot.Feedback()
        for i in range(1, 11):
            feedback_msg.progress = i * 10
            goal_handle.publish_feedback(feedback_msg)
            time.sleep(1)
        goal_handle.succeed()
        result = MoveRobot.Result()
        result.success = True
        return result

客户端代码：

class MoveRobotClient(Node):
    def __init__(self):
        super().__init__('move_robot_client')
        self.action_client = ActionClient(self, MoveRobot, 'move_robot')

    def send_goal(self, target_x, target_y):
        goal_msg = MoveRobot.Goal()
        goal_msg.target_x = target_x
        goal_msg.target_y = target_y
        self.action_client.send_goal_async(goal_msg)

动作的反馈机制特别实用。我记得有一次调试机器人导航，就是靠动作反馈发现某个路径点一直达不到，才找到算法里的 bug。

2.5 参数(Parameter)的声明与使用

参数就是节点的配置项。比如机器人的最大速度、传感器的采样频率，这些都可以做成参数，方便运行时调整。

声明参数：

class ParamNode(Node):
    def __init__(self):
        super().__init__('param_node')
        self.declare_parameter('max_speed', 1.0)
        self.declare_parameter('robot_name', 'turtlebot')
        self.declare_parameter('enable_debug', False)

读取参数：

max_speed = self.get_parameter('max_speed').value
robot_name = self.get_parameter('robot_name').value
enable_debug = self.get_parameter('enable_debug').value

动态修改参数：

self.set_parameters([Parameter('max_speed', Parameter.Type.DOUBLE, 2.0)])

你想想看，参数的好处是什么？不用改代码就能调配置。我曾经为了调一个 PID 参数，每次改完都要重新编译，后来改成参数后，直接在命令行里改，效率高多了。

好了，这五个核心概念就讲完了。节点是基础，话题、服务、动作是通信方式，参数是配置手段。把它们组合起来，你就能搭建出复杂的机器人系统。下一章我们讲如何用 launch 文件把这些节点组织起来，到时候见。