本文是 ROS2 系列教程的第 5 篇本文是 ROS2 系列教程的第 5 篇话题 Topic 通信上——发布者与订阅者。话题是 ROS2 最核心、最常用的通信机制机器人里的传感器数据、控制指令、状态信息几乎都通过话题流转。本篇文章深入话题的 Publish/Subscribe 模型与去中心化发现机制用 Python 与 C 各实现一套发布者/订阅者掌握create_publisher、create_subscription与回调机制最后精通ros2 topic全套调试命令。一、话题通信的核心思想1.1 Publish / Subscribe 模型话题Topic采用**发布/订阅Publish-Subscribe**模式核心特点发布者Publisher只负责往话题上发消息不关心谁在收。订阅者Subscriber只负责订阅话题收消息不关心谁在发。解耦双方互相不知道对方的存在通过话题名碰头。这种模式的三大优点优点说明空间解耦发布者和订阅者不需要互相引用甚至不在同一台机器时间解耦发布者发完消息即可离开订阅者后加入也能收到后续消息配合 QoS一对多/多对多一个话题可被多个发布者发、多个订阅者收话题通信是单向的数据流从发布者到订阅者如果需要请求-应答式的双向通信用服务第 7 篇。需要长任务带反馈用动作第 10 篇。1.2 话题名与消息类型话题由话题名 消息类型双重标识话题名topic name/cmd_vel 消息类型message typegeometry_msgs/msg/Twist话题名决定谁和谁通信——发布者和订阅者的话题名必须完全一致。消息类型决定传什么数据结构——类型不一致时 ROS2 会拒绝匹配严格类型安全。命名规则话题名用小写字母、数字、下划线以/开头绝对话题名如/cmd_vel、/scan、/odom。带命名空间时如/robot1/cmd_vel。查看一个话题的信息ros2 topic info /cmd_vel# Type: geometry_msgs/msg/Twist# Publisher count: 1# Subscriber count: 01.3 从 ROS1 到 ROS2去中心化ROS1 的话题通信依赖roscoreMaster做话题匹配发布者向 Master 注册我要发 /cmd_vel订阅者向 Master 注册我要收 /cmd_velMaster 撮合后双方建立 TCP 连接。Master 挂了整个系统瘫痪。ROS2 改用DDS 自动发现Discovery节点启动 → 通过 DDS 组播协议广播自己的存在含发布/订阅信息 → 其他节点收到后回应 → 双方协商 QoS → 建立 P2P 直连 → 开始传输整个流程无需中心节点新节点随时加入、随时退出系统天然分布式。这也是 ROS2 支持多机、热插拔、大规模组网的根基。二、Python 发布者与订阅者2.1 发布者Publisher# py_pkg/py_publisher.py —— Python 发布者importrclpyfromrclpy.nodeimportNodefromstd_msgs.msgimportStringclassPyPublisher(Node):def__init__(self):super().__init__(py_publisher)# 1. 创建发布者话题名 消息类型 QoS 队列长度self.pubself.create_publisher(String,chatter,10)# 2. 定时器驱动周期性发布self.timerself.create_timer(1.0,self.timer_callback)self.count0deftimer_callback(self):self.count1msgString()msg.datafHello from py_publisher #{self.count}# 3. 发布消息self.pub.publish(msg)self.get_logger().info(f发布:{msg.data})defmain():rclpy.init()nodePyPublisher()rclpy.spin(node)node.destroy_node()rclpy.shutdown()if__name____main__:main()三个关键步骤create_publisher(类型, 话题名, 队列长度)—— 第三个参数是QoS 深度消息队列最多缓存几条第 9 篇详讲。定时器定期触发回调模拟传感器周期性发数据。publish(msg)真正把消息发出去。2.2 订阅者Subscriber# py_pkg/py_subscriber.py —— Python 订阅者importrclpyfromrclpy.nodeimportNodefromstd_msgs.msgimportStringclassPySubscriber(Node):def__init__(self):super().__init__(py_subscriber)# 1. 创建订阅者话题名 消息类型 回调函数 QoSself.subself.create_subscription(String,chatter,self.listener_callback,10)deflistener_callback(self,msg):# 2. 收到消息时自动调用spin 的循环里执行self.get_logger().info(f收到:{msg.data})defmain():rclpy.init()nodePySubscriber()rclpy.spin(node)node.destroy_node()rclpy.shutdown()if__name____main__:main()订阅者回调是话题通信的核心机制spin()循环不断检查是否有新消息一旦到达就调用listener_callback。回调在单线程里串行执行所以回调里不要做耗时操作回顾第 2 篇。2.3 运行一对发布/订阅# 终端 1运行发布者ros2 run py_pkg py_publisher# 终端 2运行订阅者ros2 run py_pkg py_subscriber订阅者终端会持续打印收到: Hello from py_publisher #N。重要现象先启动订阅者、后启动发布者或反过来都能正常通信——这就是 DDS 自动发现的威力新节点加入的瞬间双方自动建立连接。三、C 发布者与订阅者3.1 C 发布者// cpp_pkg/src/publisher.cpp#includerclcpp/rclcpp.hpp#includestd_msgs/msg/string.hpp#includechrono#includememoryusingnamespacestd::chrono_literals;classCppPublisher:publicrclcpp::Node{public:CppPublisher():Node(cpp_publisher),count_(0){// 创建发布者类型 话题名 QoSpublisher_this-create_publisherstd_msgs::msg::String(chatter,10);timer_this-create_wall_timer(1s,std::bind(CppPublisher::timer_callback,this));}private:voidtimer_callback(){automsgstd_msgs::msg::String();msg.dataHello from cpp_publisher #std::to_string(count_);publisher_-publish(msg);// 发布RCLCPP_INFO(this-get_logger(),发布: %s,msg.data.c_str());}rclcpp::Publisherstd_msgs::msg::String::SharedPtr publisher_;rclcpp::TimerBase::SharedPtr timer_;intcount_;};intmain(intargc,char**argv){rclcpp::init(argc,argv);rclcpp::spin(std::make_sharedCppPublisher());rclcpp::shutdown();return0;}3.2 C 订阅者// cpp_pkg/src/subscriber.cpp#includerclcpp/rclcpp.hpp#includestd_msgs/msg/string.hppclassCppSubscriber:publicrclcpp::Node{public:CppSubscriber():Node(cpp_subscriber){// 创建订阅者类型 话题名 回调 QoSsubscription_this-create_subscriptionstd_msgs::msg::String(chatter,10,std::bind(CppSubscriber::topic_callback,this,std::placeholders::_1));}private:voidtopic_callback(conststd_msgs::msg::String::SharedPtr msg){RCLCPP_INFO(this-get_logger(),收到: %s,msg-data.c_str());}rclcpp::Subscriptionstd_msgs::msg::String::SharedPtr subscription_;};intmain(intargc,char**argv){rclcpp::init(argc,argv);rclcpp::spin(std::make_sharedCppSubscriber());rclcpp::shutdown();return0;}C 回调用std::bind绑定成员函数消息以SharedPtr共享指针传入避免拷贝。3.3 双语言混搭通信ROS2 最重要的特性之一消息类型一致即可跨语言通信。# C 发布者 Python 订阅者或反过来都能正常通信ros2 run cpp_pkg publisher# C 发ros2 run py_pkg py_subscriber# Python 收因为话题数据在 DDS 层以二进制序列化CDR 格式传输语言只是外衣。这个特性让团队可以自由选择语言性能敏感模块用 C快速迭代模块用 Python。四、常见标准消息类型ROS2 预置了大量标准消息std_msgs、geometry_msgs、sensor_msgs、nav_msgs等。先掌握最常用的几个std_msgs/msg/Header # 时间戳坐标系id几乎所有消息都嵌套它 std_msgs/msg/String # 字符串示例常用 std_msgs/msg/Int32 # 32位整数 std_msgs/msg/Float64 # 64位浮点 geometry_msgs/msg/Twist # 线速度角速度/cmd_vel 常用 geometry_msgs/msg/PoseStamped # 带时间的位姿导航目标点 sensor_msgs/msg/LaserScan # 激光雷达数据 sensor_msgs/msg/Image # 图像 nav_msgs/msg/Odometry # 里程计查看消息字段ros2 interface show geometry_msgs/msg/Twist# Vector3 linear (x, y, z 线速度)# Vector3 angular (x, y, z 角速度)五、ros2 topic 全套调试命令ros2 topic是话题排障的瑞士军刀务必全部掌握# 1. 列出所有话题ros2 topic list ros2 topic list-t# 带类型显示# 2. 查看话题信息类型、发布/订阅者数量ros2 topic info /chatter# 3. 实时回显话题内容最常用ros2 topicecho/chatter# 4. 查看话题发布频率ros2 topic hz /chatter# 5. 查看话题带宽ros2 topic bw /chatter# 6. 手动发布消息调试神器不用写代码ros2 topic pub /chatter std_msgs/msg/String{data: hello}--rate1# --rate 1 每秒发 1 条# --once 只发 1 条# --times N 发 N 条# 7. 查看话题的 QoS 设置ros2 topic info /chatter--verbose实战演练不开任何节点用ros2 topic pub发消息 另一个终端ros2 topic echo收消息验证话题链路再ros2 topic hz看频率是否稳定在 1Hz。这套组合能快速判断是发布端的问题还是订阅端的问题。六、实战双话题传感器仿真6.1 场景设计模拟一个机器人sensor_node同时发布里程计/odom整数计数和速度指令/cmd_velTwist 消息display_node订阅两者并打印。6.2 传感器节点双发布者# py_pkg/sensor_node.py —— 一个节点两个发布者importrclpyfromrclpy.nodeimportNodefromstd_msgs.msgimportInt32fromgeometry_msgs.msgimportTwistclassSensorNode(Node):def__init__(self):super().__init__(sensor_node)# 两个发布者不同话题、不同类型self.odom_pubself.create_publisher(Int32,odom,10)self.cmd_pubself.create_publisher(Twist,cmd_vel,10)self.timerself.create_timer(0.5,self.tick)self.count0deftick(self):self.count1# 发里程计odom_msgInt32()odom_msg.dataself.count self.odom_pub.publish(odom_msg)# 发速度指令每 5 拍换一次方向cmdTwist()cmd.linear.x0.5if(self.count%105)else-0.5cmd.angular.z0.2self.cmd_pub.publish(cmd)self.get_logger().info(fodom{self.count}cmd_vx{cmd.linear.x:.2f})defmain():rclpy.init()nodeSensorNode()rclpy.spin(node)node.destroy_node()rclpy.shutdown()if__name____main__:main()6.3 显示节点双订阅者# py_pkg/display_node.py —— 一个节点两个订阅者importrclpyfromrclpy.nodeimportNodefromstd_msgs.msgimportInt32fromgeometry_msgs.msgimportTwistclassDisplayNode(Node):def__init__(self):super().__init__(display_node)self.odom_subself.create_subscription(Int32,odom,self.on_odom,10)self.cmd_subself.create_subscription(Twist,cmd_vel,self.on_cmd,10)defon_odom(self,msg):self.get_logger().info(f里程计:{msg.data})defon_cmd(self,msg):self.get_logger().info(f速度指令: vx{msg.linear.x:.2f}wz{msg.angular.z:.2f})defmain():rclpy.init()nodeDisplayNode()rclpy.spin(node)node.destroy_node()rclpy.shutdown()if__name____main__:main()6.4 运行与验证# 终端 1传感器节点ros2 run py_pkg sensor_node# 终端 2显示节点ros2 run py_pkg display_node# 终端 3可视化整个话题图ros2 topic list rqt_graph# 图形化显示节点-话题连接关系rqt_graph会画出sensor_node和display_node两个椭圆中间两条带箭头的线分别指向/odom和/cmd_vel话题节点——这就是ROS 图的直观呈现。6.5 一个节点多个话题的意义本例演示了 ROS2 的重要设计一个节点可以拥有任意多个发布者/订阅者。真实机器人里sensor_node可能同时发布/scan、/odom、/imu多个话题订阅端同理。话题与节点是多对多关系而非一对一。七、常见问题排查现象原因解决订阅者收不到消息话题名或类型不匹配ros2 topic info对比双方话题名与类型收不到但topic info正常QoS 不兼容双方 QoS 策略要能匹配第 9 篇ros2 topic hz无输出发布者没在发检查定时器是否在跑、publish是否被调用回调不执行spin 没跑或回调阻塞确认有spin回调内勿做耗时操作C 编译报类型错误include 路径或类型名错#include 包名/msg/类型.hpp类型用小写文件名消息收发有延迟/抖动QoS 深度太小增大队列深度或调整 QoS第 9 篇八、总结本篇文章完成了话题通信的入门理解了 Publish/Subscribe 模型的三大解耦优势、DDS 去中心化发现机制用 Python 与 C 分别实现了发布者/订阅者并验证了跨语言通信掌握了ros2 topic全套调试命令最后通过传感器-显示双话题实战巩固了多发布者/多订阅者的能力。关键要点回顾话题 话题名 消息类型双重标识类型必须匹配。发布者create_publisher(类型, 话题名, QoS深度)订阅者create_subscription(类型, 话题名, 回调, QoS深度)。订阅者回调由spin()驱动回调内勿做耗时操作。消息类型一致即可跨语言C ↔ Python通信。ros2 topic echo/hz/pub/info是排障四件套。一个节点可有多个发布者/订阅者话题与节点多对多。下一篇预告下一篇进阶话题通信下——自定义消息与周期发布/订阅的自定义消息类型引用第 4 篇定义的接口包、固定周期与动态周期发布、高频率话题的吞吐与延迟、ros2 topic的高级用法--qos-reliability、--no-arr等以及话题数据的录制与回放ros2 bag。学完你将能构建完整的数据流系统。