4. ROS2通信机制（一）：话题通信——发布者与订阅者的实现，自定义消息接口

各位同学，咱们今天聊聊ROS2里最核心、最常用的通信方式——话题通信。

说白了，话题通信就是节点之间“你发我收”的广播模式。一个节点发布消息，其他感兴趣的节点订阅它。我刚开始接触ROS时，觉得这跟单片机里的UART广播差不多，但实际用起来，ROS2的灵活性要强得多。

4.1 话题通信的本质

话题通信有三个关键角色：发布者（Publisher）、订阅者（Subscriber）和话题（Topic）。

• 发布者：负责把数据扔到话题上。它不关心谁在听。
• 订阅者：负责从话题上取数据。它不关心谁在发。
• 话题：数据交换的通道。名字就是标识，比如/camera/image、/cmd_vel。

我在项目中遇到过一个问题：两个节点用同一个话题名，但消息类型不一样，结果编译通过了，运行时却死活收不到数据。嗯，这里要注意——话题名+消息类型必须完全匹配，缺一不可。

4.2 发布者的实现

咱们先写一个最简单的发布者。假设我们要发布一个“Hello World”字符串，每秒发一次。

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

class HelloPublisher(Node):
    def __init__(self):
        super().__init__('hello_publisher')
        self.publisher_ = self.create_publisher(String, 'hello_topic', 10)
        self.timer = self.create_timer(1.0, self.timer_callback)
        self.get_logger().info('发布者已启动')

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

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

if __name__ == '__main__':
    main()

代码很简单，但我得提醒你几个坑：

• create_publisher的第三个参数是队列大小。我建议设成10，太小容易丢消息，太大占内存。
• create_timer的单位是秒。如果你要毫秒级定时，用浮点数，比如0.1就是100ms。
• 记得调用rclpy.spin()，否则节点跑完就退出了。

4.3 订阅者的实现

有发就有收。咱们写个订阅者，接收上面那个发布者的消息。

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

class HelloSubscriber(Node):
    def __init__(self):
        super().__init__('hello_subscriber')
        self.subscription = self.create_subscription(
            String,
            'hello_topic',
            self.listener_callback,
            10
        )
        self.subscription  # 防止被垃圾回收
        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 = HelloSubscriber()
    rclpy.spin(node)
    node.destroy_node()
    rclpy.shutdown()

if __name__ == '__main__':
    main()

你想想看，订阅者的代码结构和发布者几乎对称。唯一多了一行self.subscription，这是为了防止Python的垃圾回收把订阅器给回收了。我刚开始写ROS2时，就因为这行漏了，订阅器莫名其妙不工作，排查了半天。

4.4 自定义消息接口

标准消息类型（比如String、Int32）够用吗？说实话，大部分场景不够。比如你要发一个“机器人位置+速度+时间戳”的组合数据，就得自己定义消息接口。

自定义消息接口的步骤很简单：

1. 在功能包下创建msg文件夹
2. 写一个.msg文件，定义字段
3. 修改CMakeLists.txt和package.xml
4. 编译生成代码

举个例子，我要定义一个“机器人状态”消息：

# RobotStatus.msg
float32 x
float32 y
float32 theta
float32 linear_velocity
float32 angular_velocity
builtin_interfaces/Time timestamp

然后在CMakeLists.txt里加上：

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

编译后，你就可以在代码里这样用了：

from your_package.msg import RobotStatus

msg = RobotStatus()
msg.x = 1.0
msg.y = 2.0
msg.theta = 0.5
msg.linear_velocity = 0.2
msg.angular_velocity = 0.1
msg.timestamp = self.get_clock().now().to_msg()

4.5 发布者与订阅者的完整示例

咱们把自定义消息用起来。写一个发布者，发布机器人状态：

import rclpy
from rclpy.node import Node
from your_package.msg import RobotStatus

class StatusPublisher(Node):
    def __init__(self):
        super().__init__('status_publisher')
        self.publisher_ = self.create_publisher(RobotStatus, 'robot_status', 10)
        self.timer = self.create_timer(0.1, self.timer_callback)
        self.x = 0.0

    def timer_callback(self):
        msg = RobotStatus()
        msg.x = self.x
        msg.y = 0.0
        msg.theta = 0.0
        msg.linear_velocity = 0.5
        msg.angular_velocity = 0.0
        msg.timestamp = self.get_clock().now().to_msg()
        self.publisher_.publish(msg)
        self.x += 0.1
        self.get_logger().info(f'位置: x={msg.x:.2f}')

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

订阅者这边：

import rclpy
from rclpy.node import Node
from your_package.msg import RobotStatus

class StatusSubscriber(Node):
    def __init__(self):
        super().__init__('status_subscriber')
        self.subscription = self.create_subscription(
            RobotStatus,
            'robot_status',
            self.listener_callback,
            10
        )

    def listener_callback(self, msg):
        self.get_logger().info(
            f'收到状态: x={msg.x:.2f}, y={msg.y:.2f}, '
            f'线速度={msg.linear_velocity:.2f}'
        )

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

运行这两个节点，你会看到订阅者每0.1秒收到一次位置更新。这就是话题通信的典型用法。

4.6 避坑指南

最后，我总结几个实战中容易踩的坑：

• 话题名拼写错误：发布者和订阅者的话题名必须一模一样，大小写敏感。我曾经因为把robot_status写成了robot_status_（多了个下划线），查了半小时。
• 消息类型不匹配：发布者和订阅者必须用同一个消息类型。自定义消息的话，确保两边都编译了同一个msg文件。
• 队列大小设置不当：如果发布频率远高于订阅处理速度，队列会满，新消息会覆盖旧消息。我建议根据实际频率调整，一般10-20够用。
• 忘记调用rclpy.spin()：没有spin，节点不会处理回调。这个错误我犯过不止一次。

好了，话题通信的基础就这些。下一章咱们聊服务通信，那是另一种“请求-应答”模式，跟话题的“发布-订阅”完全不同。到时候见。