DDS的全称是Data Distribution Service，也就是数据分发服务，是一套专门为实时系统设计的数据分发/订阅标准。DDS强调以数据为中心，可以提供丰富的服务质量策略，以保障数据进行实时、高效、灵活地分发，可满足各种分布式实时通信应用需求。话题、服务、动作的底层通信具体实现过程都是靠DDS来完成的，DDS的核心是通信，能够实现通信的模型和软件框架非常多，以下列出常用的四种模型。

• 第一种，点对点模型，许多客户端连接到一个服务端，每次通信时，通信双方必须建立一条连接。当通信节点增多时，连接数也会增多。而且每个客户端都需要知道服务器的具体地址和所提供的服务，一旦服务器地址发生变化，所有客户端都会受到影响。

• 第二种，Broker模型，针对点对点模型进行了优化，由Broker集中处理所有人的请求，并进一步找到真正能响应该服务的角色。这样客户端就不用关心服务器的具体地址了。不过问题也很明显，Broker作为核心，它的处理速度会影响所有节点的效率，当系统规模增长到一定程度，Broker就会成为整个系统的性能瓶颈。更麻烦是，如果Broker发生异常，可能导致整个系统都无法正常运转。之前的ROS1系统，使用的就是类似这样的架构。

• 第三种，广播模型，所有节点都可以在通道上广播消息，并且节点都可以收到消息。这个模型解决了服务器地址的问题，而且通信双方也不用单独建立连接，但是广播通道上的消息太多了，所有节点都必须关心每条消息，其实很多是和自己没有关系的。

• 第四种，以数据为中心的DDS模型，这种模型与广播模型有些类似，所有节点都可以在DataBus上发布和订阅消息。但它的先进之处在于，通信中包含了很多并行的通路，每个节点可以只关心自己感兴趣的消息，忽略不感兴趣的消息，有点像是一个旋转火锅，各种好吃的都在这个DataBus传送，我们只需要拿自己想吃的就行，其他的和我们没有关系。

可见，在这几种通信模型中，DDS的优势更加明显。

        QoS是一种网络传输策略，应用程序指定所需要的网络传输质量行为，QoS服务实现这种行为要求，尽可能地满足客户对通信质量的需求，可以理解为数据提供者和接收者之间的合约。ROS 2 引入了 DDS 规范中的服务质量策略（Quality of Service, QoS），用来精准控制节点间数据传输的可靠性、延迟和实时性。在无线网络波动、传感器高频发包或系统资源紧张时，合理配置 QoS 是保证系统稳定运行的关键。

DDS中的基本结构是Domain，Domain将各个应用程序绑定在一起进行通信，只有处于同一个DOMAIN小组中的节点才能互相通信。这样可以避免无用数据占用的资源。

核心 QoS 策略维度

ROS 2 的 QoS 主要由以下核心策略（Policy）组合而成：

策略名称	可选参数	含义与效果	典型应用场景
Reliability (可靠性)	Reliable	保证数据不丢包。若接收方丢包会触发重发，增加延迟。	关键控制指令、TF 坐标变换、Service
	Best Effort	尽力传输。不保证到达，丢包不重发，延迟最低。	高频数据（相机图像、LiDAR 点云、IMU）
Durability (持久性)	Transient Local	发布者会为后加入系统的订阅者保留最新若干条历史数据。	地图数据（/map）、静态外参配置文件
	Volatile	不保留历史。新订阅者只能接收加入后发布的新数据。	实时数据流（传感器数据、里程计）
History (历史保留)	Keep Last	配合 Depth （队列深度）使用，队列溢出时丢弃最旧数据，仅保留最新的 N 条。	大多数实时话题
	Keep All	保留所有未发送/未处理数据，受系统内存限制。	绝对不能漏掉的事件日志
Deadline (截止期限)	时长 (Duration)	规定两次发布/接收数据的最大时间间隔。超期触发回调。	关键心跳监测、传感器超时检测
Liveliness (存活性)	Automatic / Manual	确定发布者是否仍处于活跃状态（Alive）。	节点故障监控与自动降级

QoS 匹配机制（Rx/Tx Compatibility）

ROS 2 的发布者（Publisher）与订阅者（Subscriber）建立通信前，DDS 会先校验双方的 QoS 是否兼容。

匹配的核心规则是 “请求的要求不能高于提供者的承诺”：

• Reliability：Pub 必须比 Sub 更可靠。如果 Sub 设置为 Reliable，但 Pub 设置为 Best Effort，通信无法建立（Publisher 无法满足 Subscriber 的可靠性要求）。

• Durability：Pub 的持久度必须 大于等于 Sub。如果 Sub 设置为 Transient Local，Pub 设置为 Volatile，通信无法建立。

常用的预设 QoS 配置文件（QoS Profiles）

为了避免每次手动组合这六项参数，ROS 2 提供了几种预置配置：

• rclcpp::SensorDataQoS()：

  • Reliability: Best Effort

  • Durability: Volatile

  • History: Keep Last (Depth=5)

  • 适合高频传感器（相机、雷达、IMU），丢一两帧无所谓，要求低延迟。

• rclcpp::ServicesQoS()：

  • Reliability: Reliable

  • Durability: Volatile

  • 适用于 Service 请求/响应通信，要求 100% 送达。

• rclcpp::ParametersQoS()：

  • Reliability: Reliable

  • Durability: Transient Local

  • 用于参数通知，确保新加入节点能收到当前参数值。

发布者代码解析

import rclpy                     # ROS2 Python接口库
from rclpy.node import Node      # ROS2 节点类
from std_msgs.msg import String  # 字符串消息类型
from rclpy.qos import QoSProfile, QoSReliabilityPolicy, QoSHistoryPolicy # ROS2 QoS类

"""
创建一个发布者节点
"""
class PublisherNode(Node):

    def __init__(self, name):
        super().__init__(name)        # ROS2节点父类初始化

        qos_profile = QoSProfile(     # 创建一个QoS原则
            # reliability=QoSReliabilityPolicy.BEST_EFFORT,
            reliability=QoSReliabilityPolicy.RELIABLE,
            history=QoSHistoryPolicy.KEEP_LAST,
            depth=1
        )
        self.pub = self.create_publisher(String, "chatter", qos_profile) # 创建发布者对象（消息类型、话题名、QoS原则）
        self.timer = self.create_timer(0.5, self.timer_callback)         # 创建一个定时器（单位为秒的周期，定时执行的回调函数）

    def timer_callback(self):                                # 创建定时器周期执行的回调函数
        msg = String()                                       # 创建一个String类型的消息对象
        msg.data = 'Hello World'                             # 填充消息对象中的消息数据
        self.pub.publish(msg)                                # 发布话题消息
        self.get_logger().info('Publishing: "%s"' % msg.data)# 输出日志信息，提示已经完成话题发布

def main(args=None):                           # ROS2节点主入口main函数
    rclpy.init(args=args)                      # ROS2 Python接口初始化
    node = PublisherNode("qos_helloworld_pub") # 创建ROS2节点对象并进行初始化
    rclpy.spin(node)                           # 循环等待ROS2退出
    node.destroy_node()                        # 销毁节点对象
    rclpy.shutdown()                           # 关闭ROS2 Python接口

订阅者代码解析

import rclpy                                     # ROS2 Python接口库
from rclpy.node   import Node                    # ROS2 节点类
from std_msgs.msg import String                  # ROS2标准定义的String消息
from rclpy.qos import QoSProfile, QoSReliabilityPolicy, QoSHistoryPolicy  # ROS2 QoS类

"""
创建一个订阅者节点
"""
class SubscriberNode(Node):

    def __init__(self, name):
        super().__init__(name)         # ROS2节点父类初始化

        qos_profile = QoSProfile(      # 创建一个QoS原则
            # reliability=QoSReliabilityPolicy.BEST_EFFORT,
            reliability=QoSReliabilityPolicy.RELIABLE,
            history=QoSHistoryPolicy.KEEP_LAST,
            depth=1
        )

        self.sub = self.create_subscription(\
            String, "chatter", self.listener_callback, qos_profile) # 创建订阅者对象（消息类型、话题名、订阅者回调函数、QoS原则）

    def listener_callback(self, msg):                      # 创建回调函数，执行收到话题消息后对数据的处理
        self.get_logger().info('I heard: "%s"' % msg.data) # 输出日志信息，提示订阅收到的话题消息

def main(args=None):                               # ROS2节点主入口main函数
    rclpy.init(args=args)                          # ROS2 Python接口初始化
    node = SubscriberNode("qos_helloworld_sub")    # 创建ROS2节点对象并进行初始化
    rclpy.spin(node)                               # 循环等待ROS2退出
    node.destroy_node()                            # 销毁节点对象
    rclpy.shutdown()                               # 关闭ROS2 Python接口

Liveliness（存活性）QoS 策略：监测发布者（Publisher）是否还“活着”（活跃），以及在发布者掉线/卡死时及时通知订阅者（Subscriber）。

#include <rclcpp/rclcpp.hpp>
#include <std_msgs/msg/string.hpp>

class HealthMonitorNode : public rclcpp::Node {
public:
  HealthMonitorNode() : Node("health_monitor_node") {
    // 1. 配置 Liveliness QoS：要求 Publisher 每 1 秒必须报一次平安
    rclcpp::QoS qos(10);
    qos.liveliness(rclcpp::LivelinessPolicy::ManualByTopic);
    qos.liveliness_lease_duration(std::chrono::seconds(1));

    // 2. 配置订阅选项与回调函数
    rclcpp::SubscriptionOptions options;
    options.event_callbacks.liveliness_callback =
      [](rclcpp::QoSLivelinessChangedInfo & info) {
        if (info.alive_count_change < 0) {
          RCLCPP_ERROR(rclcpp::get_logger("Monitor"), "警告：发布者卡死或掉线！触发安全防御逻辑。");
        } else if (info.alive_count_change > 0) {
          RCLCPP_INFO(rclcpp::get_logger("Monitor"), "提示：发布者已恢复连接/上线。");
        }
      };

    // 3. 创建订阅者
    sub_ = this->create_subscription<std_msgs/msg::String>(
      "heartbeat", qos,
      [](std_msgs/msg::String::SharedPtr msg) { (void)msg; },
      options
    );
  }

private:
  rclcpp::Subscription<std_msgs/msg::String>::SharedPtr sub_;
};

options.event_callbacks.liveliness_callback 绑定了 Liveliness 状态变化事件。当发布者掉线或恢复时，DDS 会触发 options 里注册的回调函数，而不是触发第 3 个参数的数据回调。