本文是 ROS2 系列教程的第 6 篇本文是 ROS2 系列教程的第 6 篇话题 Topic 通信下——自定义消息与周期。上一篇掌握了话题的基本收发本篇文章解决两个工程化问题自定义消息类型的实战使用引用第 4 篇的接口包与发布周期的精细控制固定周期、动态周期、高频率话题。同时介绍ros2 topic高级用法与 ros2 bag 数据录制回放。学完你就能构建完整的数据流系统。一、自定义消息实战1.1 完整链路回顾自定义消息从定义到使用的完整链路定义 .msg/.srv 文件接口包 → colcon build 生成类型代码 → 发布者/订阅者 include/import 使用 → ros2 run 运行验证第 4 篇我们定义了my_interfaces/msg/RobotStatus.msg含 robot_name、battery_level、status_code、stamp。本篇文章围绕它展开实战。1.2 Python 使用自定义消息# py_pkg/robot_publisher.py —— 发布自定义消息importrclpyfromrclpy.nodeimportNodefrommy_interfaces.msgimportRobotStatusfrombuiltin_interfaces.msgimportTimeclassRobotPublisher(Node):def__init__(self):super().__init__(robot_publisher)# 发布自定义消息类型self.pubself.create_publisher(RobotStatus,robot_status,10)self.timerself.create_timer(1.0,self.publish_status)self.battery1.0defpublish_status(self):msgRobotStatus()msg.robot_nameturtle_01self.battery-0.01msg.battery_levelself.battery msg.status_code1# 时间戳取当前系统时间秒纳秒msg.stampTime()nowself.get_clock().now()msg.stamp.secnow.seconds_nanoseconds()[0]msg.stamp.nanosecnow.seconds_nanoseconds()[1]self.pub.publish(msg)self.get_logger().info(f发布:{msg.robot_name}电量{msg.battery_level:.2f})defmain():rclpy.init()nodeRobotPublisher()rclpy.spin(node)node.destroy_node()rclpy.shutdown()if__name____main__:main()Python 嵌套消息赋值嵌套的stamp需要单独import Time并创建实例也可以用from builtin_interfaces.msg import Time后直接msg.stamp Time()。1.3 C 使用自定义消息// cpp_pkg/src/robot_subscriber.cpp —— 订阅自定义消息#includerclcpp/rclcpp.hpp#includemy_interfaces/msg/robot_status.hppclassRobotSubscriber:publicrclcpp::Node{public:RobotSubscriber():Node(robot_subscriber){sub_this-create_subscriptionmy_interfaces::msg::RobotStatus(robot_status,10,std::bind(RobotSubscriber::on_status,this,std::placeholders::_1));}private:voidon_status(constmy_interfaces::msg::RobotStatus::SharedPtr msg){// 访问嵌套的 stamp 字段autosecmsg-stamp.sec;RCLCPP_INFO(this-get_logger(),收到: %s 电量%.2f 状态%d sec%ld,msg-robot_name.c_str(),msg-battery_level,msg-status_code,sec);}rclcpp::Subscriptionmy_interfaces::msg::RobotStatus::SharedPtr sub_;};intmain(intargc,char**argv){rclcpp::init(argc,argv);rclcpp::spin(std::make_sharedRobotSubscriber());rclcpp::shutdown();return0;}1.4 依赖接口包的声明关键使用自定义消息的包必须在package.xml声明依赖接口包dependmy_interfaces/dependC 包还要在CMakeLists.txt里find_packagefind_package(my_interfaces REQUIRED) ament_target_dependencies(robot_subscriber my_interfaces)构建顺序先构建接口包再构建使用它的包colcon build --packages-select my_interfaces colcon build --packages-select py_pkg cpp_pkg1.5 常用内置接口快速上手除了自定义接口掌握几个高频内置类型sensor_msgs/msg/LaserScan # 激光雷达ranges 数组 angle_min/max sensor_msgs/msg/Image # 图像宽高 像素编码 data geometry_msgs/msg/Twist # 速度指令linear angular geometry_msgs/msg/PoseStamped # 带时间戳位姿导航目标 nav_msgs/msg/Odometry # 里程计位姿 速度 协方差 std_msgs/msg/Header # 通用头时间戳 frame_idHeader是几乎所有消息的公共字段包含stamp时间戳和frame_id坐标系用于数据的时间同步与坐标关联# 给消息填 Header 的标准姿势msg.header.stampself.get_clock().now().to_msg()msg.header.frame_idmap二、发布周期控制2.1 固定周期发布固定周期是传感器、控制循环的标配。两种实现方式方式一定时器推荐# 10Hz 发布0.1 秒周期self.timerself.create_timer(0.1,self.publish_cb)方式二定时器周期 消息计数控制发布频率# 2Hz 发布定时器 10Hz 跑每 5 次发布一次classThrottledNode(Node):def__init__(self):super().__init__(throttled_node)self.pubself.create_publisher(String,throttled,10)self.timerself.create_timer(0.1,self.tick)# 10Hzself.counter0deftick(self):self.counter1ifself.counter%50:# 每 5 拍 → 2HzmsgString()msg.dataevery 0.5sself.pub.publish(msg)2.2 动态周期发布有些场景需要动态调整频率如依据处理负载、急停时提高频率。用timer.change_interval或重建定时器# dynamic_rate.py —— 动态调整发布周期importrclpyfromrclpy.nodeimportNodefromstd_msgs.msgimportFloat64classDynamicRateNode(Node):def__init__(self):super().__init__(dynamic_rate_node)self.pubself.create_publisher(Float64,speed,10)self.rate1.0# 当前周期秒self.timerself.create_timer(self.rate,self.tick)self.count0deftick(self):self.count1msgFloat64()msg.datafloat(self.count)self.pub.publish(msg)# 每 5 拍把周期在 0.5~2.0 秒之间循环ifself.count%50:self.rate2.0ifself.rate1.5else0.5self.timer.cancel()self.timerself.create_timer(self.rate,self.tick)self.get_logger().info(f周期调整为{self.rate}s)defmain():rclpy.init()nodeDynamicRateNode()rclpy.spin(node)node.destroy_node()rclpy.shutdown()if__name____main__:main()注意create_timer创建的定时器对象在重建前要cancel()避免旧定时器继续触发。2.3 用 ros2 topic hz 验证实际频率# 终端 1运行动态周期节点ros2 run py_pkg dynamic_rate_node# 终端 2观察实际发布频率应随周期调整而变化ros2 topic hz /speed# average rate: 2.000 → 变化为 0.500 → 1.000 ...实际频率 ≠ 设定频率受系统负载、回调阻塞影响实际频率会波动。ros2 topic hz显示的是实测值是判断发布是否正常的金标准。三、高频率话题实战3.1 100Hz 里程计仿真控制/导航场景常需要高频数据。写一个 100Hz 的里程计发布器# py_pkg/high_freq_odom.py —— 100Hz 里程计仿真importrclpyfromrclpy.nodeimportNodefromnav_msgs.msgimportOdometryfromgeometry_msgs.msgimportPoint,QuaternionclassHighFreqOdom(Node):def__init__(self):super().__init__(high_freq_odom)# 100Hz0.01 秒周期self.pubself.create_publisher(Odometry,odom,50)# 队列深一点self.timerself.create_timer(0.01,self.publish_odom)self.x0.0self.t0defpublish_odom(self):self.t1self.x0.001# 每拍前进 1mmmsgOdometry()msg.header.stampself.get_clock().now().to_msg()msg.header.frame_idodommsg.child_frame_idbase_linkmsg.pose.pose.positionPoint(xself.x,y0.0,z0.0)msg.pose.pose.orientationQuaternion(x0.,y0.,z0.,w1.)# 简单协方差第 1 个元素为位置方差msg.pose.covariance[0]0.0001self.pub.publish(msg)defmain():rclpy.init()nodeHighFreqOdom()rclpy.spin(node)node.destroy_node()rclpy.shutdown()if__name____main__:main()运行验证ros2 run py_pkg high_freq_odom ros2 topic hz /odom# 应显示 ~100.000 Hzros2 topic bw /odom# 查看带宽占用高频话题的关键配置QoS 队列深度要加大如 50否则发布快于消费时消息被丢弃如果消费者处理慢还要考虑 QoS 的history策略第 9 篇详讲。3.2 高频下的性能注意点频率注意事项≤ 10Hz常规无需特殊处理10-100Hz回调内避免耗时操作队列深度 ≥ 10100-1000Hz建议 C避免消息内大数组拷贝考虑zero-copyQoS 类型适配 1000Hz考虑共享内存传输Iceoryx或降低采样经验Python 在 100Hz 内完全可用更高频率或大数据量图像建议 C且回调里只做必要处理别在回调里写日志高频日志本身会拖慢系统。四、ros2 topic 高级用法4.1 查看 QoS 兼容性# 详细查看话题的 QoS 设置ros2 topic info /odom--verbose# 输出Publisher QoS: Reliability: RELIABLE, History: KEEP_LAST, Depth: 50 ...4.2 手动发布复杂类型# 发布 Twist 类型注意嵌套结构写法ros2 topic pub /cmd_vel geometry_msgs/msg/Twist\{linear: {x: 0.5, y: 0.0, z: 0.0}, angular: {x: 0.0, y: 0.0, z: 0.2}}\--rate10# 发布 Odometry含 headerros2 topic pub /odom nav_msgs/msg/Odometry\{header: {frame_id: odom}, child_frame_id: base_link, pose: {pose: {position: {x: 1.0}}}}\--rate54.3 用 --qos-reliability 指定 QoS# 以 BEST_EFFORT 可靠性订阅图像/雷达常用ros2 topicecho/scan --qos-reliability best_effort# 以 RELIABLE 发布ros2 topic pub /chatter std_msgs/msg/String{data: hi}--qos-reliability reliable4.4 过滤与采样# 只看数据字段省略 header 等ros2 topicecho/odom--fieldpose.pose.position# 只显示最近 N 条ros2 topicecho/odom--once五、ros2 bag数据录制与回放5.1 录制话题数据ros2 bag 是 ROS2 的数据记录工具把话题数据录到磁盘之后可回放调试、训练、复现 bug 的利器# 录制所有话题到默认目录bag 目录ros2 bag record-a# 只录制指定话题ros2 bag record /odom /scan /cmd_vel# 指定输出目录与名称ros2 bag record /odom-oodom_recording# 录制时附带 QoS 配置保证能回放ros2 bag record /scan --qos-profile-overrides\{sensor_msgs::msg::LaserScan: {reliability: best_effort}}5.2 查看与回放# 查看 bag 信息话题列表、消息数、时长ros2 bag info odom_recording# 回放 bag默认按原时间戳节奏ros2 bag play odom_recording# 回放指定话题、倍速ros2 bag play odom_recording--topics/odom--rate2.05.3 bag 的典型用途离线调试录下现场数据回家慢慢分析。算法验证同一份数据回放对比不同参数/算法效果。复现 bug把异常场景录下来反复回放定位问题。训练数据收集传感器数据用于机器学习。注意bag 回放时会重新发布数据时间戳可能保持原始值订阅者用header.stamp时要注意时间可能是过去的。六、实战完整数据流系统6.1 系统设计搭建一个完整数据流系统串联本篇所有知识点sensor_node(100Hz 发 /odom 10Hz 发 /scan) ↓ logger_node(订阅两者校验时间戳周期打印) ↓ ros2 bag record录制全程 ↓ 回放验证6.2 传感器节点双频双话题# py_pkg/dataflow_sensor.pyimportrclpyfromrclpy.nodeimportNodefromnav_msgs.msgimportOdometryfromsensor_msgs.msgimportLaserScanclassDataflowSensor(Node):def__init__(self):super().__init__(dataflow_sensor)self.odom_pubself.create_publisher(Odometry,odom,50)self.scan_pubself.create_publisher(LaserScan,scan,10)# 100Hz 里程计 10Hz 雷达self.odom_timerself.create_timer(0.01,self.pub_odom)self.scan_timerself.create_timer(0.1,self.pub_scan)self.x0.0defpub_odom(self):self.x0.001msgOdometry()msg.header.stampself.get_clock().now().to_msg()msg.header.frame_idodommsg.child_frame_idbase_linkmsg.pose.pose.position.xself.x self.odom_pub.publish(msg)defpub_scan(self):msgLaserScan()msg.header.stampself.get_clock().now().to_msg()msg.header.frame_idlasermsg.angle_min-3.14159msg.angle_max3.14159msg.angle_increment0.01msg.range_min0.1msg.range_max10.0# 模拟 360 个点全部 1.0m模拟圆形房间msg.ranges[1.0]*360self.scan_pub.publish(msg)defmain():rclpy.init()nodeDataflowSensor()rclpy.spin(node)node.destroy_node()rclpy.shutdown()if__name____main__:main()6.3 记录节点双订阅 时间戳校验# py_pkg/dataflow_logger.pyimportrclpyfromrclpy.nodeimportNodefromnav_msgs.msgimportOdometryfromsensor_msgs.msgimportLaserScanclassDataflowLogger(Node):def__init__(self):super().__init__(dataflow_logger)self.odom_subself.create_subscription(Odometry,odom,self.on_odom,50)self.scan_subself.create_subscription(LaserScan,scan,self.on_scan,10)self.odom_count0self.scan_count0defon_odom(self,msg):self.odom_count1# 每 500 拍打印一次避免刷屏ifself.odom_count%5000:self.get_logger().info(fodom #{self.odom_count}x{msg.pose.pose.position.x:.2f})defon_scan(self,msg):self.scan_count1ifself.scan_count%100:self.get_logger().info(fscan #{self.scan_count}points{len(msg.ranges)}ffirst{msg.ranges[0]:.1f})defmain():rclpy.init()nodeDataflowLogger()rclpy.spin(node)node.destroy_node()rclpy.shutdown()if__name____main__:main()6.4 完整运行流程# 终端 1传感器节点ros2 run py_pkg dataflow_sensor# 终端 2记录节点ros2 run py_pkg dataflow_logger# 终端 3验证频率ros2 topic hz /odom# ~100Hzros2 topic hz /scan# ~10Hz# 终端 4录制 10 秒ros2 bag record /odom /scan-odataflow_bag# 10 秒后 CtrlC 停止录制# 查看录制信息ros2 bag info dataflow_bag# 回放ros2 bag play dataflow_bag预期结果logger 每 5 秒打印一次 odom500 拍 × 0.01s每 1 秒打印一次 scanbag 录制后回放logger 重新收到数据——证明数据流 录制回放闭环打通。七、总结本篇文章把话题通信从能用推向工程可用自定义消息的完整使用链路、发布周期的三种控制方式固定/节流/动态、高频话题的性能要点、ros2 topic高级命令以及 ros2 bag 的录制回放。实战部分构建了一个 100Hz10Hz 双频数据流系统并验证闭环。关键要点回顾用自定义消息要在package.xml及 C 的 CMakeLists声明接口包依赖构建时先接口包后业务包。定时器驱动周期发布动态调频要先cancel()旧定时器。ros2 topic hz看实测频率实际频率 ≠ 设定频率。高频话题加大 QoS 队列深度Python 100Hz 内可用更高用 C。ros2 bag record/play录制回放是离线调试与复现 bug 的利器。Header是消息的公共字段承载时间戳与坐标系跨话题数据同步靠它。下一篇预告下一篇进入服务 Service 通信上——客户端与服务端话题是单向广播服务则是请求-应答。我们将理解 Request/Response 模型、同步与异步调用方式用 Python 与 C 实现服务端与客户端掌握ros2 service调试命令并对比话题 vs 服务的选型场景。