4、ROS2通信实战：编写发布者与订阅者、自定义消息接口、QoS配置

好，咱们今天来点真家伙。

前面几章讲了ROS2的架构、节点、生命周期，说白了都是在搭骨架。这一章，咱们要让骨架动起来——让节点之间真正说话。通信，才是中间件的灵魂。

我个人习惯把ROS2的通信分成三个层次：能用、会用、懂用。能用就是写个发布订阅跑起来；会用是能自定义消息类型；懂用则是能调QoS参数，让系统在极端场景下不崩。咱们一个一个来。

4.1 编写发布者与订阅者：最基础的“Hello World”

先来一个最经典的例子。一个节点发“Hello”，一个节点收“Hello”。

我记得我第一次写ROS2的发布者时，还按ROS1的习惯去写，结果编译报错。嗯，这里要注意：ROS2的API变化不小，尤其是初始化部分。

4.1.1 发布者节点

# 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.publisher_ = self.create_publisher(String, 'topic_hello', 10)
        timer_period = 0.5  # 0.5秒发一次
        self.timer = self.create_timer(timer_period, self.timer_callback)
        self.counter = 0

    def timer_callback(self):
        msg = String()
        msg.data = f'Hello, ROS2! 第 {self.counter} 次'
        self.publisher_.publish(msg)
        self.get_logger().info(f'发布了: {msg.data}')
        self.counter += 1

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

if __name__ == '__main__':
    main()

这段代码看着简单，但有个坑：create_publisher的第三个参数是队列大小。我在项目中遇到过，队列设太小，高频率发布时消息会丢；设太大，内存又扛不住。10是个比较稳妥的起步值。

4.1.2 订阅者节点

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

class SimpleSubscriber(Node):
    def __init__(self):
        super().__init__('simple_subscriber')
        self.subscription = self.create_subscription(
            String,
            'topic_hello',
            self.listener_callback,
            10)
        self.subscription  # 防止被垃圾回收

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

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

if __name__ == '__main__':
    main()

4.2 自定义消息接口：别再用String凑合了

实际项目中，谁会用String传结构化数据？你想想看，一个激光雷达点云、一个IMU数据，用String传？那解析起来得疯掉。

自定义消息接口，说白了就是定义你自己的数据结构。ROS2用.msg文件来定义，编译时会自动生成对应的Python/C++类。

4.2.1 创建自定义消息包

我建议单独建一个包来放所有自定义消息，方便复用。比如叫custom_interfaces。

# 在包目录下创建 msg/VehicleStatus.msg
int32 vehicle_id
float32 speed
float32 steering_angle
bool brake_active
string status_description

然后在CMakeLists.txt里加上：

rosidl_generate_interfaces(${PROJECT_NAME}
  "msg/VehicleStatus.msg"
)

编译后，你就可以这样用了：

from custom_interfaces.msg import VehicleStatus

msg = VehicleStatus()
msg.vehicle_id = 1
msg.speed = 30.5
msg.steering_angle = -0.2
msg.brake_active = False
msg.status_description = "正常行驶中"

4.3 QoS配置：让通信在极端场景下不崩

QoS（Quality of Service）是ROS2相比ROS1最大的进化之一。说白了，就是让你能控制消息的可靠性、时效性、历史数据保留策略。

为什么需要QoS？你想想看，自动驾驶场景下，控制指令丢了可能车就撞了；但激光雷达点云丢几帧，问题不大。不同数据，需要不同的保障级别。

4.3.1 QoS的核心参数

4.3.2 实战配置示例

控制指令，必须可靠送达，而且不能太旧：

from rclpy.qos import QoSProfile, ReliabilityPolicy, DurabilityPolicy, HistoryPolicy

control_qos = QoSProfile(
    reliability=ReliabilityPolicy.RELIABLE,
    durability=DurabilityPolicy.VOLATILE,
    history=HistoryPolicy.KEEP_LAST,
    depth=5,
    lifespan=0.5  # 超过0.5秒的消息直接丢弃
)

self.publisher_ = self.create_publisher(
    ControlCommand, 'control_topic', control_qos)

激光雷达点云，可以接受丢帧，但延迟要低：

lidar_qos = QoSProfile(
    reliability=ReliabilityPolicy.BEST_EFFORT,
    durability=DurabilityPolicy.VOLATILE,
    history=HistoryPolicy.KEEP_LAST,
    depth=1  # 只保留最新一帧
)

4.4 本章知识体系总览

下面这张图，是我自己梳理的ROS2通信实战的知识脉络。你可以把它当作一个检查清单：学完这章，看看这些点是不是都掌握了。

4.5 避坑指南

最后，分享几个我踩过的坑，希望能帮你省点时间。

• 消息类型不匹配：发布者和订阅者的消息类型必须完全一致。我曾经把std_msgs/String和std_msgs/Byte搞混了，编译没问题，运行时就是收不到消息。排查了半天。
• QoS不兼容：发布者和订阅者的QoS配置必须兼容。比如发布者设了RELIABLE，订阅者设了BEST_EFFORT，那订阅者可能收不到消息。ROS2会打印警告，但很多人没注意。
• 自定义消息包依赖：如果你在多个包中用同一个自定义消息，记得在package.xml里加上<exec_depend>custom_interfaces</exec_depend>。否则运行时导入会报错。

嗯，这一章的内容就到这里。通信是ROS2的核心，也是自动驾驶系统的基础。把发布订阅、自定义消息、QoS这三个点吃透，你就能搭建出稳定可靠的通信链路了。