简介：面向ROS初学者的话题（Topic）通信学习代码包，以简洁示例展示发布者（Publisher）与订阅者（Subscriber）的创建、消息定义及rostopic调试方法。压缩包共11个文件，包含4个C++源文件、4个Python脚本，以及msg消息定义、package.xml和CMakeLists.txt构建配置，可对照学习两种语言下相同通信逻辑的实现差异。示例中使用了std_msgs和自定义Person消息，覆盖chatter话题的发布订阅、定时器发布、回调函数处理等关键环节，注释对每段代码的作用均有说明，便于读者快速上手。目前已有360人学习，适合刚接触ROS、希望从零理解节点间通信机制的入门者。通过运行发布者与订阅者节点，结合rostopic echo和rqt_graph等工具，读者能直观观察消息流动过程，为后续学习服务、参数服务器等高级特性打下基础。
1. 从一条消息说起：为什么 ROS 学习的第一个门槛是 topic
拿到一个叫learning_ros_topic.zip的压缩包，大概率是一份带示例代码的 ROS 入门资源。ROS 里 node 之间通信有四板斧：topic、service、action、param，而 topic 是绝对的主角。你写第一个发布者、第一个订阅者，跑rostopic echo看到数据在终端里滚出来，那一刻才算真正摸到 ROS 的门槛。这个包要解决的问题很直接：话题怎么写、怎么发、怎么收、怎么用命令行验证，以及消息类型怎么选。
适合读这篇的人有两类。一类是刚装完 ROS（无论是 Ubuntu 原生还是虚拟机）正愁从哪下手的初学者，另一类是写过几个节点但对话题机制细节没较过真的开发者——比如queue_size到底怎么影响延迟、latch该什么时候开、rostopic hz测出来的频率不稳定是不是代码写错了。这篇会按「模型 → 实现 → 命令行实战 → 排错与进阶」的顺序，把learning_ros_topic这个标题背后那套通用做法讲透。读完你能自己写出一对发布/订阅节点，能看懂rostopic全家桶每个子命令的用途，更重要的是，遇到话题数据对不上、频率忽高忽低这类问题，知道去哪里查、查什么。
2. 话题通信模型：发布/订阅的订阅缓冲区与消息类型
ROS 的 topic 机制核心是一个异步的发布/订阅模型。发布者往某个话题名上写消息，订阅者从同一话题名下读消息，双方互不认识，中间靠 ROS master（在 ROS 2 里换成了 Discovery 机制）牵线。理解这个模型的关键是三个词：话题名、消息类型、队列长度。这三个要素决定了一个 topic 能不能通、通得顺不顺。
2.1 一次数据传输的完整链路
发布端调用publish()时，消息先进入发布端的发送队列，由底层连接（XMLRPC 用于协商、TCPROS/UDPROS 用于传数据）交给订阅端。订阅端收到数据后放进自己的订阅缓冲区，rospy.Subscriber的回调函数在消息线程里被逐个执行。这里初学者最容易误解的一点：订阅回调不是逐条实时触发的，它依赖缓冲区转储。如果回调里处理耗时太长，而消息来太快，缓冲区会积压，积压溢出后新消息直接丢弃——表现出来就是接收频率远低于发布频率，但 CPU 占用却不低。
聊到消息类型，std_msgs/String是最简单的入门类型，但实际工程里 90% 的场景都该用自带类型的消息，比如sensor_msgs/Image、nav_msgs/Odometry、geometry_msgs/Twist。learning_ros_topic里的示例大概率会带你从std_msgs起步，但你得建立意识：消息类型定义了话题的「接口契约」。发布和订阅双方类型不一致时，运行时直接报错，master 不会帮你做隐式转换。
2.2 消息队列长度：queue_size的行为边界
新版 ROS（Noetic 及同代或更新版本）里，rospy.Publisher和rospy.Subscriber的构造参数都有一个必填项queue_size。这个参数常被随手写成 10 或 1，但它的语义需要掰开揉碎讲清楚。
| 场景 | queue_size 设置 | 行为表现 |
|---|---|---|
| 订阅端处理很快 | 10 | 缓冲几乎永远为空，消息实时处理 |
| 订阅端处理慢于发布 | 10 | 缓冲积压到 10，新消息按队列策略丢弃 |
| 订阅端处理严重滞后 | 1 | 只保留最新一条，丢帧最狠但延迟最低 |
| 发布端连接未建立 | 10 | 消息在发布端队列等待，连接建好后补发（但不等同于 latch） |
参数设置的核心矛盾是：queue_size设大了，滞后订阅者读到的数据越旧；设小了，瞬时峰值一来就丢消息。常见做法是「按最大容忍延迟倒推」——比如你的控制周期是 20 Hz，就希望拿到最近 50 ms 内的数据，那queue_size设置为 1～5 之间即可，不需要堆到 100。真正要重点关注的是rostopic hz测出的实际频率和发布频率对不上时，优先怀疑这个数。
2.3 消息类型与自定义消息的取舍
起步阶段用标准库消息足够，但如果你要传的是「坐标点 + 时间戳 + 自定义 ID + 置信度」这类组合数据，就该定义.msg文件。自定义消息的路径在package/msg/下，写好后改CMakeLists.txt的add_message_files和generate_messages，同时把message_generation加进find_package、message_runtime加进catkin_package。这个流程是 ROS 工程化的基本操作，learning_ros_topic里如果没有，建议自己补一遍。
提示：定义消息字段时，字段名不要用
type、data这类和 ROS 内部符号冲突的词。遇到「无法编译 / 生成的头文件找不到字段」时，先catkin_make后source，再看devel/include/包名/消息名.h里到底生成了什么。
3. 手写一个发布/订阅对：从learning_ros_topic到可跑的最小工程
假设你已经把learning_ros_topic.zip解压到catkin_ws/src/下，接下来按标准 catkin 流程走。这个包的名字本身可能就是一个 package，也可能是个包含多个 package 的工作区目录，常见做法是看里面有没有CMakeLists.txt和package.xml。有，就是一个包；没有，就catkin_create_pkg自己建一个。
3.1 工作区构建与依赖声明
cd ~/catkin_ws catkin_make source devel/setup.bash rospack find learning_ros_topic如果不能返回路径，说明包没被找到。先检查package.xml里的包名和目录名是否一致，再确认source是否生效。包名不一致是rospack find最常见的失败原因。
包内的CMakeLists.txt至少要确保rospy（或roscpp）被 find_package 到。如果你用的是 Python 写的节点，只需catkin_install_python或干脆让脚本可执行即可；C++ 节点则需要add_executable和target_link_libraries。这里有个新手常栽的坑：Python 节点文件放在scripts/下却没有chmod +x，直接rosrun会报 Permission denied。
3.2 发布端的最小实现与参数说明
建立一个scripts/talker.py，内容如下：
#!/usr/bin/env python3 import rospy from std_msgs.msg import String def main(): rospy.init_node('talker_node') pub = rospy.Publisher('chatter', String, queue_size=10) rate = rospy.Rate(2) # 2 Hz seq = 0 while not rospy.is_shutdown(): msg = String() msg.data = f"hello {seq}" pub.publish(msg) seq += 1 rate.sleep() if __name__ == '__main__': main()init_node的节点名必须是全工作区唯一的，重复会报node already registered。Publisher第一个参数是话题名，写法是「不带斜杠的全局名」或「带 ~ 的私有名」，用chatter这种裸名字时，节点在哪个命名空间，话题就在哪个命名空间。rospy.Rate(n)是一个自适应的睡眠器，它保证循环频率接近 n Hz，但如果你在循环里做了耗时操作，实际频率只会低于 n，不会等于 n。
3.3 订阅端的回调节奏与多线程安全
订阅端代码：
#!/usr/bin/env python3 import rospy from std_msgs.msg import String def cb(msg): rospy.loginfo("got: %s", msg.data) def main(): rospy.init_node('listener_node') sub = rospy.Subscriber('chatter', String, cb, queue_size=10) rospy.spin() if __name__ == '__main__': main()回调函数在 ROS 的消息处理线程里被调用，绝不要在回调里做阻塞操作——不要time.sleep、不要发同步请求。如果确实要做耗时处理，把数据丢进队列，由另一个线程去消费。
这里额外说明spin()的行为：rospy.spin()会阻塞主线程，让 ROS 继续处理回调。如果不调用 spin，Python 脚本跑到末尾就退出了，订阅器形同虚设。若在循环里手动rospy.sleep代替spin，必须保证回调有被空闲时间处理的机会。
注意：
rospy.loginfo输出到终端的等级可以被__log_level参数或rospy.logger调整。调试时看不到打印，先rospy.get_logger().setLevel(rospy.DEBUG)看看是不是日志等级过滤掉了。
3.4 用roslaunch一次拉起两端
<launch> <node name="talker" pkg="learning_ros_topic" type="talker.py" output="screen" /> <node name="listener" pkg="learning_ros_topic" type="listener.py" output="screen" /> </launch>output="screen"把日志打到终端，调试时必加。正式部署时可以去掉，由rosout统一收集。这里的name是节点的别名，和源码里init_node的名字可以不一致，但建议保持一致，否则维护时容易混乱。
4. 命令行实战：用rostopic和rqt_graph验证话题链路
learning_ros_topic这种包存在的意义不只是跑通代码，更重要的是教会你用工具观察系统。rostopic全家桶是你定位话题问题的主要武器。
4.1rostopic list与rostopic info的精确用法
rostopic list # 列出所有活跃话题 rostopic list -v # 详细列表，包含发布者/订阅者数量和类型 rostopic info /chatter # 查看单个话题的类型、发布者、订阅者rostopic list -v是排查「话题没出现」的第一招。发布者节点都启动了，但list里没有/chatter，说明发布者的init_node和你启动的 master 不在同一个ROS_MASTER_URI下；或者发布者的代码在创建 Publisher 前就崩了。
rostopic info /chatter的输出里如果只有 Publisher 没有 Subscriber，说明订阅端节点没连上——有可能是话题名拼错了，有可能是后启动的订阅端还没来得及完成连接。话题连接是懒建立的，订阅者启动后会有几百毫秒的协商延迟。
4.2rostopic echo的过滤与原始视图
rostopic echo /chatter rostopic echo -n 5 /chatter # 只回显 5 条，适合确认数据内容 rostopic echo /chatter/data # 只看 data 字段，自动过滤出字段路径 rostopic echo --noarr /pose # 不回显数组字段，避免刷屏-n是高频实用的参数。当你的话题频率很高（比如相机 30 Hz），直接echo会刷屏到没法看，-n 3直接精确到条数。
rostopic echo打印的数据是经过序列化再反序列化的，它显示的字段类型和.msg文件严格对应。如果你在代码里给一个Float64的消息赋了整数值，这里看到的就是1.0而不是1——这种信息在调试消息类型不匹配时很有用。
4.3rostopic hz与rostopic bw测量频率和带宽
rostopic hz /chatter rostopic hz -w 20 /chatter # 20 秒滑动窗口，看平均频率更稳 rostopic bw /chatter # 测量话题的字节带宽rostopic hz打印的频率低于预期时，从三个方向排查：发布端的Rate设置是否被耗时逻辑拖累；订阅端的队列是否溢出；系统是否存在 CPU 抢占、swap 等资源问题。
rostopic bw对排查图像或点云话题很关键。如果带宽远低于预期（比如相机 30 Hz 输出但 bw 只有几百 KB/s），大概率是图像数据在预处理环节被降采样或压缩了。这个话题我们会在下一节详细展开。
4.4rqt_graph可视化与rosnode info交叉定位
rqt_graph rosnode info /talkerrqt_graph显示节点和话题的关系图，但注意它只显示当前活跃的连接。如果你看到发布者和订阅者都存在但之间没有连线，说明话题名有微妙差异，比如发布端是chatter（全局名会解析成/chatter），订阅端是~chatter（解析成/节点名/chatter），这是命名空间不匹配的经典表现。
rosnode info能看到节点发布了哪些话题、订阅了哪些话题，是「图上没连线时先查谁」的答案——查发布者和订阅者各自看到的话题名是全名还是私有名。常见做法是直接rostopic list看全名，再对比两端代码里的字符串。
4.5 用rosbag录制与回放验证系统行为
rosbag record /chatter -O test.bag rosbag info test.bag rosbag play test.bag -r 1.0rosbag record可以指定多个话题，也可以-a录制全部话题。回放时默认按原始时间戳发布，-r 0.5表示半速回放。回放时订阅者不会区分数据是实时到达的还是回放的——这正是离线调试的基础。learning_ros_topic如果带示例 bag 文件，回放后配合rostopic echo是对着数据说话的最快验证方式。
注意：
rosbag play发布消息的时间间隔由 bag 文件里的时间戳决定，与系统当前时间无关。你rostopic hz测到的频率，反映的是录制时的真实频率，而不是当前系统的性能。
5. 数据不匹配与连接失败的排错流程
话题类问题的报错通常是隐蔽的——节点不报错，但数据就是不通。这时按顺序执行下面的排查步骤，不要跳级。
第一步，确认 master 和节点在同一个 ROS 网络里。echo $ROS_MASTER_URI，在多机通信时这个变量必然会被设置成某台机器的 IP。跨机器话题通信时，还要确认ROS_IP或ROS_HOSTNAME设置正确，否则节点之间能连 master 但数据连接建立失败。
第二步，确认消息类型完全一致。rosmsg show std_msgs/String和代码里的import对齐。若自定义消息在不同机器上有不同版本的.msg文件（字段顺序或类型不一致），消息会解析错乱。同一个工作区里不同 package 依赖了不同版本的自定义消息包，是最隐蔽的坑。
第三步，观察连接建立时序。如果订阅者启动得比发布者晚，它需要一点时间去发现话题并建立连接。在这个窗口内rostopic hz显示 NaN 是正常的，并不是代码出问题。
第四步，用rosnode ping验证节点间网络可达。这个命令检查的是到指定节点的 XMLRPC 连通性，不是话题数据的连通性。如果 ping 通但 echo 没数据，问题出在数据连接上——重点查防火墙是否放行了 TCP 数据端口（默认范围 1024～65535 随机分配）。
第五步，检查是否有多个 master 或节点名重复。如果你开了多个终端，每个终端都source过，但有的终端里ROS_MASTER_URI被覆盖成了别的值，那么所有节点各说各话，topic 必然不通。
排错时记住一个原则：从话题列表出发，而不是从代码出发。先用rostopic list看话题是否存在，再用rostopic info看两端是否都连上，最后才去翻代码。80% 的问题出在话题名拼写或命名空间上，而不是逻辑本身。
6. 三个进阶技巧：latch、rospy.wait_for_message与话题可视化调试
最后一章给三个平时不会被教材重点讲但实战高频的技巧，它们能让你的话题调试效率再上一个台阶。
6.1latch：给即时消息设置「记忆」
pub = rospy.Publisher('map_meta', nav_msgs/OccupancyGrid, queue_size=1, latch=True)latch=True的发布者在第一次 publish 后会保存这个消息，后续新订阅者连接时立即收到这一条，不需要等待发布者再次发送。典型场景是地图、静态变换、机器人描述等「低频但重要」的数据。如果不设 latch，新订阅者必须等到下一个发布周期才能拿到数据——如果发布频率是 0.1 Hz，那么最坏等待 10 秒。
使用注意：latch=True时queue_size实际意义不大，因为只有最新一条会保留，队列长度设 1 即可。调试时最容易误判的是，你改了发布数据后忘记重新publish，订阅者拿到的还是旧值——这不是 bug，是 latch 的预期行为（在 ROS 里被当作特性而非缺陷）。
6.2rospy.wait_for_message：单次同步查询，不写回调
import rospy from std_msgs.msg import String def wait_once(): msg = rospy.wait_for_message('chatter', String, timeout=5.0) rospy.loginfo("mo message: %s", msg.data)wait_for_message在需要「拉取一次当前值」的场景里非常顺手——你不想写一个订阅者节点，也不关心持续的数据流，只想知道这个时刻话题上有没有数据。实现原理是内部创建一个临时订阅者，收到第一条消息后立即 unregister。
两个坑：一是timeout参数单位是秒，超时抛ROSTimeoutException，不捕获会崩；二是wait_for_message依赖话题上有消息到达，如果发布频率极低，你可能会等满 timeout。这时可以先去rostopic hz确认发布者是否存活。
6.3 可视化调试：用rqt_plot和rviz的 Topic 面板
rqt_plot /chatter/datarqt_plot能绘制数值型消息字段随时间的变化曲线，适合观察传感器趋势、PID 输出、速度指令是否符合预期。它的地址路径支持和rostopic echo类似，支持/pose/x这样的字段路径——这意味着你不需要自定义消息也能直接画字段。
如果话题是sensor_msgs/Image或sensor_msgs/LaserScan，rviz的 Topic 显示面板是唯一靠谱的验证工具。在弹出的 Add 面板里选 By topic，找到对应的话题名，就能看到可视化结果。这里有个小技巧：rviz 里的 Fixed Frame 如果设置不对，点云或地图会显示在错误的位置——先确认 TF 树正常，再怀疑话题数据本身。
最后一个具体技巧是用rostopic echo加--filter按条件筛选：
rostopic echo --filter "m.data == 'hello 3'" /chatter这个命令只打印data等于hello 3的消息，在没有rostopic时写grep的效率完全没法比。语法是 Python 表达式，变量名代表消息对象，字段访问用点号。已知答案去验证、还是想从海量输出里捞特定数据，这都是在命令行上真正省时间的做法。到此，topic 这条路你已经从「看别人写」走到了「自己能查」的阶段，剩下的就是在自己的工程里去踩那些只属于你的坑了。
本文还有配套的精品资源，点击获取
