3、话题通信实战：发布者节点创建、订阅者节点创建、自定义消息类型、Gazebo中的传感器数据发布

好，咱们直接进入正题。话题通信，说白了就是ROS2里最核心的数据搬运机制。一个节点发，一个节点收，中间通过话题名来匹配。今天我就带你手把手把这一整套流程跑通。

3.1 发布者节点创建——从零开始写一个talker

我个人习惯，写任何节点之前，先想清楚三件事：发什么类型的数据？发到哪个话题？多久发一次？ 想清楚了，代码就是填空。

下面是一个最简单的发布者节点，每隔1秒发布一个字符串消息。

import rclpy
from rclpy.node import Node
from std_msgs.msg import String

class TalkerNode(Node):
    def __init__(self):
        super().__init__('talker_node')
        self.publisher_ = self.create_publisher(String, 'chatter', 10)
        self.timer = self.create_timer(1.0, self.timer_callback)
        self.get_logger().info('发布者节点已启动')

    def timer_callback(self):
        msg = String()
        msg.data = 'Hello, Gazebo!'
        self.publisher_.publish(msg)
        self.get_logger().info(f'发布了: {msg.data}')

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

if __name__ == '__main__':
    main()

3.2 订阅者节点创建——让数据有地方去

有发的就得有收的。订阅者节点的写法几乎是对称的，你想想看，是不是只需要把publish换成subscribe就行了？

import rclpy
from rclpy.node import Node
from std_msgs.msg import String

class ListenerNode(Node):
    def __init__(self):
        super().__init__('listener_node')
        self.subscription = self.create_subscription(
            String,
            'chatter',
            self.listener_callback,
            10
        )
        self.get_logger().info('订阅者节点已启动')

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

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

嗯，这里要注意：回调函数里千万别做耗时操作。我曾经在回调里写了个文件写入，结果消息一多，整个节点卡死。回调函数应该轻量、快速，复杂逻辑请扔到单独的线程或定时器里。

3.3 自定义消息类型——摆脱标准类型的束缚

std_msgs里的String、Int32这些，应付简单场景还行。但如果你要发一个包含位置、速度、温度的综合数据包呢？这时候就得自己定义消息类型了。

步骤很简单，就三步：

1. 创建msg文件：在包目录下新建msg/文件夹，里面放一个SensorData.msg文件。
2. 写字段定义：每行一个字段，格式是类型 名称。
3. 改CMakeLists.txt和package.xml：让编译器知道你要生成这个消息的代码。

举个例子，SensorData.msg的内容：

float32 temperature
float32 humidity
float32 pressure
string sensor_id
builtin_interfaces/Time timestamp

我在项目中遇到过最坑的事：消息类型定义好了，但忘了在CMakeLists.txt里添加rosidl_generate_interfaces调用，结果编译死活过不去。你如果也遇到类似问题，先检查这一步。

3.4 Gazebo中的传感器数据发布——让仿真世界说话

Gazebo里的传感器，比如IMU、激光雷达、摄像头，它们本质上就是一个个发布者节点。你不需要自己写传感器驱动，Gazebo已经帮你封装好了。你要做的，就是订阅这些话题。

常用的传感器话题如下：

举个例子，如果你想在Gazebo里获取IMU数据，写个订阅者就行：

from sensor_msgs.msg import Imu

class ImuSubscriber(Node):
    def __init__(self):
        super().__init__('imu_subscriber')
        self.sub = self.create_subscription(
            Imu, '/imu/data', self.imu_callback, 10
        )

    def imu_callback(self, msg):
        # 获取四元数
        q = msg.orientation
        # 获取角速度
        av = msg.angular_velocity
        # 获取线加速度
        la = msg.linear_acceleration
        self.get_logger().info(f'加速度: x={la.x:.2f}, y={la.y:.2f}, z={la.z:.2f}')

3.5 本章知识体系总览

下面这张图，把话题通信的核心流程和本章涉及的关键节点都串起来了。你可以把它当作一个快速参考。

这张图里，上半部分是发布-订阅的标准流程，下半部分左边是自定义消息的创建路径，右边是Gazebo里常见的传感器话题。你可以对照着图，把代码跑一遍，感受一下数据是怎么流动的。

好了，话题通信的实战部分就到这里。代码不多，但每一步都有讲究。你动手试试，遇到问题很正常——我当年也是踩了无数坑才摸清楚的。