4、ROS2基础与通信机制：节点、话题、服务、动作的编程模型与QoS配置

好，咱们进入ROS2的核心地带。说实话，ROS1到ROS2的升级，最让我觉得脱胎换骨的就是通信机制这块。你想想看，自动驾驶系统里，传感器数据、控制指令、状态反馈，哪个不是靠通信在跑？如果通信不稳，车就别想开稳。

我个人习惯把ROS2的通信比作一个「消息快递系统」。节点就是收发快递的人，话题是公共信箱，服务是打电话问客服，动作则是下单后跟踪物流。咱们一个一个来拆解。

4.1 节点：系统的最小单元

节点，说白了就是一个可执行程序。每个节点只干一件事，比如激光雷达节点只管发点云，规划节点只管算路径。我刚开始做项目时，总喜欢把一堆功能塞进一个节点里，结果调试起来想死的心都有。后来学乖了——一个节点只做一件事，做好它。

创建节点很简单，Python里这样写：

import rclpy
from rclpy.node import Node

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

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

嗯，这里要注意：rclpy.spin() 会阻塞主线程，让你的节点一直活着。如果你有多个节点要同时跑，就得用多线程或者多进程了。我在做多传感器融合时，就吃过这个亏——一个节点spin住了，其他节点全饿死。

4.2 话题：发布-订阅模型

话题是ROS2里最常用的通信方式。一个节点发布消息，多个节点可以同时订阅。就像电台广播，你只管发，谁爱听谁听。

我记得有一次做数据采集，激光雷达的点云数据量特别大。我一开始用默认的QoS配置，结果发现数据老是丢包。后来才意识到——话题通信的质量完全取决于QoS配置。咱们先看代码，再聊QoS。

发布者示例：

self.publisher_ = self.create_publisher(String, 'topic_name', 10)
msg = String()
msg.data = 'Hello, ROS2!'
self.publisher_.publish(msg)

订阅者示例：

self.subscription = self.create_subscription(
    String, 'topic_name', self.callback, 10)

那个数字 10 就是队列深度。什么意思？就是消息缓存区能存多少条。我踩过的坑是：如果回调函数处理太慢，队列满了，新消息就会丢。所以队列深度要根据你的处理速度来调，不是越大越好。

4.3 服务：请求-响应模型

话题是单向的，服务是双向的。你发一个请求，对方给你一个响应。就像你问「现在几点了？」，对方回答「下午三点」。适合那些需要确认的场景，比如「启动录制」、「切换模式」。

服务端示例：

from example_interfaces.srv import AddTwoInts

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

self.srv = self.create_service(AddTwoInts, 'add_two_ints', add_callback)

客户端示例：

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)

4.4 动作：带反馈的长时间任务

动作是服务的升级版。适合那些需要跑一会儿的任务，比如「导航到A点」、「执行泊车」。你可以中途取消，也可以实时收到进度反馈。

动作由三部分组成：目标、反馈、结果。我做个比喻：你叫了个外卖，目标是「送到家」，反馈是「骑手已取餐、正在路上」，结果是「已送达」。

动作服务器端核心代码：

from example_interfaces.action import Fibonacci

class FibonacciActionServer(Node):
    def __init__(self):
        super().__init__('fibonacci_action_server')
        self.action_server = ActionServer(
            self, Fibonacci, 'fibonacci',
            self.execute_callback)

动作客户端：

self.action_client = ActionClient(self, Fibonacci, 'fibonacci')
goal_msg = Fibonacci.Goal()
goal_msg.order = 10
self.action_client.send_goal_async(goal_msg, feedback_callback=self.feedback_cb)

4.5 QoS配置：通信质量的保障

终于到QoS了。这是ROS2相比ROS1最大的改进之一。QoS决定了消息怎么传输、怎么缓存、怎么丢弃。我把它总结成三个核心参数：

参数	含义	我的建议
History	保留最近N条还是全部	传感器数据用KEEP_LAST，日志用KEEP_ALL
Reliability	可靠传输还是尽力而为	控制指令用RELIABLE，点云用BEST_EFFORT
Durability	晚订阅能否收到历史数据	地图数据用TRANSIENT_LOCAL，实时数据用VOLATILE

举个例子。激光雷达点云数据量巨大，如果用RELIABLE模式，网络稍微一卡，整个系统就堵死了。所以我一般用BEST_EFFORT，丢几帧无所谓，实时性更重要。但控制指令就不一样了——丢一条指令，车可能就撞了，必须用RELIABLE。

配置QoS的代码：

from rclpy.qos import QoSProfile, ReliabilityPolicy, HistoryPolicy

qos_profile = QoSProfile(
    depth=10,
    reliability=ReliabilityPolicy.BEST_EFFORT,
    history=HistoryPolicy.KEEP_LAST
)

self.publisher = self.create_publisher(
    PointCloud2, 'lidar_points', qos_profile)

4.6 四种通信模型的选择策略

最后，我给大家一个选择指南。在自动驾驶系统里，怎么选通信方式？

• 话题：传感器数据、状态信息、高频数据流。比如IMU、GPS、图像。
• 服务：一次性请求-响应。比如「启动采集」、「切换模式」、「查询状态」。
• 动作：长时间任务，需要反馈和取消。比如「录制一段数据」、「执行一段路径」。
• QoS：根据数据重要性、实时性、可靠性要求来配。没有万能配置，只有最适合的配置。

嗯，总结一下。ROS2的通信机制其实不复杂，但细节很多。我做了这么多年自动驾驶，最大的体会就是：通信设计决定了系统的上限。节点拆得细、话题分得清、QoS配得对，你的系统才能跑得稳、跑得快。

下一章咱们聊聊数据采集的具体实现，包括怎么同步多传感器、怎么处理时间戳。到时候我会分享一些我在实车测试中踩过的坑，保证让你少走弯路。