ROS 通信机制（一）：Topic 发布-订阅与消息结构（三）

ROS 通信机制（一）：Topic 发布-订阅与消息结构（三）

通用基础优先 · 第三篇｜ROS 通识系列｜Docker 实战

一图读懂本文：pub/sub 模型 → 消息结构 → 动手写节点 → QoS 契约 → 接回数采。

上一篇我们用命令行"看见"了 Topic 在广播。这一篇把它拆到底：Topic 上传的到底是什么、怎么用 Python 写自己的发布/订阅节点、以及 ROS 2 才有的 QoS 为什么重要。

一、Topic 的本质：异步、流式、多对多

把 Topic 想成一个广播电台：

• 发布者（Publisher）只管往频道里发，不关心谁在听；
• 订阅者（Subscriber）只管听，不关心谁在发；
• 一个频道（Topic）可以有多个发布者、多个订阅者——天然的一对多、多对多；
• 发布和订阅解耦且异步：订阅者没起来，发布者照样发（只是没人收）。

这正好匹配机器人：相机节点持续发图像，识别节点、录制节点、显示节点各取所需，彼此不阻塞。

二、Message（消息）：Topic 上跑的"数据类型"

Topic 里传的不是裸字节，而是有结构的 .msg 接口。ROS 内置大量标准消息，常用的：

JointState 是数采的命根子——我们录 bag，主要就是录它（配合 Image、IMU）。

新手视角 · 读懂 JointState 的四个字段：name＝关节叫什么（如 left_shoulder_pitch）；position＝关节转到哪个角度；velocity＝转多快；effort＝用了多大力气。一条 JointState 就是机器人在某一瞬间"全身姿势 + 速度 + 出力"的自拍。采集时它通常 100Hz-1kHz 连拍，是训练数据里最核心的一条流。
算一笔带宽账（数采设计前必做）：2 个 RGB 相机 + 1 个深度相机：图像 3 × 640×480×3B × 30Hz ≈ 83 MB/s；JointState 42 关节 × 8B × 1kHz ≈ 0.34 MB/s——图像才是存储大头，关节角几乎可以忽略。所以采集设计要提前回答：录原始图像还是压缩？每个相机多少 Hz？这些直接决定一天能采多少条 episode、硬盘能撑几天。

三、动手：用 Python 写发布/订阅节点

ROS 2 的 Python 客户端库叫 rclpy。下面两个最小节点，一个每秒发递增数字，一个打印收到的值。

# ———- publisher_node.py ———-
import rclpy
from rclpy.node import Node
from std_msgs.msg import Int32

class Pub(Node):
def init(self):
super().init(‘my_pub’)
self.pub = self.create_publisher(Int32, ‘count’, 10)
self.t = self.create_timer(1.0, self.cb) # 1Hz
self.n = 0
def cb(self):
msg = Int32(); msg.data = self.n; self.n += 1
self.pub.publish(msg)
self.get_logger().info(f’发布了 {msg.data}')

rclpy.init(); rclpy.spin(Pub()); rclpy.shutdown()

import rclpy
from rclpy.node import Node
from std_msgs.msg import Int32

class Sub(Node):
def init(self):
super().init(‘my_sub’)
self.create_subscription(Int32, ‘count’, self.cb, 10)
def cb(self, msg):
self.get_logger().info(f’收到了 {msg.data}')

rclpy.init(); rclpy.spin(Sub()); rclpy.shutdown()

运行（容器内两个终端，各自 source 后）：

python3 subscriber_node.py # 终端 A
python3 publisher_node.py # 终端 B

概念验证：发布者根本不知道订阅者存在；两边通过 Topic 名 'count' 自动对接。

四、QoS：ROS 2 才有的"通信质量契约"

ROS 1 的通信是"尽力而为"，丢就丢了。ROS 2 引入 QoS（Quality of Service），让你可以声明通信的可靠性与历史策略：

坑点：发布者和订阅者的 QoS 必须兼容才能通信。一个 RELIABLE 一个 BEST_EFFORT，两端连不上——这是 ROS 2 新手最常见的"为什么收不到数据"。
老手温故 · 官方预设的 sensor data QoS：相机/雷达驱动常用 qos_profile_sensor_data（BEST_EFFORT + KEEP_LAST(5)）——图像本就"下一帧更新"，重传旧帧没有意义；而关节指令这类"丢一帧会出事"的数据用 RELIABLE。注意兼容方向：发布端 BEST_EFFORT 时，订阅端必须 ≤ BEST_EFFORT——你用默认 RELIABLE 去订阅相机话题会连不上。调试利器：ros2 topic info /话题名 –verbose，直接看两端 QoS 声明是否匹配。

五、接回数采平台

我们真实采集时，本质就是：机器人本体（KUAVO）的驱动节点持续 发布 JointState / Image / Imu，我们的采集节点订阅这些 Topic 并录成 bag。Topic 名、消息类型、QoS，就是跨品牌适配层要先对齐的"接口契约"。

六、小结与下篇

• Topic = 异步广播，多对多，适合流式传感器数据；
• 传输的是结构化 .msg，常用 JointState/Image/PoseStamped；
• 用 rclpy 几十行就能写发布/订阅；
• QoS 是 ROS 2 必修课：可靠性/历史/持久性，且两端必须兼容；
• 下一篇：Service 与 Action——什么时候"广播"不够，要"问答"和"带进度的长任务"。

参考资料：ROS 2 官方 Tutorials（Writing a simple publisher and subscriber，Python）；rclpy API；QoS 策略 docs.ros.org/en/jazzy/Concepts/About-Quality-of-Service-Settings。示例基于 ros:jazzy / rclpy。