第三讲：话题通信实战——发布者与订阅者模式、自定义消息类型、多话题同步机制
好，咱们直接进入正题。话题通信是ROS2里最基础、也是最常用的通信方式。说白了，就是一个节点把数据“喊”出来，另一个节点“听”到并处理。今天这一讲，我会带着你从最基础的发布订阅写起，再到自定义消息类型，最后聊聊多话题同步的那些坑。
3.1 发布者与订阅者模式：从零搭建一个“喊话”系统
先看一个最简单的例子。假设我们有一个传感器节点，每隔1秒发布一次“Hello ROS2”字符串。另一个节点负责接收并打印。
发布者节点（talker）
import rclpy
from rclpy.node import Node
from std_msgs.msg import String
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.count = 0
def timer_callback(self):
msg = String()
msg.data = f'Hello ROS2: {self.count}'
self.pub.publish(msg)
self.get_logger().info(f'发布: {msg.data}')
self.count += 1
def main(args=None):
rclpy.init(args=args)
node = Talker()
rclpy.spin(node)
node.destroy_node()
rclpy.shutdown()
订阅者节点（listener）
import rclpy
from rclpy.node import Node
from std_msgs.msg import String
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}')
def main(args=None):
rclpy.init(args=args)
node = Listener()
rclpy.spin(node)
node.destroy_node()
rclpy.shutdown()
运行方式很简单：开两个终端，分别运行 ros2 run 包名 talker 和 ros2 run 包名 listener。你会看到发布者每秒喊一次，订阅者同步收到。
3.2 自定义消息类型：让数据“说人话”
内置的String、Int32这些类型，说实话太简陋了。实际项目中，我们经常需要传递复合数据。比如一个超声波传感器，要同时传距离、温度、置信度。这时候就得自定义消息。
第一步：创建.msg文件
在包目录下新建 msg/Ultrasound.msg：
float32 distance
float32 temperature
uint8 confidence
string sensor_id
builtin_interfaces/Time timestamp
第二步：修改CMakeLists.txt和package.xml
在CMakeLists.txt中添加：
rosidl_generate_interfaces(${PROJECT_NAME}
"msg/Ultrasound.msg"
)
package.xml中添加：
<build_depend>rosidl_default_generators</build_depend>
<exec_depend>rosidl_default_runtime</exec_depend>
<member_of_group>rosidl_interface_packages</member_of_group>
第三步：编译并使用
colcon build --packages-select 你的包名
然后在代码里这样用：
from 你的包名.msg import Ultrasound
msg = Ultrasound()
msg.distance = 1.25
msg.temperature = 23.5
msg.confidence = 200
msg.sensor_id = 'ultra_01'
msg.timestamp = self.get_clock().now().to_msg()
3.3 多话题同步机制：让数据“对齐”
实际项目中，一个节点往往要订阅多个话题。比如机器人导航，要同时接收激光雷达数据、里程计数据、IMU数据。但这些数据频率不同，到达时间也不同。怎么保证它们“对齐”呢？
方案一：时间戳对齐（最常用）
每个消息都带上时间戳。订阅者收到后，根据时间戳做缓存和匹配。ROS2的 message_filters 包提供了现成的工具：
from message_filters import Subscriber, TimeSynchronizer, ApproximateTimeSynchronizer
# 精确时间同步（要求时间戳完全一致）
sub_laser = Subscriber(self, LaserScan, 'scan')
sub_odom = Subscriber(self, Odometry, 'odom')
ts = TimeSynchronizer([sub_laser, sub_odom], 10)
ts.registerCallback(self.sync_callback)
# 近似时间同步（允许一定时间差）
ats = ApproximateTimeSynchronizer(
[sub_laser, sub_odom],
queue_size=10,
slop=0.1 # 允许0.1秒的误差
)
ats.registerCallback(self.sync_callback)
方案二：缓存队列 + 手动匹配
如果数据频率差异很大（比如激光10Hz，IMU 100Hz），用时间同步器反而容易丢数据。我习惯自己维护一个缓存字典：
class SyncNode(Node):
def __init__(self):
super().__init__('sync_node')
self.buffer = {'laser': None, 'odom': None}
self.create_subscription(LaserScan, 'scan', self.laser_cb, 10)
self.create_subscription(Odometry, 'odom', self.odom_cb, 10)
def laser_cb(self, msg):
self.buffer['laser'] = msg
self.try_sync()
def odom_cb(self, msg):
self.buffer['odom'] = msg
self.try_sync()
def try_sync(self):
if self.buffer['laser'] and self.buffer['odom']:
# 检查时间差是否在允许范围内
dt = abs(self.buffer['laser'].header.stamp -
self.buffer['odom'].header.stamp)
if dt < 0.05: # 50ms以内认为同步
self.process_sync(
self.buffer['laser'],
self.buffer['odom']
)
3.4 实战避坑指南
| 常见问题 | 原因 | 解决方案 |
|---|---|---|
| 订阅者收不到消息 | 话题名拼写错误或命名空间不匹配 | 用 ros2 topic list 确认话题名 |
| 消息顺序错乱 | 多个发布者竞争同一话题 | 使用唯一话题名或加时间戳排序 |
| 自定义消息编译失败 | CMakeLists.txt中漏了依赖 | 检查rosidl_generate_interfaces配置 |
| 同步回调不触发 | 时间戳误差过大或队列太小 | 增大slop参数或queue_size |
嗯，这一讲的内容就这些。说白了，发布订阅模式是ROS2的基石，自定义消息让数据更“聪明”，多话题同步则是实战中的硬骨头。我建议你动手写一个完整的例子：一个节点发布自定义的传感器数据，另一个节点订阅并做时间同步。遇到问题别怕，多看看 ros2 topic echo 的输出，数据是不会骗人的。
ros2 topic hz /话题名 可以看发布频率，用 ros2 topic echo /话题名 --field header.stamp 可以只看时间戳。
下一讲我们会聊服务通信——那种“一问一答”的模式。到时候你会发现，有些场景用话题反而别扭，服务才是正解。先消化今天的内容，有问题随时翻翻ROS2官方文档，或者直接看源码。好，下课。