ROS 节点里最容易被当成"样板代码"抄下去的三个东西，大概就是ros::spin()、ros::spinOnce()和rospy.spin()了。教程里往往一行带过，一句"加上它就能收到消息"就结束。可真到了项目里，这三行决定的是整个节点的事件处理模型：谁来消费回调、什么时候消费、消费不过来会丢什么、CPU 被谁占满、Ctrl+C 之后能不能干净退场。我见过不止一个节点，订阅端的代码写得毫无问题，topic 名、消息类型、队列长度全都对，就是收不到数据，折腾半宿才发现主循环里少了那行ros::spinOnce()。也见过 Python 节点把业务逻辑一股脑写在rospy.spin()后面，跑起来之后那段代码一次都没执行过，排查半天以为是话题名拼错，其实是根本没理解这个函数是阻塞的。
这篇东西想解决的，就是"知道要写，但不知道为什么要写"这个状态。下面的内容会从回调队列这个底层结构讲起，把三个函数拆到"每一帧到底发生了什么"的粒度，再给出四种可以直接抄的代码骨架、spinOnce 调用频率与队列长度的定量算法，以及我在实际调试中撞过的几类典型故障。刚接触 ROS 的同学可以顺着代码块先跑起来，已经写过几个节点的老手可以直接跳到第 4、5 节，那里是参数设计和排查经验，属于文档里基本不会写、但线上调试一定会用到的部分。
1. 三个 spin 到底在干什么：先把心智模型建对
1.1 消息到了节点之后，是谁在调用你的回调
很多人对 ROS 的一个隐性误解是：消息一到，回调就自动执行了。其实不是。消息从网络（或者共享内存）到达进程之后，roscpp 的传输层会把它反序列化、包成一个"待执行的回调对象"，然后塞进一个队列里，仅此而已。真正把回调函数执行起来的是另一拨人——Spinner。你可以把队列理解成餐厅的出餐口，回调是已经做好的菜，Spinner 就是那个端着托盘往外送的传菜员。菜做出来了没人端，客人就一直饿着，这就是"消息收到了但回调没进"的本质。
这个队列在 roscpp 里叫CallbackQueue，默认情况下每个节点有一个全局的全局回调队列，通过ros::getGlobalCallbackQueue()拿到。订阅回调、定时器回调、服务回调、ros::Timer触发的那些函数，全都往这个队列里进。这里有个特别容易被忽略的点：定时器回调也走这个队列。所以一个节点哪怕只订阅了ros::Timer、没订阅任何 topic，也必须 spin，否则定时器永远不会触发。这个坑我踩过一次，当时以为是createTimer的周期参数单位写错了，实际改回毫秒后依然不动，最后才反应过来主循环里压根没有 spin 的调用。
至于rospy，情况完全不一样。Python 端的每个Subscriber在构造时就会拉起一个独立的接收线程（内部叫 ReceiveThread），回调直接在那个线程里执行，根本不需要主线程去"驱动"。这就是为什么很多人会发现：Python 节点里不写rospy.spin()，用rospy.sleep()或者Rate.sleep()照样能收到消息。因为回调压根不在主线程里跑。C++ 和 Python 在这件事上的行为差异，是新手最容易搞混的地方，后面 1.3 节会单独展开。
1.2 ros::spin() 和 ros::spinOnce() 的分水岭在哪
这两个函数的关系，用一个类比讲最清楚：ros::spinOnce()是"去出餐口看一眼，现在有什么端什么，端完就走"；ros::spin()是"我就在这儿站到打烊为止"。
从实现上看，ros::spinOnce()本质就是一次getGlobalCallbackQueue()->callAvailable(ros::WallDuration(0))。它把队列里当前此刻已经就绪的回调全部取出来执行一遍，执行完立刻返回。注意"当前此刻"这个限定——它在等待超时上给了 0，也就是说不会为了等新消息而阻塞。这一行执行完，控制权立刻回到你手上，你可以接着做自己的计算、发自己的消息、睡自己的觉。
ros::spin()则是一个内部循环，大致长这样：
// 简化后的 roscpp 内部实现，帮助理解，不是逐字源码 void SingleThreadedSpinner::spin() { if (!callback_queue_) callback_queue_ = ros::getGlobalCallbackQueue(); while (ros::ok()) { // 每次最多等 100ms，有回调就执行，没有就空转回来继续等 callback_queue_->callAvailable(ros::WallDuration(0.1)); } }看清楚这个while (ros::ok())就知道，ros::spin()不会返回，直到节点被关闭（ros::shutdown()被调用，或者ros::ok()变假）。这也是新手最常见的困惑来源：在ros::spin()后面写清理代码、写"发布结果"、写日志收尾，全都不可能执行。它下面那几行对程序来说是不存在的。
两者的取舍其实非常明确：你的节点如果有自己的主循环（读传感器、算控制量、跑状态机、做可视化渲染），就必须用spinOnce()；如果节点是纯事件驱动的（来一条消息处理一条，没有自己的节拍），用ros::spin()最省事。而且如果后面你还要加多线程，ros::spinOnce()的写法基本可以平滑过渡到AsyncSpinner，ros::spin()的写法就得整体重构。
1.3 rospy.spin() 为什么看起来什么都没做
把rospy.spin()的源码翻出来看，会发现它做的事情非常"朴素"——基本就是一个while not rospy.is_shutdown(): time.sleep(0.001)的空转循环。它不消费队列，不调用回调，甚至在主线程里什么都没算。那它的意义在哪？
意义在于占住主线程，让进程别退出。Python 脚本从main一路执行到最后一行就结束了，rospy.init_node()之后进程并不会因为注册了订阅者就一直活着。你需要在末尾放一个不会返回的调用把主线程挂住，rospy.spin()就是干这个的。它是一个"活着"的声明，不是一个"工作"的声明。
理解这一点之后，很多现象就说得通了。比如你在rospy.spin()后面写了关节点清理、写了结果汇总，那永远是死代码；比如你想在主循环里做点周期性的事情，用rospy.spin()就完全是走错方向，应该换成：
# 错误写法：后面的逻辑永远不执行 rospy.spin() do_something_periodically() # 永远不会到这里 # 正确写法：主循环自己控节拍 rate = rospy.Rate(20) # 20Hz while not rospy.is_shutdown(): do_something_periodically() rate.sleep()还有一个非常反直觉的点：在 Python 里，while not rospy.is_shutdown(): ... rate.sleep()这种主循环里，回调依然是正常执行的，因为它们在独立的接收线程里。所以 Python 侧的"spin 与主循环"其实不构成互斥关系，你完全可以在主循环里干活，同时回调在后台处理消息。这也意味着 Python 节点天然就是多线程的，共享数据的加锁问题从第一天就存在，很多人在写 Python 节点时没意识到这点，直到出现两个线程同时改一个列表导致IndexError才反应过来。
2. 底层机制拆解：队列、Spinner 与线程模型
2.1 CallbackQueue 是怎么攒下待处理回调的
CallbackQueue内部维护的是一个按"就绪时间"排序的结构，回调对象带着一个待执行时间戳入队，callAvailable()每次把时间戳已经到了的取出来执行。ros::WallDuration(0.1)这个参数的作用是：如果队列里暂时没有就绪回调，最多在这里等 100 毫秒；一旦有东西进来立刻返回执行，不用傻等满 100ms。这也是为什么ros::spin()在空闲时的 CPU 占用很低——它大部分时间阻塞在等条件变量上，不是在忙等。
队列还有一个关键属性：容量。订阅时给的那个queue_size参数，某种程度上就在影响这个容量上限。队列满了之后的行为是丢最老的还是新的，取决于具体实现和参数。这就引出了第 4 节要详细算的问题：你的主循环多久调一次spinOnce()，直接决定了队列里会积压多少条消息；如果积压量长期超过容量，丢包就是必然的，而且丢得悄无声息，没有任何报错。
这里我想强调一个实操中很常见的误区：很多人以为queue_size设大一点就"更安全"，其实是在给自己埋雷。设成 1000，看起来不会丢消息了，实际上你的回调处理的是几百毫秒之前的"陈旧数据"，对于控制类节点来说，拿旧数据算控制量比丢数据还危险。队列长度应该根据"能容忍的最大延迟"倒推，而不是无脑加大。
2.2 单线程 Spinner 的工作循环长什么样
单线程 Spinner 的整个生命周期就是一个串行循环：等回调 → 执行 → 等回调 → 执行。这意味着所有回调共享同一个线程，彼此之间完全串行。A 回调执行的时候，B 回调只能排队等。如果你的订阅回调里做了耗时的操作（比如读一个大文件、调一次同步服务、做一次重计算），其他所有话题的回调都会被阻塞，表现出来就是"某个话题明明发布正常，订阅端就是不响应"。
这也解释了一个经典死锁场景：在订阅回调里调用actionlib的waitForResult()。因为waitForResult()需要等待结果回调被触发，而结果回调又要靠同一个 Spinner 线程来执行——线程正卡在waitForResult()里，永远轮不到执行结果回调，于是双方互等，程序卡死。规则很简单：单线程 Spinner 下的回调函数，必须短、必须非阻塞。任何"我要在这里等一下结果"的写法都是定时炸弹。
判断自己的回调是不是太慢，有个土办法：在回调开头和结尾各打一个时间戳，跑一段时间后看最大耗时。我一般会盯住 P99 而不是平均值，因为平均值好看但偶发的长尾回调才是真正导致队列堆积的元凶。
2.3 多线程场景：AsyncSpinner 与 MultiThreadedSpinner
当单线程扛不住的时候，ROS 提供了两种扩展方式，都很好用但语义不同。
ros::AsyncSpinner是"后台起 N 个线程，主线程还给你"：
ros::AsyncSpinner spinner(4); // 4 个线程消费回调队列 spinner.start(); // 非阻塞，立刻返回 // 主线程可以继续做自己的事，或者直接挂在这里等关闭 ros::waitForShutdown();ros::MultiThreadedSpinner则是"阻塞式多线程 spin"，它会一直卡在spin()里不出来，所以你不能在它后面写主循环：
ros::MultiThreadedSpinner spinner(4); // 4 个线程 spinner.spin(); // 阻塞，直到 shutdown两者选哪个，看你的主线程还有没有活。有活（要跑控制循环、要算规划）就用AsyncSpinner；纯粹事件驱动、主线程没事干，用MultiThreadedSpinner更省心。
但多线程不是免费的午餐，它立刻带来两个新问题。第一是回调可能并发执行，同一个话题的两条消息可能同时被两个线程处理，如果你的回调里有共享状态（一个计数器、一个缓存表、一个std::vector），必须自己加锁。第二是执行顺序不再有保证，多线程下"先来的消息先处理完"这个假设不成立。对于需要严格时序的数据（比如 IMU 积分、里程计累加），要么把这类回调单独放一个线程，要么就在回调内部做时间戳排序，别指望 Spinner 帮你保序。
ROS 1 里还有一个更细粒度的玩法：自定义回调队列。可以给某个NodeHandle单独绑一个CallbackQueue，然后自己决定这个队列用几个线程去消费。这个做法的价值在于隔离——把每秒上百条的高频传感器话题和低频的控制指令分开，避免高频话题把队列塞满、把低频的指令回调饿死。这个方案稍微复杂一些，后面第 3 节给代码。
3. 实操骨架：四种典型写法与适用边界
3.1 主循环 + spinOnce 的标准骨架
这是工业现场用得最多的一种结构，也是我最推荐的默认写法。它的核心是把"处理回调"变成一个显式动作，放在自己主循环的固定位置上，节拍完全可控。
#include <ros/ros.h> int main(int argc, char** argv) { ros::init(argc, argv, "my_node"); ros::NodeHandle nh; Subscriber sub = nh.subscribe("sensor_data", 10, sensorCallback); ros::Publisher pub = nh.advertise<MsgType>("result", 10); ros::Rate loop_rate(50); // 主循环 50Hz，周期 20ms while (ros::ok()) { // 1. 先把这一周期到达的回调全部消费掉，保证下面用的是最新数据 ros::spinOnce(); // 2. 基于最新数据做业务计算 processAndPublish(pub); // 3. 按固定频率进入下一轮 loop_rate.sleep(); } return 0; }这段代码里有两个位置选择值得说明。spinOnce()我放在循环的最前面，理由是"先拿最新数据再算"，避免用上一周期的陈旧值做决策。如果你把它放在循环末尾，那么第一轮循环时回调还没被消费过，业务代码拿到的就是初始值，这种"第一帧空跑"在某些初始化敏感的场景里会出问题。
关于ros::Rate的精度，实测下来在普通 Linux 桌面环境下，50Hz 的Rate::sleep()抖动通常在零点几毫秒量级，够用；但如果你需要微秒级严格周期，Rate就不够了，得考虑用实时内核或者干脆硬件定时。另外ros::Rate::sleep()会扣掉循环体的执行时间，也就是说它保证的是"整体周期"，不是"休眠时长"，这一点很多人搞反了，以为循环体耗时越长周期越长——实际上是耗时长到接近周期时，Rate会发出"loop rate too slow"的警告。
3.2 ros::spin() 阻塞式写法与控制权交接
纯事件驱动的节点，代码可以短到十几行：
int main(int argc, char** argv) { ros::init(argc, argv, "event_node"); ros::NodeHandle nh; nh.subscribe("input_a", 10, callbackA); nh.subscribe("input_b", 10, callbackB); ROS_INFO("node started, entering spin"); ros::spin(); // 这里开始阻塞，后面的代码不会执行 ROS_INFO("never printed"); // 骗你的，永远不会打印 return 0; }这种写法的优点是极简，不需要考虑节拍，来了就处理。缺点也很明显：所有逻辑都必须塞进回调，节点没有任何自己的节拍感。如果你的业务里有"每隔 100ms 检查一次状态机"这样的需求，在这个结构下只能靠ros::Timer来模拟，而不是直接写循环。
用ros::Timer模拟周期任务时，别忘了 1.1 节说过的那个坑——定时器回调也走队列，也需要 spin，好在这里ros::spin()已经解决了。定时器的写法：
void timerCallback(const ros::TimerEvent& e) { // 这里的逻辑每 100ms 执行一次 } ros::Timer timer = nh.createTimer(ros::Duration(0.1), timerCallback); ros::spin();TimerEvent里有current_expected、current_real、last_expected、last_real这几个字段，能看出定时器是不是被拖慢了。如果current_real和current_expected差距越来越大，说明回调队列堵了，这比什么都灵，是个天然的"系统健康度探针"。
3.3 定时器 + spinOnce 的组合
有些场景既想要主循环的自主性，又不想手写Rate控节拍，可以混着用：用spinOnce()保持对回调队列的控制，用Timer承担周期任务。不过要说实话，这两个一起用，收益通常不大，反而让代码里出现两套节拍，后来维护的人容易懵。我的建议是二选一，别混。
真正值得用混合方案的是多队列隔离场景，前面 2.3 节提过。写法是：
ros::CallbackQueue fast_queue; ros::CallbackQueue slow_queue; ros::NodeHandle nh_fast; nh_fast.setCallbackQueue(&fast_queue); nh_fast.subscribe("high_freq_sensor", 1, fastCallback); // 高频，队列短 ros::NodeHandle nh_slow; nh_slow.setCallbackQueue(&slow_queue); nh_slow.subscribe("command", 10, slowCallback); // 低频，队列可以长一点 // 高频队列用独立线程全力消费 ros::AsyncSpinner fast_spinner(2, &fast_queue); fast_spinner.start(); // 主循环里只处理低频队列，保证指令不会被高频数据饿死 ros::Rate rate(20); while (ros::ok()) { slow_queue.callAvailable(); // 只消费低频队列 do_business_logic(); rate.sleep(); }这个写法的价值在调试时才体现得出来。我做过一个节点，订阅一个 200Hz 的传感器话题和一个手动下发的控制指令话题，刚开始共用一个队列，结果指令回调经常被排在几百条传感器消息后面，操作员按下按钮之后要等一两秒才有反应，体验极差。拆队列之后延迟降到几十毫秒，问题当场解决。
3.4 Python 侧 rospy.spin() 与主循环的取舍
Python 的选择更简单，因为回调本来就在独立线程里跑，你唯一要决定的是"主线程干什么"。
情况一，什么都不用干，纯事件驱动：
import rospy def callback(msg): rospy.loginfo("received") rospy.init_node("py_node") rospy.Subscriber("input", MsgType, callback) rospy.spin() # 挂住主线程情况二，主线程有周期任务：
rospy.init_node("py_node") rospy.Subscriber("input", MsgType, callback) pub = rospy.Publisher("output", MsgType, queue_size=10) rate = rospy.Rate(20) # 20Hz while not rospy.is_shutdown(): result = compute_something() pub.publish(result) rate.sleep() # 这里不用 spin，回调在别的线程照常跑这里有个我需要反复提醒的点：Python 版本不要在while循环里再调rospy.spin()，那是个死循环，一旦进去主循环就再也不转了。有些人为了"保险"加了一行，结果整个周期任务停摆，还以为是Rate出问题。
另外，Python 的rospy.Rate.sleep()和time.sleep()在行为上有个差异：前者在仿真时间（/use_sim_time）下会跟着仿真时钟走，后者走的是墙钟时间。如果你的节点跑在仿真环境里，周期任务用time.sleep()会导致仿真时间和实际进度脱节，一定要用rospy.Rate。
4. 频率与队列的参数设计：spinOnce 该多久调一次
4.1 从订阅频率和回调耗时倒推循环周期
这部分是我最想展开的，因为它能直接回答"为什么我的消息在丢"这个高频问题。先定义几个量：
| 符号 | 含义 | 典型取值 |
|---|---|---|
| f | 话题发布频率（Hz） | 100 |
| t | 单次回调的执行耗时（ms） | 2 |
| T | 主循环周期（ms） | 20 |
| N | 单个周期内平均到达的消息条数 | f × T / 1000 |
| C | 单个周期内回调处理总耗时（ms） | N × t |
| Q | 订阅时设置的 queue_size | 待定 |
举一组具体数字。假设订阅一个 100Hz 的 IMU 话题，回调里做一次简单的低通滤波，实测耗时 2ms。主循环跑 50Hz，即 T = 20ms。
那么 N = 100 × 20 / 1000 = 2，也就是每个周期平均会积压 2 条消息等待处理；C = 2 × 2 = 4ms，占整个 20ms 周期的 20%，CPU 上是完全健康的。
但如果主循环降到 10Hz，T = 100ms，那么 N = 10，C = 20ms，占周期的 20%——占用比没变，但最坏延迟变了。最坏情况下，一条消息在第 0 毫秒刚进队，主循环恰好在第 0.1 毫秒开始处理别的，要等到下一个周期第 100 毫秒才被消费，延迟接近 100ms。对于需要做姿态估计的场景，100ms 的延迟基本等于不可用。
由此得出一个粗糙但好用的估算式：最坏端到端延迟 ≈ T + t。想控制延迟在 30ms 以内，主循环周期就得压到 20ms 以下。
4.2 queue_size 与丢包的定量关系
队列长度的下限，应该满足"一个周期内到达的消息全都装得下"，也就是：
queue_size ≥ ceil(f × T / 1000) + 余量
沿用上面的例子，f = 100Hz，T = 20ms，单周期到达 2 条，那么queue_size = 2理论上刚好够，但考虑到发布端抖动、系统调度延迟，实际应该给 5~10。设成 10 的话，最坏情况下能容忍 5 个周期的突发堆积。
如果 queue_size 设成 1 会怎样？这就是"只保留最新一帧"的模式，丢包率会很高，但对于控制类应用，有时候主动丢旧数据反而是对的——你只关心最新的传感器读数，历史值毫无意义。这种情况下queue_size = 1配合较短的主循环周期，效果比queue_size = 100好得多。这个取舍很多人没想明白，看到丢包就去加大队列，结果引入了更大的延迟。
我把常见场景的选择整理成一张表，可以直接对着用：
| 场景 | 关注点 | 建议 queue_size | 建议主循环周期 |
|---|---|---|---|
| 高频传感器（IMU、编码器） | 只要最新值 | 1 ~ 2 | ≤ 消息周期的 1/2 |
| 控制指令（手动下发） | 一条都不能丢 | 10 ~ 50 | 不必太快，10~20Hz 足够 |
| 图像/点云 | 允许跳帧 | 1 | 按处理能力定 |
| 状态/事件通知 | 不能丢，频率低 | 50 以上 | 按业务需要 |
顺带说个容易忽略的点：queue_size在订阅端和发布端都存在，含义还不完全一样。发布端的队列是发送缓冲区，满了会丢；订阅端的队列是接收待处理缓冲区。两边都调正确才有意义，只调一边等于白调。我见过有人把订阅端设成 100，发布端还是默认的 10，结果还是丢——因为消息在发出去的路上就已经被丢了。
4.3 休眠策略与 CPU 占用的平衡
ros::Rate用起来方便，但它的精度上限受限于系统调度。如果你把频率设得极高（比如 1000Hz），实际跑出来可能是 800~1200Hz 乱跳，因为sleep的最小精度就是一个调度时间片。任何靠 sleep 实现的高频循环，都不要指望频率特别准。
真要精确，路子有两条：一是用ros::WallTimer或者系统级的定时器 API 来做节拍源；二是干脆不用固定频率，改成事件驱动，靠回调本身驱动计算（这也是ros::spin()那套写法的优势）。
反过来，如果主循环里spinOnce()被调得过于频繁（比如不加Rate，裸while循环里一直调），会怎样？callAvailable在没有就绪回调时会阻塞等待，等不到就一直等到超时，所以 CPU 不会被占满，但循环的实际节奏会变成"有消息就狂奔、没消息就卡住"，非常不均匀。这种不均匀在调试时很难发现，因为日志看起来正常，只有打出时间戳才能看出来。所以我的习惯是：只要写了spinOnce()，就一定给它配一个Rate，让节拍稳定下来。
还有一点关于ros::Duration::sleep()和ros::Rate::sleep()的差别，我实测过多次：单次Duration(0.001).sleep()的实际耗时通常在 1.05ms 到 1.2ms 之间，会稳定地偏大一点点；Rate则会做累积补偿，长时间跑下来平均周期更准。所以做固定频率循环，一律用Rate，不要自己在循环末尾手写Duration睡眠。
5. 故障排查实录：那些让你怀疑人生的卡点
5.1 回调压根不触发
这是最高频的一类问题，排查顺序建议固定下来，别东一榔头西一棒子。
第一步，确认消息真的到了进程。用rostopic hz看发布端频率是否正常，再用rostopic echo确认内容能打出来。这两步都通过，说明发布端没问题。
第二步，检查话题名解析。ROS 的话题名有相对名、全局名、私有名三套规则，nh.subscribe("data", ...)在命名空间下会被解析成/ns/data，而你在另一个节点里可能订阅的是/data。这个坑极其常见，尤其是节点被launch文件加ns属性包裹之后。解决办法是用ros::names::resolve()打印一下最终名字，或者干脆全用全局名。
第三步，检查 spin 有没有被调用。C++ 节点里如果只有nh.subscribe(...)然后进了一个不带spinOnce的while循环，回调永远不执行，表现就是"订阅了但收不到"。这是自己给自己挖的坑，但每年都有大量人掉进去。
第四步，检查回调本身有没有被"埋"。如果回调里第一行是个耗时的同步操作，或者异常在回调内被吞掉了，也可能看起来像"没触发"。加一行ROS_INFO在回调的第一行，能快速区分是"没进来"还是"进来了但挂了"。
5.2 程序卡死、Ctrl+C 退不掉
ros::spin()后面接了ros::shutdown()之类的收尾逻辑，是常见的"退不掉"原因之一——spin()本身会在 shutdown 后返回，但如果你在回调里又调了一次ros::spin()，整个线程就进入了嵌套等待，信号处理会很别扭。规则是：ros::spin()在一个进程里只应该出现在最外层，回调内部绝不要再调。
第二种卡死来源于阻塞式调用。前面提过的waitForResult()，还有在回调里等待ros::service::call()返回的场景，都得小心。特别是单线程 Spinner 下，任何"在回调里等另一个回调"的写法都是死锁配方。
第三种是while (true)而不是while (ros::ok())。前者在收到 SIGINT 后不会退出，因为你压根没检查关闭标志。这个改动成本极低，但漏掉的人不少。同时建议在信号处理上不要自己另起炉灶，ros::init已经帮你装了 SIGINT 处理器，正常用ros::ok()就能优雅退出。
Python 侧的"退不掉"通常是另一个原因：某个接收线程在join时被卡住，或者你在回调里开了个不带超时的Queue.get()。Python 线程退出不会强制终止，主线程rospy.spin()返回之后，如果子线程还在跑，进程会挂在那里不退出。这种时候只能靠daemon=True或者显式超时来收尾。
5.3 多线程下的数据竞争与闪退
用了AsyncSpinner之后，回调并发执行的场景必须重新审视所有共享状态。最典型的症状是：单线程跑得好好的，改成 4 线程之后偶发崩溃或者数据错乱，而且现象不稳定、难以复现。
排查方法很直接：把所有被回调访问的成员变量列出来，逐个问一句"这个变量会不会被两个线程同时读写"。C++ 里的std::vector被并发push_back会直接导致堆损坏，崩溃位置往往在完全无关的代码里，非常难定位。加锁是最省事的解法，虽然性能有损失，但比起排查崩溃的时间成本，这个代价微不足道。
还有一个容易被忽略的点：ros::Publisher在多线程下发布同一条消息时，publish()本身是线程安全的，但如果消息对象是共享的成员变量、你在一个线程里改、另一个线程里发，那就是数据竞争。稳妥的写法是每条消息在栈上构造，发布完就析构，不共享。
5.4 一张速查表
| 现象 | 最可能原因 | 快速验证 | 处理方式 |
|---|---|---|---|
| 订阅无回调 | 缺 spin 调用 | 循环里加spinOnce() | 补上，或改用ros::spin() |
| 只订阅 topic 正常，Timer 不触发 | 同上，Timer 也走队列 | 同左 | 同左 |
收不到但rostopic echo正常 | 话题名命名空间不匹配 | 打印ros::names::resolve() | 改用全局名或对齐命名空间 |
| 延迟越来越大 | 回调耗时 > 消息间隔 | 打回调耗时时间戳 | 精简回调，或改多线程 |
| 偶发数据错乱 | 多线程共享状态 | 关掉多线程看是否复现 | 加锁或拆分数据 |
| Ctrl+C 后进程不退出 | while(true)或回调内阻塞 | 检查循环条件与回调体 | 改while(ros::ok())，去掉阻塞调用 |
| 定时器周期不准 | 队列拥堵或系统负载高 | 看TimerEvent的时间差 | 缩短回调，或降低频率 |
Python 节点spin后逻辑不执行 | rospy.spin()是阻塞的 | 看代码层级 | 改成主循环 +rate.sleep() |
6. 几条不太容易从文档里学到的经验
第一条，关于spinOnce()的调用位置，我的习惯是把它放在循环开头，但在一些特殊场景会刻意放到末尾：如果业务是"基于上一个周期的数据生成指令"，放末尾反而更符合语义。这个没有标准答案，关键是一旦定了就写进注释，因为下一任维护者大概率会怀疑这个顺序。
第二条，调试回调问题时，最有效的工具不是看日志，而是打时间戳。在回调进入和退出各记一次ros::Time::now()，在主循环每一轮也记一次，跑几十秒后统计这几个序列，队列堆积、回调耗时、循环周期全都能看出来。这种做法比任何猜测都快。
第三条，ros::spin()和ros::spinOnce()不要同时出现在一个进程里。见过有人写了ros::spin()之后又在Timer回调里调spinOnce()，逻辑上就是"在吃饭的时候又去端菜"，短时间没事，遇到特殊时序就卡死。事件模型只应该有一套。
第四条，Python 节点尽量别在回调里做重活。因为回调在独立线程跑，一旦这个线程被一个耗时 5 秒的操作占住，后续消息会在这个订阅者内部积压，而其他订阅者的线程还在正常跑，表现出来的现象是"有的 topic 正常、有的 topic 延迟爆炸"，非常迷惑。重活要么丢到主循环，要么丢到线程池。
最后一条，也是我觉得最值得强调的：这三个函数看起来像模板，实际上它们是节点架构的"总开关"。写第一行main的时候，就该先想清楚节点是事件驱动还是周期驱动，再决定用哪一套。等代码写了几百行再回头改这个模型，改的就不是一两行，而是整个控制流。
