1. 环境准备与功能包创建在开始构建ROS2话题通信应用之前我们需要确保开发环境已经正确配置。我推荐使用Ubuntu 20.04和ROS2 Foxy版本这是目前最稳定的长期支持版本之一。如果你还没有安装ROS2可以参考官方文档进行安装。首先创建一个工作空间这是ROS2项目的基础结构。打开终端执行以下命令mkdir -p ~/ros2_ws/src cd ~/ros2_ws/src接下来创建功能包。我们将使用Python来开发这个示例ROS2对Python的支持非常友好。执行以下命令创建一个名为temperature_sensor的功能包ros2 pkg create temperature_sensor --build-type ament_python --dependencies rclpy std_msgs这个命令会创建一个基本的功能包结构并自动添加rclpy和std_msgs作为依赖项。rclpy是ROS2的Python客户端库而std_msgs包含了ROS2标准消息类型。创建完成后你会看到src目录下多了一个temperature_sensor文件夹。进入这个文件夹你会看到如下结构temperature_sensor/ ├── package.xml ├── setup.cfg ├── setup.py └── temperature_sensor/ └── __init__.py2. 自定义消息类型定义虽然ROS2提供了丰富的标准消息类型但在实际项目中我们经常需要自定义消息。在这个示例中我们将创建一个模拟温度传感器的消息类型。在功能包目录下创建msg文件夹mkdir -p ~/ros2_ws/src/temperature_sensor/msg然后在msg文件夹中创建Temperature.msg文件内容如下float32 temperature string unit string sensor_name uint32 timestamp这个自定义消息包含了温度值、单位、传感器名称和时间戳。接下来需要修改几个文件来确保ROS2能够识别这个自定义消息。首先修改package.xml添加以下内容build_dependrosidl_default_generators/build_depend exec_dependrosidl_default_runtime/exec_depend member_of_grouprosidl_interface_packages/member_of_group然后修改CMakeLists.txt如果你使用的是C或setup.pyPython项目。因为我们使用的是Python所以修改setup.pydata_files[ (share/ament_index/resource_index/packages, [resource/ package_name]), (share/ package_name, [package.xml]), (share/ package_name /msg, [msg/Temperature.msg]), ],3. 编写发布者节点现在我们来编写温度传感器的发布者节点。在temperature_sensor目录下创建temperature_publisher.py文件#!/usr/bin/env python3 import rclpy from rclpy.node import Node from temperature_sensor.msg import Temperature import random from datetime import datetime class TemperaturePublisher(Node): def __init__(self): super().__init__(temperature_publisher) self.publisher_ self.create_publisher(Temperature, temperature_data, 10) timer_period 1.0 # 1秒发布一次 self.timer self.create_timer(timer_period, self.timer_callback) self.sensor_name living_room_sensor def timer_callback(self): msg Temperature() msg.temperature random.uniform(18.0, 25.0) # 模拟18-25度之间的温度 msg.unit Celsius msg.sensor_name self.sensor_name msg.timestamp int(datetime.now().timestamp()) self.publisher_.publish(msg) self.get_logger().info(fPublishing: {msg.temperature}°{msg.unit}) def main(argsNone): rclpy.init(argsargs) temperature_publisher TemperaturePublisher() try: rclpy.spin(temperature_publisher) except KeyboardInterrupt: pass temperature_publisher.destroy_node() rclpy.shutdown() if __name__ __main__: main()这个发布者节点每隔1秒会发布一个随机的温度值。代码中我们创建了一个TemperaturePublisher类继承自Node在初始化时创建了一个发布者定时器回调函数负责生成并发布消息。4. 编写订阅者节点接下来我们编写订阅者节点用于接收并处理温度数据。在同一个目录下创建temperature_subscriber.py文件#!/usr/bin/env python3 import rclpy from rclpy.node import Node from temperature_sensor.msg import Temperature class TemperatureSubscriber(Node): def __init__(self): super().__init__(temperature_subscriber) self.subscription self.create_subscription( Temperature, temperature_data, self.listener_callback, 10) self.subscription # 防止未使用变量警告 self.temperature_history [] def listener_callback(self, msg): self.temperature_history.append(msg.temperature) if len(self.temperature_history) 10: self.temperature_history.pop(0) avg_temp sum(self.temperature_history) / len(self.temperature_history) self.get_logger().info( fReceived: {msg.temperature}°{msg.unit} from {msg.sensor_name}\n fAverage (last 10): {avg_temp:.2f}°{msg.unit} ) def main(argsNone): rclpy.init(argsargs) temperature_subscriber TemperatureSubscriber() try: rclpy.spin(temperature_subscriber) except KeyboardInterrupt: pass temperature_subscriber.destroy_node() rclpy.shutdown() if __name__ __main__: main()这个订阅者节点不仅会显示接收到的温度数据还会计算并显示最近10次温度的平均值。这种模式在实际应用中很常见比如监控系统温度变化趋势。5. 配置与运行在运行之前我们需要确保Python脚本可以被执行并且ROS2能够找到它们。首先给脚本添加执行权限chmod x ~/ros2_ws/src/temperature_sensor/temperature_sensor/temperature_publisher.py chmod x ~/ros2_ws/src/temperature_sensor/temperature_sensor/temperature_subscriber.py然后修改setup.py在console_scripts部分添加我们的节点entry_points{ console_scripts: [ temperature_publisher temperature_sensor.temperature_publisher:main, temperature_subscriber temperature_sensor.temperature_subscriber:main, ], },现在可以编译并运行我们的项目了。在工作空间根目录下执行cd ~/ros2_ws colcon build --packages-select temperature_sensor source install/setup.bash编译完成后可以分别在两个终端中运行发布者和订阅者# 终端1 ros2 run temperature_sensor temperature_publisher # 终端2 ros2 run temperature_sensor temperature_subscriber6. 调试与可视化ROS2提供了强大的工具来调试和可视化话题通信。我们可以使用以下命令来查看当前系统中的话题列表ros2 topic list要查看特定话题的内容可以使用ros2 topic echo /temperature_data要查看话题的详细信息包括发布者和订阅者数量ros2 topic info /temperature_data对于更直观的可视化ROS2提供了rqt_graph工具可以显示节点和话题之间的关系图rqt_graph在实际开发中我经常遇到的一个问题是消息没有正确传递。这时候可以按照以下步骤排查使用ros2 topic list确认话题是否存在使用ros2 topic info查看发布者和订阅者数量使用ros2 topic echo确认消息内容是否符合预期检查节点的日志输出是否有错误信息7. 进阶话题操作除了基本的话题通信ROS2还提供了一些高级功能。例如我们可以直接从命令行发布消息ros2 topic pub /temperature_data temperature_sensor/msg/Temperature {temperature: 22.5, unit: Celsius, sensor_name: cli_publisher, timestamp: 1234567890}另一个有用的命令是ros2 topic hz它可以测量话题的发布频率ros2 topic hz /temperature_data在实际项目中我们可能还需要考虑消息的序列化和反序列化效率。ROS2使用Protocol Buffers作为底层序列化机制对于大量数据传输非常高效。我曾经在一个机器人项目中处理图像数据通过优化消息结构将传输效率提升了30%。8. 实际应用中的注意事项在真实项目中使用ROS2话题通信时有几个关键点需要注意服务质量(QoS)设置ROS2引入了灵活的服务质量策略可以控制消息的可靠性、持久性等。例如对于关键数据可以使用可靠传输而对于高频但不关键的数据可以使用尽力而为传输。消息版本兼容性当消息类型发生变化时需要考虑向后兼容性。我建议在消息定义中添加注释说明每个字段的用途并在修改时谨慎考虑对现有系统的影响。性能考量高频或大数据量的消息传输可能会影响系统性能。在实际项目中我曾经遇到因为消息频率过高导致CPU占用率飙升的问题最终通过降低频率和优化消息大小解决了这个问题。错误处理健壮的节点应该能够处理各种异常情况比如消息丢失、格式错误等。在订阅者回调函数中添加适当的错误处理逻辑可以大大提高系统的稳定性。