3、ROS2核心概念实战：节点(Node)、话题(Topic)、服务(Service)、动作(Action)的代码实现

好，咱们直接进入正题。

ROS2的四大核心概念——节点、话题、服务、动作，说白了就是机器人程序之间互相通信的四种方式。你想想看，一个自动驾驶系统里，感知模块要给规划模块传数据，控制模块要下发指令给底盘，这些通信怎么搞？就是靠这四兄弟。

我个人习惯把这四个概念分成两类：流式通信和请求-应答式通信。话题是流式的，服务是请求-应答的，动作则是两者的结合体。嗯，咱们一个一个来。

3.1 节点(Node)：一切的基础

节点，就是ROS2里最小的执行单元。每个节点负责一个具体的功能，比如读取激光雷达数据、控制电机、发布车辆状态等等。

我记得刚接触ROS2时，总想把所有功能写在一个大文件里。后来被坑惨了——调试起来简直噩梦。所以我的建议是：一个节点只干一件事。

核心要点：节点之间通过「名称」来唯一标识，不能重名。同一个进程中可以跑多个节点，这叫「节点组合」(Composition)。

创建一个节点其实很简单。看代码：

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()会阻塞主线程。如果你需要做其他事情，可以用多线程或者定时器。我曾经在项目里忘了加rclpy.init()，结果节点一直报错，找了半天才发现——嗯，这种低级错误谁都会犯。

3.2 话题(Topic)：发布-订阅模式

话题是ROS2里最常用的通信方式。一个节点发布消息到某个话题，其他节点订阅这个话题就能收到消息。典型的「发布-订阅」模式。

为什么叫「话题」？你可以想象成电台广播。电台（发布者）在某个频道（话题名）上播放节目，听众（订阅者）只要调到这个频道就能收听。发布者和订阅者互相不知道对方的存在，这就是解耦。

我的经验：话题名最好用层级结构，比如/vehicle/steering/status，而不是steering_status。这样在大型系统里不容易冲突，也方便调试。

来个发布者的例子：

class Talker(Node):
    def __init__(self):
        super().__init__('talker')
        self.pub = self.create_publisher(String, 'chatter', 10)
        self.timer = self.create_timer(1.0, self.timer_callback)
        self.i = 0

    def timer_callback(self):
        msg = String()
        msg.data = f'Hello, ROS2! 第{self.i}次'
        self.pub.publish(msg)
        self.get_logger().info(f'发布了: {msg.data}')
        self.i += 1

订阅者呢？更简单：

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

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

注意那个数字10，这是队列大小。什么意思？如果订阅者处理速度跟不上发布速度，消息会暂存在队列里。队列满了怎么办？旧消息会被丢弃。我曾经在高速场景下把队列设成1，结果丢包严重，车辆控制出现抖动——后来改成100就稳了。

3.3 服务(Service)：请求-应答模式

话题是单向的，发布者只管发，不管有没有人收。但有些场景需要「一问一答」，比如「请给我当前车辆的位置」——这时候就该服务上场了。

服务由两部分组成：客户端(Client)发起请求，服务端(Server)处理请求并返回响应。

注意：服务是同步的。客户端发送请求后会阻塞等待响应。如果服务端处理时间很长，客户端会一直卡着。所以耗时操作别用服务，用动作。

先定义服务接口文件AddTwoInts.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)
        rclpy.spin_until_future_complete(self, self.future)
        return self.future.result()

这里有个细节：call_async是异步调用，不会阻塞。但如果你需要等待结果，可以用spin_until_future_complete。我个人习惯用异步方式，这样节点还能处理其他事情。

3.4 动作(Action)：带反馈的长时间任务

动作是ROS2里最强大的通信方式。它结合了话题和服务的特点：

  • 像服务一样，有目标(Goal)和结果(Result)
  • 像话题一样，可以持续反馈(Feedback)进度
  • 还支持取消(Cancel)操作

你想想看，自动驾驶里的「变道」操作——从发起变道到完成变道，中间需要持续反馈当前进度（比如「已转向30%」），如果发现危险还能取消。这就是动作的典型应用场景。

动作的三要素：

  • 目标(Goal)：告诉动作服务器你要干什么
  • 反馈(Feedback)：服务器在执行过程中不断告诉你进度
  • 结果(Result)：最终执行完毕后的输出

定义动作接口MoveTo.action：

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

动作服务器实现（简化版）：

class MoveServer(Node):
    def __init__(self):
        super().__init__('move_server')
        self.action_server = ActionServer(
            self, MoveTo, 'move_to', self.execute_callback)

    def execute_callback(self, goal_handle):
        self.get_logger().info(f'收到目标: ({goal_handle.request.target_x}, {goal_handle.request.target_y})')
        
        feedback_msg = MoveTo.Feedback()
        feedback_msg.current_x = 0.0
        feedback_msg.current_y = 0.0
        
        for i in range(10):
            if goal_handle.is_cancel_requested:
                goal_handle.canceled()
                result = MoveTo.Result()
                result.success = False
                result.message = '任务被取消'
                return result
            
            feedback_msg.current_x += goal_handle.request.target_x / 10
            feedback_msg.current_y += goal_handle.request.target_y / 10
            feedback_msg.progress = (i + 1) / 10.0
            goal_handle.publish_feedback(feedback_msg)
            time.sleep(0.5)
        
        goal_handle.succeed()
        result = MoveTo.Result()
        result.success = True
        result.message = '到达目标位置'
        return result

动作客户端：

class MoveClient(Node):
    def __init__(self):
        super().__init__('move_client')
        self.action_client = ActionClient(self, MoveTo, 'move_to')

    def send_goal(self, x, y):
        goal_msg = MoveTo.Goal()
        goal_msg.target_x = x
        goal_msg.target_y = y
        
        self.action_client.wait_for_server()
        send_goal_future = self.action_client.send_goal_async(
            goal_msg, feedback_callback=self.feedback_callback)
        
        send_goal_future.add_done_callback(self.goal_response_callback)

    def feedback_callback(self, feedback_msg):
        fb = feedback_msg.feedback
        self.get_logger().info(f'进度: {fb.progress*100:.0f}%, 当前位置: ({fb.current_x:.2f}, {fb.current_y:.2f})')

    def goal_response_callback(self, future):
        goal_handle = future.result()
        if not goal_handle.accepted:
            self.get_logger().info('目标被拒绝')
            return
        self.get_logger().info('目标被接受')
        result_future = goal_handle.get_result_async()
        result_future.add_done_callback(self.result_callback)

    def result_callback(self, future):
        result = future.result().result
        self.get_logger().info(f'结果: {result.message}, 成功: {result.success}')

避坑指南：我曾经在动作回调里做了大量计算，导致反馈消息发送不及时。后来发现动作服务器默认是单线程的，耗时操作会阻塞反馈。解决方案是用AsyncActionServer或者把耗时操作放到子线程里。

3.5 四种通信方式对比

特性 话题(Topic) 服务(Service) 动作(Action)
通信模式 发布-订阅 请求-应答 目标-反馈-结果
是否同步 异步 同步(可异步调用) 异步
是否支持反馈 否 否 是
是否支持取消 否 否 是
典型场景 传感器数据、状态发布 查询、配置、触发 导航、机械臂控制、变道
QoS设置 支持 有限支持 支持

嗯，这张表你最好保存下来。选型的时候拿出来对照一下，基本不会出错。

3.6 实战建议

最后给几个实战中的小建议：

  • 能用话题就别用服务。服务是同步的，容易造成系统阻塞。我见过有人用服务传激光雷达数据，结果帧率直接掉到个位数。
  • 动作的反馈频率别太高。每10ms发一次反馈？没必要。100ms一次足够了，否则CPU都在处理反馈消息。
  • 节点命名要有规律。比如/perception/lidar_node、/planning/path_planner。这样用ros2 node list一看就清楚。
  • 调试时多用命令行工具。ros2 topic echo、ros2 service call、ros2 action send_goal，这些命令能帮你快速验证通信是否正常。

好了，这一章的内容就到这里。四种通信方式你都亲手写过代码了吗？别光看，打开终端敲一遍。代码跑通了，才算真正学会。

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