ROS2 QoS 配置策略

同一个话题，ros2 topic list 能看见，自己的回调却一直不执行；相机画面正常，但图像和 IMU 的同步回调不触发；机器人收到速度指令后，在通信断开时还沿用上一次速度——这些现象不一定是同一种问题。

QoS 要解决的是通信双方怎样交换和保留数据。它不是线程调度器，也不是控制器，更不是机器人安全机制。理解这几个边界，比背下 RELIABLE 和 BEST_EFFORT 更重要。

本文以 Ubuntu 22.04 / ROS 2 Humble 为基线，按“概念 → 配对 → 配置 → 复现 → 工程取舍”的顺序展开。API 与默认值核对到官方 Humble 分支；文中的实验给出了预期现象，尚未在本仓库的 ROS 开发镜像中完成运行验收，不能把示例当成已测量的延迟或真机安全结论。

如果接下来要把指令接到关节、电机和控制循环，可以继续阅读：ROS 2 控制系统：ros2_control、Controller Manager、Executor 与硬件接口。

1. 先建立一条完整的数据链

传感器采样 / 指令生成
        │   消息时间戳、序号、业务有效期
        ▼
Publisher → DDS 发送缓存 → 网络 / 进程内传输
                              │
                              ▼
                      DDS 接收缓存
                              │ Executor 调度回调
                              ▼
                     应用缓存 / 同步器 / 控制器
                              │ 检查新鲜度与状态
                              ▼
                         执行结果反馈

这张图里至少有三种“到了”：

1. 数据到了接收端中间件。
2. Executor 调用了订阅回调，应用拿到了消息。
3. 应用接受并完成了消息对应的工作，例如电机真的运动到目标位置。

不能把第 1 步直接当成第 3 步。可靠通信也不能证明回调没有阻塞、控制器接受了目标，或者驱动器没有报错。Humble 的 Executor 通常通过 wait set 判断实体是否有工作，消息在被取走前仍由中间件管理；因此“节点还活着”不代表“回调及时执行”。官方 Executor 机制说明

1.1 Profile 不是整个节点的统一开关

一组 QoS 策略组成一个 profile。一个节点内的不同 publisher、subscription、service 可以有不同配置；同一个话题上的多个端点，也可能只有一部分彼此兼容。

例如定位节点可以同时使用：

• 图像订阅：允许丢帧，优先处理新画面。
• 地图订阅：需要晚加入时取得当前地图。
• 定位结果发布：面向消费者分别评估可靠性和缓存。
• 重定位服务：需要请求、应答和明确的超时处理。

所以不要问“这个节点应该用哪种 QoS”，先问“这个端点消费的是什么数据，过期后还有没有价值”。

2. 七组策略，分别约束什么

2.1 History 与 Depth：缓存数量，不是缓存时间

KEEP_LAST(N) 表示有限历史；KEEP_ALL 表示尽量保留历史，但仍受到 DDS 资源限制。depth 只在 KEEP_LAST 下表达这个数量上限。

发布侧和订阅侧有各自的缓存，并不是双方共享一个长度为 N 的队列。应用可能另外还有工作队列、message_filters 队列和算法内部窗口。

假设相机 30 Hz，算法每帧需要 80 ms，算法理想吞吐量只有约 12.5 Hz。把订阅深度从 5 调到 100，并不会让计算变快：

• 如果逐帧处理，积压会把实时画面变成“几秒前的录像”。
• 如果业务只需要当前目标位置，应该考虑丢弃待处理旧帧，而不是无限加缓存。
• 如果业务是离线重建，保留全部原始帧有价值，但要通过录制、磁盘吞吐和离线计算设计保证完整性。

粗略地说，频率为 f 的稳定数据流，N 个样本覆盖的时间跨度约为 (N - 1) / f。这只是历史窗口估计，不是延迟上界；网络、调度和应用队列还会增加等待时间。

Humble 的类型定义还区分 history、depth 与其他时长策略，不能把一个 depth=10 理解成“消息 10 ms 后过期”。RMW QoS 类型定义

2.2 Reliability：需要补传，不等于业务必然完成

BEST_EFFORT 更接近“尝试发送，允许在传输条件不足时丢失”；RELIABLE 要求可靠交付机制，可能涉及确认、重传和额外等待。

原先笔记里的“RELIABLE 确保消息一定会送达”过于绝对。更准确地说：它提供中间件层的可靠性语义，不承诺在任意断网、进程崩溃、资源耗尽、消息过期等条件下，应用都能及时处理每一条消息，也不承诺业务上的 exactly-once。

对控制系统有一个很实际的区别：

连续速度设定值：旧的 0.8 m/s 往往不如当前的 0.0 m/s 有价值
执行一次夹爪闭合：需要知道是否接受、是否执行、是否成功

前者通常需要小缓存、新鲜度和失联停机；后者更需要带请求标识的服务或 Action、执行状态与超时恢复。二者不能都靠“调成 reliable”解决。

可靠性也不等同于传输协议名称：不能写成“reliable 就是 TCP，best effort 就是 UDP”。DDS/RMW 的实现与传输选择是另一层问题。ROS 2 Humble QoS 概念

2.3 Durability：晚来的读者，要不要补历史

VOLATILE 不要求为晚加入订阅者保留旧数据；TRANSIENT_LOCAL 让仍存活的发布端保存一定历史，用于符合要求的晚加入订阅者。

一个典型时间线：

t0  发布端发布地图 A
t1  没有新地图产生
t2  RViz / 导航节点才启动

volatile 订阅：等待下一次新地图
transient_local 订阅 + 兼容发布端：可取得缓存中的地图 A

需要注意四件事：

1. 想补历史，发布端和订阅端都应配置 TRANSIENT_LOCAL，只改一边不够。
2. 历史数量受发布侧 history/depth 等条件约束；深度 1 通常是“最新一份快照”，不是全部版本。
3. TRANSIENT_LOCAL 不是磁盘数据库；原发布端消失后，不能指望新进程自动继承它的缓存。
4. 不要给持续运动指令随手加持久性，否则刚启动的消费者可能把旧指令当成当前意图。

2.4 Deadline：应该多久有一次数据

Deadline 关注相邻数据更新之间允许的最大间隔。例如预计 50 Hz 更新，可以按业务允许的抖动设定 30 ms，而不是机械地把 deadline 设成恰好 20 ms。

它有两个容易混淆的层次：

• 匹配阶段：发布端声明的 deadline 能否满足订阅端要求。
• 运行阶段：匹配后是否出现 deadline missed 事件。

如果发布端声明 20 ms、订阅端允许 50 ms，约束可以满足；反过来发布端声明 50 ms、订阅端要求 20 ms，则不能满足。有限时长下可以记作：

offered_deadline <= requested_deadline

Deadline 不是定时器：设置 20 ms 不会替你创建 50 Hz 的 publisher。它也不是执行期限：收到 deadline missed 事件时，并不会自动停止机器人。

2.5 Lifespan：这份消息还值不值得交付

Lifespan 关注一条样本从发布起的有效寿命。过期样本可能被中间件丢弃，而不是晚到后仍交给消费者。

这与消息 header.stamp 不同：相机图像可能采集后 150 ms 才调用 publish()。即使发布后的 lifespan 没超时，图像也早已不符合“采样到执行不超过 100 ms”的业务要求。

因此对新鲜度敏感的应用通常还需要检查：

• 采样或指令生成时间，而不仅是 publish 时间。
• 时间戳使用的时钟是否一致。
• 应用自身缓存有没有再次把消息放旧。

订阅回调已经取出的消息也不会因为 lifespan 到期就从你的应用变量里自动消失。

2.6 Liveliness 与 Lease Duration：发布实体是否仍活跃

AUTOMATIC 由中间件管理活跃性；MANUAL_BY_TOPIC 用于需要应用显式维护话题活跃性的场景，例如通过 publisher API 断言活跃。Lease Duration 是允许多长时间没有满足活跃性条件。

“活着”和“产生有效结果”不是一回事。进程的通信线程仍然工作时，算法线程可能已经卡住；反过来，一个低频发布配置的节点也未必应该被判死。

可以用几类问题区分这几种机制：

2.7 System Default：把决定交给当前实现

SYSTEM_DEFAULT 不等于“永远 reliable、深度永远 10”。它把相关决策留给 RMW/DDS 实现或供应商配置。时长的 DEFAULT 也不是“0 秒立即过期”；它表示未明确指定，由实现解释。

可复现实验应记录实际端点 QoS、ROS/RMW 版本及供应商配置，不只记录源码中的 preset 名称。遇到 UNKNOWN 或默认值时，也不要把静态兼容性检查返回的 warning 当成“肯定兼容”。Humble RMW profile 与兼容性检查接口

3. 发布端与订阅端怎样配对

3.1 Request vs Offered，不要求逐项相同

订阅端提出需求，发布端提供能力。影响匹配的策略需要同时满足要求。

最常见的问题是传感器驱动提供 best effort，而自己使用默认 reliable 订阅。话题和消息类型都对，数据依然不会到这个订阅者。

持久性则要连“是否补历史”一起判断：

有限 deadline 和 lease duration 的匹配方向都是“提供的间隔不大于请求允许的间隔”。发布端默认 deadline 配上订阅端明确的有限 deadline，不要想当然地认为能匹配。Liveliness 中，automatic 发布端也不能满足 manual-by-topic 的请求。官方 Humble 兼容性表

3.2 两个容易过度推广的结论

误区一：双方 depth 不一样就无法通信。

深度主要影响各自缓存，并不要求双方写同一个数字。发布端深度 10、订阅端深度 1，可以是有意设计的“发送端保留一定历史，接收端只缓存最新”。

误区二：一个话题已经有订阅者收到数据，说明所有订阅者都能收到。

每对 publisher/subscription 分别匹配。例如同一个 reliable publisher，可以连上一个 best-effort 监视器和一个 reliable 记录器；但将 publisher 改为 best effort 后，记录器可能失配，监视器仍正常。

此外，“兼容”只表示允许建立通信，不保证网络可达、类型正确、Executor 被调度或者回调逻辑没有异常。

4. Humble 内置 Profile 的准确默认值

下面是官方 Humble RMW 定义中的关键字段，不是所有 ROS 版本的永久承诺：

这些普通 profile 的 deadline、lifespan、lease duration 使用默认时长值，liveliness 使用 system default。注意三个修正：SensorData 不是 depth 1，普通服务不是 depth 1，Parameter Events 也不是 KEEP_ALL 或默认持久化快照。Humble qos_profiles.h 定义

如果要“当前有哪些参数”，应查询参数服务，而不是启动后只监听 /parameter_events 并假定已经得到了所有历史。

4.1 QoS(10) 与 SensorDataQoS() 不是一回事

auto default_qos = rclcpp::QoS(10);
// KEEP_LAST(10), RELIABLE, VOLATILE

auto sensor_qos = rclcpp::SensorDataQoS();
// KEEP_LAST(5), BEST_EFFORT, VOLATILE

auto latest_image_qos = rclcpp::SensorDataQoS();
latest_image_qos.keep_last(1);
// 这是自己修改后的 profile，不再是原封不动的 SensorData 默认值。

rclcpp::QoS 的构造器、keep_last()、best_effort()、transient_local() 等链式配置可以在 Humble API 中核对。C++ 的 volatile 是关键字，因此方法叫 durability_volatile()。rclcpp Humble QoS API

4.2 Action 不是“一个 Reliable + Depth 10 的话题”

Action 包含 goal、cancel、result 三类服务，以及 feedback、status 两类话题。它们承担的角色不同，因此不能用一行 QoS 描述整个 Action。

例如目标已经接受，不代表结果已经成功；反馈暂时缺失，也不能简单等同于目标被取消。排查 Action 应分别看服务调用、反馈和状态端点，再检查客户端/服务端生命周期。Humble rcl_action 客户端端点与默认选项

5. 从业务要求推导配置

下面都是评估起点，不是“这个话题必须永远如此”的规定。最终应与实际驱动、控制器接口契约和测量结果一致。

5.1 感知链路：允许丢帧，不等于允许瞎丢信息

检测器跳过几张图像可能只是帧率降低；IMU 预积分少一段采样，误差却可能被持续带入状态估计。应区分“业务允许少处理一些帧”和“算法假定每个样本都在”。

图像、点云还有一个容量问题：一帧未压缩 1920 × 1080 × 3 的图像约为 6.2 MB，30 Hz 仅像素载荷就约 187 MB/s。深度 10 的样本载荷就可能达到约 62 MB，还没计算不同端点的缓存、序列化和应用副本。这个估算说明为什么“先把所有队列调大”通常需要成本评估。

5.2 连续控制：新鲜的零速度，胜过过期的可靠速度

假设遥控端 20 Hz 发指令，业务要求超过 250 ms 没有有效新指令就进入安全策略。一个合理的设计讨论应包括：

1. VOLATILE，避免新启动消费者补收旧动作意图。
2. 小深度，让待处理的设定值不过度堆积。
3. 若带时间戳，检查生成时间、允许年龄和时钟关系。
4. 使用本机单调时钟记录“最后一次接受有效指令”的时间。
5. Watchdog 超时后执行明确的停止/保持/降级状态，而不是继续沿用旧指令。
6. 收到新包不一定刷新 watchdog：NaN、越界、过期、来源非法的指令应拒绝。
7. 底层驱动器还应有自己的通信超时和急停机制。

这里的 20 Hz、250 ms 都是教学假设，不能直接作为某台机械臂、无人机或移动机器人的安全阈值。飞行器失联时尤其不能把“所有输出归零”当成普适安全策略。

5.3 QoS 与控制循环不要揉成一个概念

在典型 ros2_control 系统中，订阅回调接收目标，控制循环读取状态、执行 controller update、写硬件命令。QoS 是目标如何进入系统的一部分，不等于每个电机都必须通过 DDS 跑一次 PID。

多线程 Executor 也不能替代 QoS：它可能改善回调互相阻塞的问题，但不能让不兼容的端点互通，也不能让已经过期的消息重新有效。完整分工见控制器与硬件接口专题。

6. C++：显式订阅配置与 QoS 事件

下面是一个独立节点源文件，订阅 /qos_demo/imu，检查端点 deadline，并注册不兼容事件。它用于通信诊断，不是实时控制回路，日志本身也会带来开销。

#include <memory>

#include "rclcpp/rclcpp.hpp"
#include "rclcpp/qos_event.hpp"
#include "sensor_msgs/msg/imu.hpp"

class ImuQosObserver final : public rclcpp::Node
{
public:
  ImuQosObserver() : Node("imu_qos_observer")
  {
    auto qos = rclcpp::SensorDataQoS();
    qos.keep_last(5);
    qos.deadline(rclcpp::Duration::from_seconds(0.1));

    rclcpp::SubscriptionOptions options;
    options.event_callbacks.deadline_callback =
      [this](rclcpp::QOSDeadlineRequestedInfo & event) {
        RCLCPP_WARN(
          get_logger(), "deadline missed: total=%d, change=%d",
          event.total_count, event.total_count_change);
      };
    options.event_callbacks.incompatible_qos_callback =
      [this](rclcpp::QOSRequestedIncompatibleQoSInfo & event) {
        RCLCPP_ERROR(
          get_logger(), "incompatible QoS: policy=%d, total=%d",
          static_cast<int>(event.last_policy_kind), event.total_count);
      };

    subscription_ = create_subscription<sensor_msgs::msg::Imu>(
      "/qos_demo/imu", qos,
      [this](sensor_msgs::msg::Imu::ConstSharedPtr msg) {
        RCLCPP_INFO(
          get_logger(), "received IMU, frame=%s", msg->header.frame_id.c_str());
      },
      options);
  }

private:
  rclcpp::Subscription<sensor_msgs::msg::Imu>::SharedPtr subscription_;
};

int main(int argc, char ** argv)
{
  rclcpp::init(argc, argv);
  rclcpp::spin(std::make_shared<ImuQosObserver>());
  rclcpp::shutdown();
  return 0;
}

保存为已有 ament_cmake 包的 src/imu_qos_observer.cpp，在 CMakeLists.txt 的 ament_package() 之前添加：

find_package(rclcpp REQUIRED)
find_package(sensor_msgs REQUIRED)

add_executable(imu_qos_observer src/imu_qos_observer.cpp)
target_compile_features(imu_qos_observer PUBLIC cxx_std_17)
ament_target_dependencies(imu_qos_observer rclcpp sensor_msgs)
install(TARGETS imu_qos_observer DESTINATION lib/${PROJECT_NAME})

package.xml 同时声明以下依赖：

<depend>rclcpp</depend>
<depend>sensor_msgs</depend>

假设包名是 qos_examples，在 ROS 容器的工作空间构建：

source /opt/ros/humble/setup.bash
colcon build --packages-select qos_examples
source install/setup.bash
ros2 run qos_examples imu_qos_observer

这里订阅端要求 100 ms deadline，因此测试发布端也必须提供兼容的有限 deadline；仅仅用一个默认 QoS、实际上 20 Hz 的 publisher，并不能替代这个声明。下一节的 Python publisher 可以作为对端。

事件 API 及异常类型来自 Humble rclcpp QoS events。不同 RMW 对事件的支持可能不同；显式注册不支持的事件可能抛出 UnsupportedEventTypeException。实际项目应在初始化阶段报告能力缺失，不能静默声称监控已经生效。

7. Python：把 Deadline 与应用 Watchdog 放在一起观察

下面一个文件有 pub、sub 两种角色。发布端以 20 Hz 发送用于演示的 IMU 消息；订阅端同时观察 DDS deadline 和本机接收间隔。

注意这是合成消息：orientation_covariance[0] = -1.0 表示没有姿态估计，不应把它作为真实 IMU 数据送进机器人估计器。

import time

import rclpy
from rclpy.duration import Duration
from rclpy.node import Node
from rclpy.qos import (
    QoSDurabilityPolicy,
    QoSHistoryPolicy,
    QoSProfile,
    QoSReliabilityPolicy,
)
from rclpy.qos_event import SubscriptionEventCallbacks
from sensor_msgs.msg import Imu


def make_qos():
    return QoSProfile(
        history=QoSHistoryPolicy.KEEP_LAST,
        depth=5,
        reliability=QoSReliabilityPolicy.BEST_EFFORT,
        durability=QoSDurabilityPolicy.VOLATILE,
        deadline=Duration(seconds=0.1),
    )


class QosDemo(Node):
    def __init__(self):
        super().__init__('qos_demo')
        role = self.declare_parameter('role', 'sub').value
        self.last_receive = None
        self.was_stale = None

        if role == 'pub':
            self.publisher = self.create_publisher(
                Imu, '/qos_demo/imu', make_qos())
            self.timer = self.create_timer(0.05, self.publish_sample)
        elif role == 'sub':
            events = SubscriptionEventCallbacks(
                deadline=self.on_deadline,
                incompatible_qos=self.on_incompatible,
            )
            self.subscription = self.create_subscription(
                Imu, '/qos_demo/imu', self.on_message, make_qos(),
                event_callbacks=events,
            )
            self.timer = self.create_timer(0.05, self.check_watchdog)
        else:
            raise ValueError('role must be pub or sub')

    def publish_sample(self):
        msg = Imu()
        msg.header.stamp = self.get_clock().now().to_msg()
        msg.header.frame_id = 'demo_imu'
        msg.orientation_covariance[0] = -1.0
        self.publisher.publish(msg)

    def on_message(self, msg):
        # 单调时钟用于本机“多久没收到”，不跨设备比较。
        self.last_receive = time.monotonic()

    def on_deadline(self, event):
        self.get_logger().warning(
            f'DDS deadline missed: total={event.total_count}')

    def on_incompatible(self, event):
        self.get_logger().error(
            f'incompatible QoS: policy={event.last_policy_kind}')

    def check_watchdog(self):
        stale = (
            self.last_receive is None
            or time.monotonic() - self.last_receive > 0.2
        )
        if stale != self.was_stale:
            self.get_logger().info(f'application watchdog stale={stale}')
            self.was_stale = stale


def main():
    rclpy.init()
    node = QosDemo()
    try:
        rclpy.spin(node)
    except KeyboardInterrupt:
        pass
    finally:
        node.destroy_node()
        rclpy.try_shutdown()


if __name__ == '__main__':
    main()

保存为 qos_demo.py，在同一 ROS 环境的两个终端分别运行：

# 终端 A：发布端
python3 qos_demo.py --ros-args -r __node:=qos_demo_pub -p role:=pub

# 终端 B：订阅端
python3 qos_demo.py --ros-args -r __node:=qos_demo_sub -p role:=sub

预期先看到订阅端从 stale 转为非 stale。停止 A 后，B 的接收 watchdog 应转回 stale；DDS deadline 事件的观察结果还要结合具体 RMW 实现和端点消失时间判断，不把它当成唯一断线探测。

Python 的配置字段和事件名称分别可核对 rclpy Humble QoS API 与 rclpy Humble QoS events。显式注册不支持的事件可能抛出 UnsupportedEventTypeError，应记录 RMW 能力后再调整监控方案。

7.1 这个 Watchdog 为什么还不能用于真机安全

演示只验证“回调收到数据”的间隔，没有验证消息内容和采样年龄。而且 timer 与 subscription 仍在同一 Executor 中：如果执行线程卡住，watchdog 回调也可能一起卡住。

示例的 create_timer() 默认还使用节点时钟。若开启 use_sim_time 并暂停 /clock，定时器可能不再触发，即使回调内部用的是 monotonic，检查也无从执行。需要按真实时间运行的监控，应另外选择合适的定时器时钟及执行线程，并单独验证暂停和时钟跳变行为。

真正的控制系统要定义哪些检查进入独立的控制循环，哪些由驱动器或硬件承担。使用多线程也必须考虑 callback group、锁竞争和调度延迟，不能因为换了 Executor 就声称已经具有硬实时保护。

7.2 跨机器算消息年龄要先解决时钟

“当前 ROS 时间 - header.stamp”只有在时间域和同步关系成立时才有意义。应分别考虑：

• 两台实体计算机的时钟同步误差。
• use_sim_time 是否一致，/clock 是否在推进。
• rosbag 回放是否循环、暂停或改变时间。
• 相机硬件时间戳有没有正确映射到 ROS 时间。
• 零时间戳、未来时间戳和时钟跳变如何处理。

本机 time.monotonic() 适合判断本机接收间隔，不应写进消息后直接和另一台机器的 monotonic 值相减。

8. 三个最值得亲手做的 CLI 实验

以下命令应在已经配置好的 Humble ROS 环境里运行，不需要接真机。各终端 source /opt/ros/humble/setup.bash，并使用相同的 ROS domain。若在容器内做实验，尽量使用同一容器的多个终端，先排除 Docker 跨容器发现问题。

CLI 默认值不一定等于 C++/Python API 默认值；Humble 的 topic echo 还可能根据发现的发布者选择可靠性和持久性。因此实验显式写出这些选项，避免工具自动适配把问题掩盖。Humble topic pub 实现、Humble topic echo 实现

8.1 实验 A：制造可靠性不匹配

终端 A 持续发布：

ros2 topic pub /qos_demo/text std_msgs/msg/String '{data: "sample"}' \
  --rate 2 --qos-reliability best_effort \
  --qos-durability volatile --qos-history keep_last --qos-depth 5

终端 B 故意要求 reliable：

ros2 topic echo /qos_demo/text std_msgs/msg/String \
  --qos-reliability reliable --qos-durability volatile

预期：B 无法从这个发布端收到 sample，支持相应事件时还会看到不兼容提示。终端 C 查看证据：

ros2 topic info /qos_demo/text --verbose

停止 B，再以 best effort 重启：

ros2 topic echo /qos_demo/text std_msgs/msg/String \
  --qos-reliability best_effort --qos-durability volatile

预期：回调现在能够收到数据。这个实验验证的是可靠性匹配，不是网络丢包率。若第二步仍不通，应先检查消息类型、domain、发现和网络，不要不断盲调 depth。

8.2 实验 B：晚加入读者取得快照

终端 A 只发一次，但让发布进程继续存活 60 秒：

ros2 topic pub /qos_demo/snapshot std_msgs/msg/String '{data: "map-v1"}' \
  --once --wait-matching-subscriptions 0 --keep-alive 60 \
  --qos-reliability reliable --qos-durability transient_local \
  --qos-history keep_last --qos-depth 1

等 A 显示已经发布，再启动 B；要在 A 的 60 秒保活期间完成：

ros2 topic echo /qos_demo/snapshot std_msgs/msg/String --once \
  --qos-reliability reliable --qos-durability transient_local \
  --qos-history keep_last --qos-depth 1

预期：B 取得缓存的 map-v1。改用 volatile 订阅重新做一遍，则不应依赖收到这条旧消息，因为 A 已经没有继续发布。

这里两个 CLI 参数很关键：–wait-matching-subscriptions 0 让发布发生在没有订阅者时；–keep-alive 60 防止发布端立刻退出丢掉本次实验所需的本地历史。这两个选项在所核对的 Humble pub.py 中存在；若安装包更旧，先看 ros2 topic pub --help 再调整实验。

8.3 实验 C：同一话题的两个读者，结果可以不同

保留实验 A 的 best-effort 发布端，同时启动两个 echo，一个 reliable、一个 best effort。再执行 topic info --verbose。

需要记录的不是“这个话题通/不通”，而是：

• 哪个 publisher 的端点标识与实际 QoS？
• 哪个 subscription 的 QoS？
• 哪一对成功收到消息？
• 失配日志对应哪个策略？

这能建立最重要的排障习惯：按端点配对分析，不按话题名一刀切。

9. rosbag2：录制和回放也是通信端点

录制时，rosbag2 是订阅者；回放时，rosbag2 变成发布者。它会尝试适配 QoS，但混合发布端、晚加入快照和版本差异仍可能需要显式覆盖。

9.1 为录制指定 QoS

例如写一份 record_qos.yaml：

/camera/image_raw:
  history: keep_last
  depth: 10
  reliability: best_effort
  durability: volatile

/map:
  history: keep_last
  depth: 1
  reliability: reliable
  durability: transient_local

然后录制：

ros2 bag record /camera/image_raw /map \
  -o qos_record_001 --qos-profile-overrides-path record_qos.yaml

qos_record_001 应使用新的输出目录名称，避免把不同实验混成一份记录。录制 /map 前先确认它确实由兼容的 transient-local publisher 提供；不要把上面 YAML 当作任何地图节点的通用配置。

9.2 回放侧按消费者需求配置

假设离线消费者明确请求 reliable，可用独立的 play_qos.yaml：

/camera/image_raw:
  history: keep_last
  depth: 10
  reliability: reliable
  durability: volatile

/map:
  history: keep_last
  depth: 1
  reliability: reliable
  durability: transient_local

ros2 bag info qos_record_001
ros2 bag play qos_record_001 --qos-profile-overrides-path play_qos.yaml

回放改成 reliable，不能恢复录制阶段从未写入 bag 的消息。它只改变“已有 bag 数据如何提供给新的消费者”。同样，录制端有消息也不表示全部消息已完整写入：还应检查计数、持续时间、应用序号、磁盘吞吐和丢失报告。

覆盖文件的字段名、时长格式和参数可核对 Humble rosbag2 QoS override 指南。不要通过覆盖去伪装比实际回放节奏更严格的 deadline。

10. 图像同步、TF 和参数的常见边界

10.1 message_filters 没有触发，先分清两个队列

同步器前面还有 ROS subscription。第一步必须确保每个输入能被自己的订阅者收到，然后才讨论同步规则。

图像 Publisher → 图像 Subscription 的 QoS 缓存 ─┐
                                               ├→ 同步器自己的队列 → 同步回调
IMU Publisher → IMU Subscription 的 QoS 缓存 ───┘

给同步器设 queue_size=30，不等于它前面的 DDS subscription 就自动有深度 30，更不能修复 best-effort publisher 与 reliable subscriber 的不匹配。

Humble 的 C++ message_filters::Subscriber 可以显式传入 rmw_qos_profile_t。假设 image_sub_ 是相应成员，可以这样订阅传感器话题：

image_sub_.subscribe(this, "/camera/image_raw", rmw_qos_profile_sensor_data);

Python 示例中，也应显式传 qos_profile：

import message_filters
from rclpy.qos import qos_profile_sensor_data
from sensor_msgs.msg import Image

# node 是已创建并保留生命周期的 rclpy.Node。
image_sub = message_filters.Subscriber(
    node, Image, "/camera/image_raw",
    qos_profile=qos_profile_sensor_data,
)

不要照抄前文用于 deadline 实验的 make_qos() 去订阅默认相机驱动：那个 profile 要求 100 ms deadline；这里使用没有额外有限 deadline 要求的 sensor_data profile。

进一步再查：时间戳是否同一时间域、帧是否能配对、ApproximateTime 的时间容差是否合理、输入频率是否悬殊。每条输入与它自己的发布端需要兼容，不是所有不同话题必须逐字段使用同一套 QoS。Humble C++ Subscriber API、Humble Python message_filters 实现

10.2 /tf 与 /tf_static 不能合并理解

动态 TF 描述随时间变化的位姿；静态 TF 描述固定关系，通常需要支持晚加入读者。应优先使用 tf2_ros 官方 broadcaster/listener，保留它们为不同端点设计的 QoS。

不要因为“静态 TF 用 transient local”，就把动态 /tf 也机械地改成同样的深度和持久性；更不要把多个独立变换当成一个 String 快照简单覆盖。TF 消息内容、发布端的变换集合维护以及消费者的时间缓存，还各有语义。Humble TF QoS 定义、静态 TF broadcaster 接口

10.3 YAML 不是所有节点都自动接受的万能覆盖

rosbag2 的 override 是 rosbag2 明确提供的功能。第三方驱动是否允许通过参数改 QoS、参数叫什么、何时生效，应看它自己的接口文档和实际端点。

因此修改配置后要重新运行 ros2 topic info --verbose 检查，而不是因为 YAML 文件里出现了 reliable 就认为 DDS 端点已经改变。某些配置只在实体创建时读取，需要重建节点或重新配置生命周期；不要假定运行时 ros2 param set 总能修改已存在的 publisher。

11. 按证据排障，不靠随机切换 Reliable

11.1 第一层：实体能否互相发现

printenv ROS_DISTRO RMW_IMPLEMENTATION ROS_DOMAIN_ID ROS_LOCALHOST_ONLY
ros2 doctor --report
ros2 node list
ros2 topic list -t
ros2 topic info /目标话题 --verbose

环境变量为空不代表没有使用默认值；要结合发行版、启动环境和节点报告解释结果。检查 namespace、remap、消息类型和多个同名节点，再看容器网络、DDS 发现配置、防火墙、多播等问题。

如果根本发现不到端点，先查发现与网络，不要从 deadline 开始改。

11.2 第二层：逐对比较实际 QoS

重点抄下：reliability、durability、history/depth、deadline、liveliness、lease duration。一个话题可能有多个 publisher；不要只看第一段输出就结束。

建立兼容关系后，再检查端点连接数量、incompatible QoS 事件和显式 profile 的 echo 结果。

11.3 第三层：消息是否新鲜、回调是否被调度

ros2 topic hz 测到的是该工具观察到的到达频率，不自动等于传感器真实采样频率，也不表示用户自己的回调具有同样表现。终端打印大图像/点云还会干扰观察，应尽量使用计数和统计而不是逐字段刷屏。Humble topic hz 实现与说明

11.4 一份有价值的实验记录

以后遇到问题，笔记至少保留：

环境：镜像 ID / 架构 / ROS 与包版本 / RMW / DDS 配置
拓扑：同机还是跨机、是否 Docker、网络路径、domain
端点：完整话题名 / 类型 / publisher 与 subscriber 的实际 QoS
负载：频率 / 单条字节数 / 回调耗时 / 并发订阅者
时间：采样年龄 / 接收间隔 / Executor 等待 / 时钟同步方式
故障：拔网、停进程、暂停发布、慢消费者中的哪一种
证据：日志 / bag / 计数 / 丢样与恢复时间
结论：只改了什么、改善了什么、还有什么没验证

特别是进行网络故障实验时，应在隔离仿真环境或无执行器的测试节点上完成，不要为了观察 QoS 直接断开运行中真机的控制网络。

12. 选型检查单与复习题

12.1 上线前的 QoS 检查单

• [ ] 区分连续设定值、一次性任务、状态流和完整快照。
• [ ] 明确数据能丢多少、能旧多久、晚加入者需要什么。
• [ ] 记录每个关键端点的实际 profile，不仅记录 preset 名称。
• [ ] 验证可靠性、持久性、deadline 和 liveliness 相关兼容性。
• [ ] 估算缓存字节数，测量处理吞吐，不把大 depth 当作性能优化。
• [ ] 检查采样时间域、跨机同步和回放时钟。
• [ ] 将通信活跃、应用响应和硬件执行成功分开监测。
• [ ] 对控制指令定义拒绝条件、有效期、失联状态和恢复条件。
• [ ] 验证 rosbag 录制与回放，记录丢样和端点配置。
• [ ] 验证所选 RMW 的事件能力；没有事件不等于没有故障。
• [ ] 在目标架构和真实负载下重新测量，而不是只在空载 echo 中验收。

12.2 用自己的话回答这六个问题

1. 为什么 SensorDataQoS() 发布的图像，QoS(10) 订阅可能收不到？
2. 为什么地图 publisher 是 transient local，但 volatile 订阅仍不应依赖补到旧地图？
3. 发布端 deadline 10 ms、订阅端 deadline 30 ms 是否匹配？反过来呢？
4. 数据 100 Hz 到达、算法只能 20 Hz 处理，把 depth 从 5 调到 100 解决了什么，又引入了什么？
5. Deadline、Liveliness 和本机接收 watchdog，哪一个能够直接证明电机安全？
6. 为什么把 rosbag 回放改成 reliable，不能补回原录制阶段丢失的数据？

对应的核心答案是：端点需求高于供给、历史请求不同、提供间隔必须足够短、缓存不增加吞吐、三者都不能独自证明硬件安全，以及回放只能提供已经记录的数据。

13. 版本与验证边界

本文保留原文章标题、发布日期和链接，在 2026-09-06 按 Humble 主源扩充。示例里的 QoS API、CLI 选项和默认值都应与本机安装版本再核对；不要把 Rolling/Jazzy 的新参数直接套在 Humble 上。

目前完成的是文档和源码层核对，而不是 ROS 容器、无线网络或真实控制器的运行验收。后续在统一 Docker 工作流中复现实验时，应把“接口可用”“端点匹配”“负载下延迟”“故障后安全响应”分别列为验收项，而不是只写一个“QoS 测试通过”。