ROS2 ：Node 与 Topic 初探（Python）
在 ROS2 中，节点（Node） 和 话题（Topic） 是最基础、最常用的通信机制。掌握如何查询、监控、调试它们，是每一位 ROS2 开发者必备的技能。本文将简单介绍 ROS2 中 Node 和 Topic 的基本操作，包括命令行工具和 Python 代码实现，并教你如何快速定位“谁在发布、谁在订阅”。
1. 概念速览
- Node：ROS2 图中的一个计算单元，可以完成特定任务（如读取传感器、控制电机）。每个节点都有自己的命名空间。
- Topic：节点间传递消息的“总线”。节点可以发布（publish） 消息到某个话题，也可以订阅（subscribe） 该话题接收消息。话题采用异步、多对多的通信模式。
2. 常用命令行操作（查询、列表、结构、数据、频率）
以下命令基于 ROS2 Humble，在其他版本中基本一致。
2.1 节点操作
| 目的 | 命令 |
|---|---|
| 列出所有活跃节点 | ros2 node list |
| 查看节点详细信息 | ros2 node info <node_name> |
| 列出节点中定义的话题/服务 | ros2 node info 输出中会显示 |
$ ros2 node list /talker /listener $ ros2 node info /talker /talker Subscribers: /parameter_events: rcl_interfaces/msg/ParameterEvent Publishers: /chatter: std_msgs/msg/String /parameter_events: rcl_interfaces/msg/ParameterEvent Services: /talker/describe_parameters: rcl_interfaces/srv/DescribeParameters ...2.2 话题操作
| 目的 | 命令 |
|---|---|
| 列出所有活跃话题 | ros2 topic list（加 -t 可显示消息类型） |
| 查看话题详细信息（类型、发布/订阅者数量） | ros2 topic info <topic_name> |
| 实时打印话题数据 | ros2 topic echo <topic_name> |
| 查看话题发布频率（Hz） | ros2 topic hz <topic_name> |
| 查看话题带宽（B/s） | ros2 topic bw <topic_name> |
| 手动发布数据到话题 | ros2 topic pub <topic_name> <msg_type> '<data>' |
| 查找使用某消息类型的话题 | ros2 topic find <type_name> |
| 查看话题的消息结构 | ros2 interface show <msg_type> 或 ros2 topic echo --once |
# 列出话题和消息类型 $ ros2 topic list -t /chatter [std_msgs/msg/String] /parameter_events [rcl_interfaces/msg/ParameterEvent] /rosout [rcl_interfaces/msg/Log] # 查看话题详细信息 $ ros2 topic info /chatter Type: std_msgs/msg/String Publisher count: 1 Subscription count: 1 # 实时监控数据 $ ros2 topic echo /chatter data: "Hello ROS2" # 查看发布频率（稳定后才能显示） $ ros2 topic hz /chatter average rate: 10.012 Hz3. 如何知道谁发布、谁订阅？
两种方法：
- 使用
ros2 node info：对每个节点查看其 publishers 和 subscribers 列表。 - 使用
ros2 topic info：只告诉你发布者和订阅者的数量，不直接给节点名。要获得节点名，可以结合ros2 node list和ros2 node info逐个查询。
更快捷的技巧：
运行 rqt_graph 图形化工具，它会动态绘制出所有节点、话题以及发布/订阅关系。
rqt_graph4. Python 代码示例：创建一个发布者节点和一个订阅者节点
我们将编写两个简单的 Python 节点：
talker：每秒发布一次字符串消息到/chatter话题。listener：订阅/chatter话题并打印收到的消息。
4.1 创建功能包
source /opt/ros/humble/setup.bash ros2 pkg create --build-type ament_python my_py_nodes --dependencies rclpy std_msgs4.2 编写发布者节点 talker.py
在 my_py_nodes/my_py_nodes/ 目录下创建 talker.py：
#!/usr/bin/env python3 import rclpy from rclpy.node import Node from std_msgs.msg import String import time class Talker(Node): def __init__(self): super().__init__('talker') # 创建发布者，话题为 /chatter，队列大小10 self.publisher_ = self.create_publisher(String, '/chatter', 10) timer_period = 1.0 # 秒 self.timer = self.create_timer(timer_period, self.timer_callback) self.counter = 0 def timer_callback(self): msg = String() msg.data = f'Hello ROS2, count: {self.counter}' self.publisher_.publish(msg) self.get_logger().info(f'Publishing: "{msg.data}"') self.counter += 1 def main(args=None): rclpy.init(args=args) node = Talker() rclpy.spin(node) node.destroy_node() rclpy.shutdown() if __name__ == '__main__': main()4.3 编写订阅者节点 listener.py
在相同目录下创建 listener.py：
#!/usr/bin/env python3 import rclpy from rclpy.node import Node from std_msgs.msg import String class Listener(Node): def __init__(self): super().__init__('listener') # 创建订阅者，话题为 /chatter，回调函数为 callback self.subscription = self.create_subscription( String, '/chatter', self.callback, 10 ) self.subscription # 防止被垃圾回收 def callback(self, msg): self.get_logger().info(f'I heard: "{msg.data}"') def main(args=None): rclpy.init(args=args) node = Listener() rclpy.spin(node) node.destroy_node() rclpy.shutdown() if __name__ == '__main__': main()4.4 修改 setup.py
在 entry_points 中添加两个可执行程序：
entry_points={ 'console_scripts': [ 'talker = my_py_nodes.talker:main', 'listener = my_py_nodes.listener:main', ], },4.5 编译与运行
colcon build --packages-select my_py_nodes source install/setup.bash终端1（启动发布者）：
ros2 run my_py_nodes talker终端2（启动订阅者）：
ros2 run my_py_nodes listener现在应该能看到订阅者终端打印出发布者发送的消息。
5. 调试与监控实例
在运行上述两个节点后，可以实践之前的命令：
# 查看所有节点 ros2 node list # 输出： /talker /listener # 查看话题列表 ros2 topic list -t # 查看 /chatter 的发布/订阅关系 ros2 node info /talker # 看到 Publisher: /chatter ros2 node info /listener # 看到 Subscriber: /chatter # 实时打印话题数据 ros2 topic echo /chatter # 查看话题发布频率（发布者以 1Hz 发送） ros2 topic hz /chatter # 输出：average rate: 1.000 Hz6. 进阶操作：ros2 topic pub 手动发布
即使没有发布者节点，也可以手动向话题发送消息。例如：
ros2 topic pub /chatter std_msgs/msg/String "{data: 'Manual message'}" --once如果希望以固定频率重复发送，加 --rate 2 参数（每秒2次）。
7. 常见问题与技巧
- 话题消息类型不匹配：使用
ros2 topic info查看期望类型，再与你的消息对比。 - 看不到数据：检查是否所有节点使用了相同的 ROS_DOMAIN_ID（默认0）。
- 频率不稳定：检查回调函数是否执行耗时操作，或者使用
create_wall_timer控制发布频率。 - 查看消息结构：
ros2 interface show std_msgs/msg/String。
8. 总结
通过本文，你应该已经掌握了：
- 使用
ros2 node和ros2 topic命令行工具查询节点、话题的列表、类型、发布/订阅关系、数据内容和频率。 - 如何快速找到“谁发布了哪个话题，谁订阅了它”。
- 用 Python 编写发布者和订阅者节点，并运行、调试。
- 手动发布测试消息的方法。
这些技能是构建和调试复杂 ROS2 机器人系统的基础。在实际项目中，你还可以结合 rqt_graph、ros2 bag 录制话题数据等工具进一步提升效率。现在，去构建你自己的 ROS2 应用吧！
老徐，2026/04/04
