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

各位同学，欢迎来到第四章。从这一章开始，我们正式进入ROS2的核心——通信机制。

我个人一直觉得，ROS2最迷人的地方就是它的通信方式。它让机器人里的各个模块，比如传感器、控制器、执行器，能够像一群配合默契的队友一样协同工作。而话题通信，就是这群队友之间最常用、最基础的对讲机。

4.1 话题通信的本质：发布者与订阅者

先别急着敲代码，我们得先搞明白一个概念：什么是话题通信？

说白了，就是一个节点（Node）把数据扔到一个叫“话题”的通道里，另一个节点从这个通道里把数据取走。扔数据的叫发布者（Publisher），取数据的叫订阅者（Subscriber）。

你想想看，这就像电台广播。电台（发布者）不管有没有人在听，它只管播。听众（订阅者）只要调对频率（话题名），就能收到节目。而且，一个话题可以有多个发布者，也可以有多个订阅者，互不干扰。

核心特点：
  • 异步通信：发布者和订阅者不需要同时在线，也不需要互相等待。
  • 一对多/多对多：一个话题可以被多个节点订阅，也可以被多个节点发布。
  • 匿名性：发布者不知道谁在订阅，订阅者也不知道谁在发布。它们只关心话题名和数据类型。

我在做第一个机器人项目时，就犯过一个低级错误。当时我让激光雷达的数据发布到 /scan 话题，但导航模块订阅的是 /laser_scan。结果机器人愣在原地不动，我还以为是硬件坏了。排查了半天才发现，就是话题名没对上。嗯，这里要注意，话题名必须完全一致，包括大小写和斜杠。

4.2 自定义消息接口：为什么需要它？

ROS2内置了很多标准消息类型，比如 std_msgs/String、geometry_msgs/Twist。但实际项目中，你总会遇到一些特殊的数据结构。比如，我要发布一个“机器人状态”消息，里面包含电池电量、当前速度、温度三个字段。标准消息里没有现成的，怎么办？

这时候就需要自定义消息接口了。

自定义消息，说白了就是你自己定义的一个数据结构。你可以把任意多个基础类型组合在一起，形成一个全新的消息类型。ROS2会帮你自动生成对应的C++或Python代码，你直接拿来用就行。

我的习惯：只要项目中出现了超过3个相关数据需要一起传递，我就会考虑定义一个新的消息接口。这样代码更清晰，维护起来也方便。

4.3 实战：创建自定义消息接口

好，理论讲完了，我们动手干。我会带你一步步创建一个自定义消息，然后实现发布者和订阅者。

4.3.1 创建消息定义文件

首先，在你的功能包目录下，创建一个 msg 文件夹。然后新建一个 RobotStatus.msg 文件。

# RobotStatus.msg
# 定义机器人状态消息

float32 battery_level      # 电池电量，范围0.0~1.0
float32 current_speed      # 当前速度，单位 m/s
float32 temperature        # 核心温度，单位 摄氏度
string  robot_name         # 机器人名字
bool    is_moving          # 是否在移动

注意，每一行格式是：类型 字段名。ROS2支持的基础类型包括 int32、float64、string、bool 等，也支持数组和嵌套其他消息类型。

4.3.2 修改 CMakeLists.txt 和 package.xml

定义好 .msg 文件后，需要告诉编译系统去处理它。

修改 CMakeLists.txt：

# 找到消息生成依赖
find_package(rosidl_default_generators REQUIRED)

# 添加消息文件
set(msg_files
  "msg/RobotStatus.msg"
)

# 生成消息代码
rosidl_generate_interfaces(${PROJECT_NAME}
  ${msg_files}
  DEPENDENCIES std_msgs
)

# 确保依赖被导出
ament_export_dependencies(rosidl_default_runtime)

修改 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>

我曾经在这里卡了整整一个下午。编译总是报错找不到消息类型，后来发现是 package.xml 里忘了加 <member_of_group> 这一行。ROS2的依赖管理非常严格，少一行都不行。

4.3.3 编译并验证

回到工作空间根目录，执行编译：

colcon build --packages-select your_package_name

编译成功后，可以用 ros2 interface 命令查看消息定义：

ros2 interface show your_package_name/msg/RobotStatus

如果能看到你定义的字段，说明消息接口创建成功了。

4.4 实现发布者节点

消息接口有了，接下来我们写一个发布者节点，定时发布机器人状态。

# robot_status_publisher.py
import rclpy
from rclpy.node import Node
from your_package_name.msg import RobotStatus

class RobotStatusPublisher(Node):
    def __init__(self):
        super().__init__('robot_status_publisher')
        # 创建发布者，话题名为 'robot_status'，队列深度为 10
        self.publisher_ = self.create_publisher(RobotStatus, 'robot_status', 10)
        # 定时器，每 1 秒发布一次
        self.timer = self.create_timer(1.0, self.timer_callback)
        self.get_logger().info('机器人状态发布者已启动')

    def timer_callback(self):
        msg = RobotStatus()
        msg.battery_level = 0.85
        msg.current_speed = 0.5
        msg.temperature = 36.5
        msg.robot_name = 'TurtleBot3'
        msg.is_moving = True

        self.publisher_.publish(msg)
        self.get_logger().info(f'发布状态: 电量={msg.battery_level}, 速度={msg.current_speed}')

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

if __name__ == '__main__':
    main()
注意：队列深度（queue_size）参数很关键。如果发布频率很高但订阅者处理速度慢，队列会积压。设置太大可能占用大量内存，设置太小可能丢数据。我个人习惯在实时性要求高的场景设为 10，在数据量大的场景设为 100。

4.5 实现订阅者节点

有发布就有订阅。我们写一个订阅者节点，接收并打印机器人状态。

# robot_status_subscriber.py
import rclpy
from rclpy.node import Node
from your_package_name.msg import RobotStatus

class RobotStatusSubscriber(Node):
    def __init__(self):
        super().__init__('robot_status_subscriber')
        # 创建订阅者，订阅 'robot_status' 话题
        self.subscription = self.create_subscription(
            RobotStatus,
            'robot_status',
            self.listener_callback,
            10  # 队列深度
        )
        self.subscription  # 防止被垃圾回收
        self.get_logger().info('机器人状态订阅者已启动')

    def listener_callback(self, msg):
        self.get_logger().info(f'收到状态: 机器人={msg.robot_name}, '
                               f'电量={msg.battery_level:.2f}, '
                               f'速度={msg.current_speed:.2f}, '
                               f'温度={msg.temperature:.1f}, '
                               f'移动中={msg.is_moving}')

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

if __name__ == '__main__':
    main()

4.6 运行与测试

打开两个终端，分别运行发布者和订阅者：

# 终端1
ros2 run your_package_name robot_status_publisher

# 终端2
ros2 run your_package_name robot_status_subscriber

如果一切正常，订阅者终端会每秒打印一次机器人状态信息。

你也可以用命令行工具来验证：

# 查看话题列表
ros2 topic list

# 查看话题信息
ros2 topic info /robot_status

# 直接监听话题数据
ros2 topic echo /robot_status

4.7 本章知识体系

为了让你更直观地理解本章的知识结构，我画了一张图：

ROS2话题通信核心流程 发布者节点 Publisher Node 自定义消息 RobotStatus.msg 话题 /robot_status Topic Channel 订阅者节点 Subscriber Node 关键步骤 1. 定义 .msg 文件 → 2. 配置 CMakeLists.txt 和 package.xml 3. 编译生成代码 → 4. 创建发布者/订阅者 → 5. 运行测试 队列深度（queue_size）：控制消息缓冲区大小 实时场景用 10，大数据场景用 100

4.8 避坑指南与经验总结

最后，我把自己这些年踩过的坑总结一下，希望能帮你少走弯路：

  • 话题名拼写错误：这是最常见的错误。ROS2不会报错，只是订阅者收不到数据。建议用 ros2 topic list 确认话题名。
  • 消息类型不匹配：发布者和订阅者必须使用完全相同的消息类型。我曾经把 std_msgs/String 和 std_msgs/string 搞混过，大小写敏感。
  • 忘记 source 环境：每次编译完新消息接口后，记得 source install/setup.bash，否则找不到自定义消息。
  • 队列深度设置不当：如果订阅者处理速度慢，队列会溢出。可以用 rqt_graph 工具观察节点间的通信状况。
我的调试习惯：先用 ros2 topic echo 确认话题上有数据在流动，再用 ros2 topic hz 查看发布频率。如果数据流正常但订阅者没反应，那问题大概率在订阅者代码里。

好了，这一章的内容就到这里。话题通信是ROS2的基石，自定义消息接口则是你构建复杂系统的第一步。把这些基础打牢，后面的服务通信、动作通信学起来就会轻松很多。

上一章 返回目录 下一章