4. ROS2通信机制（上）：话题通信——发布者与订阅者的实现与底层原理
各位同学，今天我们聊聊ROS2里最核心的东西——话题通信。说白了，就是节点之间怎么“说话”的。你想想看，自动驾驶系统里，激光雷达要告诉感知模块“我看到了什么”，规划模块要告诉控制模块“接下来怎么走”。这些消息怎么传？靠的就是话题通信。
我个人习惯把话题通信比作“电台广播”。发布者就是电台主持人，他对着麦克风说话，不管有没有人听。订阅者就是听众，调到你喜欢的频道就能收到。双方不需要认识对方，也不需要知道对方在哪。这种解耦设计，在自动驾驶这种复杂系统里太重要了。
4.1 发布者与订阅者的基本概念
先看一个最简单的例子。假设我们要写一个发布者，每隔一秒发一条“Hello, ROS2”的消息。
import rclpy
from rclpy.node import Node
from std_msgs.msg import String
class Talker(Node):
def __init__(self):
super().__init__('talker')
self.publisher = 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.publisher.publish(msg)
self.get_logger().info(f'Publishing: {msg.data}')
self.i += 1
def main(args=None):
rclpy.init(args=args)
node = Talker()
rclpy.spin(node)
node.destroy_node()
rclpy.shutdown()
这段代码看着简单，但里面有几个关键点。create_publisher的第一个参数是消息类型，第二个是话题名，第三个是队列大小。队列大小这个参数，我在项目中踩过坑，后面会细说。
订阅者的代码也很直观：
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'I heard: {msg.data}')
嗯，这里要注意，订阅者的回调函数是在收到消息时被触发的。你不需要自己写循环去轮询，ROS2的底层帮你做好了。
4.2 底层原理：DDS在背后做了什么？
很多同学学完ROS2，只知道怎么调API，但不知道底层怎么跑的。我个人觉得，不理解底层原理，出了问题你根本不知道怎么排查。
ROS2的通信底层用的是DDS（Data Distribution Service）。DDS是一个工业级的实时数据分发标准。ROS2选择了DDS作为传输层，而不是像ROS1那样自己写TCP/UDP。
为什么会这样？因为DDS提供了很多好东西：
- 自动发现：节点启动后，DDS会自动找到其他节点，不需要你手动配置IP和端口
- QoS策略：你可以控制消息的可靠性、持久性、生命周期等
- 多对多通信：一个发布者可以对应多个订阅者，反之亦然
- 实时性保证：DDS支持实时调度，这对自动驾驶很关键
我画了一张图，帮你理解整个通信链路：
你看，发布者只管往DDS里写数据，订阅者只管从DDS里读数据。DDS负责把数据从A点送到B点。至于数据走的是UDP还是共享内存，那是DDS自己决定的。
4.3 QoS策略：别让消息丢了
说到QoS，我得好好讲讲。我在做自动驾驶项目时，遇到过一个问题：激光雷达的数据偶尔会丢帧，导致感知模块出现短暂盲区。排查了半天，发现是QoS设置不对。
ROS2的QoS策略有几个关键参数：
| QoS参数 | 说明 | 典型取值 |
|---|---|---|
| reliability | 可靠性策略 | RELIABLE（可靠）/ BEST_EFFORT（尽力） |
| durability | 持久性策略 | VOLATILE（易失）/ TRANSIENT_LOCAL（持久） |
| depth | 队列深度 | 10, 100, 1000等 |
| lifespan | 消息有效期 | 如10秒，过期丢弃 |
举个例子，激光雷达的数据对实时性要求高，但对丢包有一定容忍度。这时候用BEST_EFFORT更合适，因为RELIABLE模式会重传丢失的数据包，反而增加了延迟。
但控制指令就不一样了。你想想看，如果刹车指令丢了，那后果不堪设想。所以控制指令必须用RELIABLE模式。
核心原则：传感器数据用BEST_EFFORT，控制指令用RELIABLE。这是我在项目中总结出来的经验。
4.4 队列深度：一个容易被忽视的坑
队列深度这个参数，很多人随便填个10就完事了。我曾经也这么干，直到有一次系统在高负载下出现了消息丢失。
队列深度决定了DDS在内存中缓存多少条消息。如果订阅者处理速度跟不上发布者的发送速度，消息就会在队列里堆积。队列满了之后，新消息会覆盖旧消息（取决于具体的QoS设置）。
怎么选？我建议这样：
- 低频数据（如GPS，10Hz）：队列深度设10就够了
- 中频数据（如摄像头，30Hz）：队列深度设50-100
- 高频数据（如激光雷达，100Hz+）：队列深度设200以上
小技巧：如果你不确定队列深度设多少，可以先用大一点的值，然后通过rqt_graph或ros2 topic bw命令观察实际的消息吞吐量，再调整到合适的值。
4.5 避坑指南：我踩过的那些雷
做自动驾驶中间件这几年，我在话题通信上踩过不少坑。分享几个典型的：
- 话题名拼写错误：发布者和订阅者的话题名必须完全一致，大小写敏感。我曾经因为一个字母大小写不对，排查了整整一下午。
- 消息类型不匹配：发布者和订阅者必须使用相同的消息类型。ROS2在编译时会检查，但如果你用动态加载的方式，还是可能出错。
- QoS不兼容：发布者和订阅者的QoS设置必须兼容。比如发布者用RELIABLE，订阅者用BEST_EFFORT，DDS会报错。
- 忘记初始化：rclpy.init()必须在创建节点之前调用。这个错误很基础，但新手经常犯。
警告：千万不要在生产环境中使用默认的QoS设置。默认值通常是BEST_EFFORT + VOLATILE，这对控制指令来说是致命的。一定要根据实际需求显式设置QoS。
4.6 实战：如何调试话题通信
最后，我教大家几个调试话题通信的实用命令。这些命令在ROS2里非常有用：
ros2 topic list：列出所有活跃的话题ros2 topic info /chatter：查看话题的详细信息，包括发布者和订阅者数量ros2 topic echo /chatter：实时打印话题上的消息内容ros2 topic hz /chatter：查看话题的发布频率ros2 topic bw /chatter：查看话题的带宽占用
我记得有一次，系统突然卡顿，我用ros2 topic bw一看，发现某个话题的带宽异常高，原来是某个节点在疯狂发送无用数据。找到问题后，很快就解决了。
好了，话题通信的基础就讲到这里。发布者和订阅者看似简单，但背后涉及DDS、QoS、队列管理等很多细节。理解这些底层原理，你才能写出稳定可靠的自动驾驶系统。
可可精品资料库