4、ROS2通信机制（一）：话题通信

发布者与订阅者实现 · 自定义消息接口 · 话题通信的QoS配置

聊到 ROS2 通信，话题（Topic）是出镜率最高的机制。说白了，它就是异步、单向、一对多的数据管道。发布者只管往外扔数据，订阅者按需接收，两边互不等待。我刚开始接触 ROS 时，总觉得这跟 UDP 广播有点像——但 ROS2 加上了 QoS，让通信变得可控多了。

4.1 发布者与订阅者的基本实现

先看一个最简例子。我们创建一个发布节点，每隔 1 秒发一个字符串。我个人习惯用 Python 快速验证逻辑，C++ 用于性能敏感的场景。这里用 Python 演示：

# publisher.py
import rclpy
from rclpy.node import Node
from std_msgs.msg import String

class SimplePublisher(Node):
    def __init__(self):
        super().__init__('simple_publisher')
        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

订阅端更简单——收到消息就打印：

# subscriber.py
class SimpleSubscriber(Node):
    def __init__(self):
        super().__init__('simple_subscriber')
        self.sub = self.create_subscription(String, 'chatter', self.callback, 10)

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

💡 我的小习惯： 创建话题时，队列深度（queue_size）别随便填 10。如果订阅者处理慢，消息会积压甚至丢包。我一般在传感器数据流里用 5~10，低频状态用 1~3。

4.2 自定义消息接口

内置的 String、Int32 够用吗？显然不够。你想想看，一个激光雷达数据包含距离、强度、时间戳，总不能拆成三个话题发吧？自定义 .msg 文件就是干这个的。

在包目录下新建 msg/ 文件夹，创建 RadarData.msg：

# RadarData.msg
float32[] ranges
float32[] intensities
builtin_interfaces/Time stamp

然后在 CMakeLists.txt 或 setup.py 里注册。我记得第一次自定义消息时忘了改 package.xml，编译死活过不去……后来发现必须加上 <build_depend>rosidl_default_generators</build_depend>。

⚠️ 我曾经踩过的坑： 自定义消息的包名不能带连字符，ROS2 的接口生成器会报错。还有，消息字段类型尽量用 builtin_interfaces/Time 而不是自己拼时间戳，否则后面做 TF 同步时会很痛苦。

使用自定义消息时，发布者和订阅者都要 import 对应的接口：

from my_robot_interfaces.msg import RadarData
pub = self.create_publisher(RadarData, 'radar_data', 10)

4.3 话题通信的 QoS 配置

QoS（Quality of Service）是 ROS2 对比 ROS1 最实在的升级。它解决了“丢包怎么办”“消息延迟怎么控”这类实际问题。我把它理解成通信契约——发布者和订阅者必须达成一致，否则连不上。

QoS 策略	常用值	适用场景
History	KEEP_LAST / KEEP_ALL	KEEP_LAST 节省内存，KEEP_ALL 用于日志记录
Depth	队列深度 (如 10)	控制缓存消息数量
Reliability	RELIABLE / BEST_EFFORT	传感器用 BEST_EFFORT，控制指令用 RELIABLE
Durability	VOLATILE / TRANSIENT_LOCAL	延迟订阅者是否需要历史数据

举个例子。激光雷达数据量大，偶尔丢一帧没关系，用 BEST_EFFORT 加 KEEP_LAST 5：

from rclpy.qos import QoSProfile, ReliabilityPolicy, HistoryPolicy
qos = QoSProfile(
    depth=5,
    reliability=ReliabilityPolicy.BEST_EFFORT,
    history=HistoryPolicy.KEEP_LAST
)
self.pub = self.create_publisher(LaserScan, 'scan', qos)

            🔑 关键认知： 发布者和订阅者的 QoS 必须兼容。比如发布者用 RELIABLE，订阅者用 BEST_EFFORT，通信会降级或失败。我建议团队内统一 QoS 配置文件，避免“我发了你收不到”的灵异事件。
        

说到 QoS，我还想提一下 QoS 兼容性矩阵。ROS2 官方文档有张表，但说白了就一句话：双方 Reliability 和 Durability 要能匹配。如果你发现订阅者收不到消息，八成是 QoS 没对齐。

4.4 综合示例：带 QoS 的自定义消息通信

把上面知识点串起来。假设我们要发一个带有置信度的目标检测结果：

# DetectionResult.msg
int32 class_id
float32 confidence
geometry_msgs/Point position

发布者设置 QoS 为 RELIABLE + KEEP_LAST 3，订阅者用相同配置：

# 发布者
qos = QoSProfile(depth=3, reliability=ReliabilityPolicy.RELIABLE)
pub = self.create_publisher(DetectionResult, 'detection', qos)

# 订阅者（必须一致）
sub = self.create_subscription(DetectionResult, 'detection', callback, qos)

嗯，这里要注意：如果订阅者忘记设置 QoS，默认是 KEEP_LAST 10 + RELIABLE，跟发布者深度不匹配时也能通信，但深度取最小值。我建议显式声明，别依赖默认值。

🔧 调试小技巧： 用 ros2 topic info /your_topic --verbose 可以查看发布者和订阅者的 QoS 配置。如果显示不兼容，终端会直接报 warning。

最后聊点个人感受。话题通信是 ROS2 的基石，但很多人只停留在调 API 的层面。我建议你多想想：QoS 到底保护了什么？ 是防止订阅者被淹没？还是保证关键指令不丢失？想清楚这些，你写出来的系统才更健壮。