2、ROS2基础通信机制（一）：深入理解话题（Topic）通信——发布者与订阅者的创建、自定义消息接口、QoS配置实战
各位同学，欢迎来到ROS2实战的第二讲。
今天咱们聊的话题，是ROS2里最核心、最常用的通信方式——话题（Topic）。说白了，它就是节点之间传递数据的“管道”。一个节点发，一个节点收，一对多、多对一都行。我刚开始接触ROS时，觉得这玩意儿不就是个消息队列嘛，有啥好学的？后来在项目里踩了坑才发现，里面的门道真不少。
2.1 发布者与订阅者的创建：从“Hello World”开始
先来点实在的。咱们写一个最简单的发布者和订阅者。我习惯用Python，因为调试快，但C++性能更好。今天先用Python演示，原理是通用的。
发布者节点（talker）
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)
timer_period = 0.5 # 0.5秒发一次
self.timer = self.create_timer(timer_period, self.timer_callback)
self.i = 0
def timer_callback(self):
msg = String()
msg.data = 'Hello, ROS2! %d' % self.i
self.publisher.publish(msg)
self.get_logger().info('发布: "%s"' % msg.data)
self.i += 1
def main(args=None):
rclpy.init(args=args)
node = Talker()
rclpy.spin(node)
node.destroy_node()
rclpy.shutdown()
if __name__ == '__main__':
main()
订阅者节点（listener）
import rclpy
from rclpy.node import Node
from std_msgs.msg import String
class Listener(Node):
def __init__(self):
super().__init__('listener')
self.subscription = self.create_subscription(
String, 'chatter', self.listener_callback, 10)
self.subscription # 防止被垃圾回收
def listener_callback(self, msg):
self.get_logger().info('收到: "%s"' % msg.data)
def main(args=None):
rclpy.init(args=args)
node = Listener()
rclpy.spin(node)
node.destroy_node()
rclpy.shutdown()
if __name__ == '__main__':
main()
嗯，这里要注意：create_publisher和create_subscription的最后一个参数是QoS深度。我刚开始写的时候，随手填了个10，觉得够用了。后来在实车上，数据量一大，直接丢包丢到怀疑人生。所以，QoS不是随便填的，咱们后面细说。
rclcpp依赖，卡了半小时。嗯，都是泪。
2.2 自定义消息接口：别再用String凑合了
实际项目中，你不可能只传个字符串。比如自动驾驶里，你要传车辆状态、GPS坐标、障碍物列表。这时候就需要自定义消息接口。
ROS2的消息接口定义在.msg文件里。我习惯在包目录下建一个msg文件夹，比如VehicleStatus.msg：
# VehicleStatus.msg
float32 speed # 车速，单位 m/s
float32 steering_angle # 方向盘转角，单位 rad
bool brake_active # 刹车是否激活
uint8 gear # 档位：0-P, 1-R, 2-N, 3-D
然后在CMakeLists.txt里添加：
rosidl_generate_interfaces(${PROJECT_NAME}
"msg/VehicleStatus.msg"
)
编译后，你就可以在代码里直接用了：
from your_package.msg import VehicleStatus
msg = VehicleStatus()
msg.speed = 10.5
msg.steering_angle = 0.3
msg.brake_active = False
msg.gear = 3
float，结果编译报错。ROS2里浮点数必须用float32或float64，没有float。另外，字段名不要用关键字，比如type、class，否则你会哭的。
2.3 QoS配置实战：为什么你的数据丢了？
QoS（Quality of Service）是ROS2里最容易被忽视，但也是最关键的部分。你想想看，自动驾驶里，控制指令丢了，车可能就撞了。所以，QoS必须搞明白。
ROS2的QoS策略主要有几个维度：
| 策略 | 说明 | 常用场景 |
|---|---|---|
| History | 保留多少历史数据 | KEEP_LAST（保留最近N条）、KEEP_ALL（全保留） |
| Depth | 队列深度（History为KEEP_LAST时有效） | 控制指令用1，传感器数据用10 |
| Reliability | 可靠性策略 | RELIABLE（保证送达）、BEST_EFFORT（尽力而为） |
| Durability | 持久性 | VOLATILE（不持久）、TRANSIENT_LOCAL（持久化最后一条） |
举个例子。在自动驾驶的控制指令场景下，我建议用：
from rclpy.qos import QoSProfile, ReliabilityPolicy, HistoryPolicy
qos = QoSProfile(
depth=1,
reliability=ReliabilityPolicy.RELIABLE,
history=HistoryPolicy.KEEP_LAST
)
self.publisher = self.create_publisher(Twist, 'cmd_vel', qos)
为什么深度设成1？因为控制指令是实时性的，旧的指令没意义。你想想看，如果队列里积压了10条指令，车还在执行第1条，第10条早就过时了。所以，深度1就够了。
而对于激光雷达数据，我建议用：
qos = QoSProfile(
depth=10,
reliability=ReliabilityPolicy.BEST_EFFORT,
history=HistoryPolicy.KEEP_LAST
)
激光雷达数据量大，丢几帧没关系，但延迟不能高。所以用BEST_EFFORT，丢了就丢了，别等重传。
2.4 知识体系总览
为了让你更直观地理解本章的知识结构，我画了一张图：
这张图把本章的核心逻辑串起来了。发布者通过话题发布数据，订阅者接收并处理。中间的关键是QoS配置和消息接口定义。你把这个图记在脑子里，后面实战就不会乱。
2.5 实战中的避坑指南
- 消息类型不匹配： 我曾经在调试时，发布者发的是
Twist，订阅者却用Pose接收，结果一直收不到数据。检查了半小时才发现类型对不上。所以，发布者和订阅者的消息类型必须完全一致。 - QoS不匹配： 发布者用RELIABLE，订阅者用BEST_EFFORT，会导致连接失败。ROS2会尝试协商，但最好保持一致。
- 节点命名冲突： 同一个话题下，节点名不能重复。我习惯在节点名里加个前缀，比如
/vehicle/control_node，避免冲突。
ros2 topic list查看所有话题，用ros2 topic echo /chatter实时查看数据。这两个命令我每天用几十次，比看代码快多了。
好了，今天的内容就到这里。话题通信是ROS2的基石，你把它搞透了，后面的服务、动作通信就轻松多了。记住，多动手写代码，别光看。我当年就是看了三天理论，一写代码就报错。嗯，实践出真知。