3、ROS2通信机制实战：编写发布者与订阅者、自定义消息接口、使用rqt工具进行可视化调试

好，咱们直接进入正题。ROS2最核心的东西是什么？说白了就是通信。节点之间怎么说话、怎么传数据，全靠这套机制。今天我就带你手把手把发布者、订阅者、自定义消息、还有rqt可视化调试全过一遍。

3.1 发布者与订阅者——最基础的通信模型

ROS2的通信模型，说白了就是「广播」模式。一个节点往话题里发数据，其他节点想听就订阅。我刚开始学的时候，总觉得这跟ROS1差不多嘛。但实际用起来，ROS2的DDS底层让数据传输更可靠、更实时。

先看一个最简单的发布者。假设我们要发一个「Hello ROS2」字符串，频率是1Hz。

# talker.py - 发布者节点
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 = 1.0  # 1秒发一次
        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()

嗯，这里要注意几个点。第一，create_publisher的第三个参数是队列大小，我建议设成10就够了。第二，定时器回调里别做耗时操作，不然会阻塞其他回调。

再看订阅者怎么写。说白了就是反过来，等着收数据。

# listener.py - 订阅者节点
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()

3.2 自定义消息接口——让数据更丰富

标准消息类型（String、Int32这些）够用吗？说实话，在自动驾驶场景里远远不够。你想想看，一个激光雷达点云数据，或者一个车辆状态信息，包含多个字段。这时候就需要自定义消息了。

ROS2自定义消息用.msg文件定义。我习惯在功能包下建一个msg文件夹。比如我们要定义一个「车辆状态」消息：

# msg/VehicleState.msg
float32 speed          # 车速，单位 m/s
float32 steering_angle # 方向盘转角，单位 rad
float32 acceleration   # 加速度，单位 m/s^2
uint8 gear             # 档位：0-N, 1-D, 2-R, 3-P
string vehicle_id      # 车辆ID
builtin_interfaces/Time timestamp  # 时间戳

定义好之后，要在CMakeLists.txt和package.xml里注册。这一步很多人会忘。

# CMakeLists.txt 中添加
rosidl_generate_interfaces(${PROJECT_NAME}
  "msg/VehicleState.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>

编译成功后，就可以在代码里用了。发布自定义消息的写法跟标准消息一样：

from your_package.msg import VehicleState

msg = VehicleState()
msg.speed = 10.5
msg.steering_angle = 0.3
msg.acceleration = 0.0
msg.gear = 1
msg.vehicle_id = 'AV-001'
msg.timestamp = self.get_clock().now().to_msg()
self.publisher.publish(msg)

3.3 rqt工具——可视化调试利器

代码写完了，怎么知道数据对不对？总不能每次都打印日志吧。rqt就是ROS2的「仪表盘」，能让你实时看到话题数据、节点关系、甚至绘制曲线。

启动rqt很简单：

rqt

我个人最常用的几个插件：

插件名称	功能	使用场景
Topic Monitor	查看所有话题的实时数据	调试发布者是否正常发数据
Plot	绘制数值型话题的曲线	观察车速、转角等连续变化量
Node Graph	显示节点和话题的拓扑图	检查通信链路是否完整
Message Publisher	手动发布话题消息	测试订阅者响应是否正确

举个例子。你启动发布者和订阅者后，打开rqt的Topic Monitor，选中/chatter话题，就能看到字符串数据一条条刷新。如果数据没出现，八成是话题名写错了。

3.4 完整实战：从消息定义到可视化

咱们把上面讲的串起来，做一个完整的例子。假设我们要模拟一个简单的自动驾驶场景：车辆发送自身状态，另一个节点接收并显示。

1. 创建功能包：ros2 pkg create vehicle_communication --build-type ament_python --dependencies rclpy std_msgs builtin_interfaces
2. 定义VehicleState.msg：放在msg/目录下，内容如上
3. 编写发布者：模拟车辆每0.1秒发送一次状态
4. 编写订阅者：接收并打印状态，同时用rqt Plot绘制车速曲线
5. 编译运行：colcon build后启动节点，打开rqt观察

运行效果应该是这样的：订阅者终端每秒打印10次车辆状态，rqt的Plot上显示一条平滑的车速曲线。如果车速跳变，说明数据有问题，可以回头检查发布者的逻辑。

好了，这一章的内容就到这。通信机制是ROS2的基石，搞懂了发布-订阅模型，后面学服务、动作就轻松多了。记住，多动手写代码，多打开rqt看看数据，比死记硬背强一百倍。