ROS 2 DDS 中间件通信优化与 QoS 策略详解

2.3 安装其他 DDS 实现

# 安装 CycloneDDS
sudo apt install ros-humble-rmw-cyclonedds-cpp

# 安装 RTI Connext (需要许可证)
sudo apt install ros-humble-rmw-connext-cpp

# 验证安装
ros2 run your_package your_node --help | grep -i "implementation"

三、QoS（服务质量）策略详解

3.1 QoS 的核心概念

QoS 定义了消息传输的特性和保证：

QoS = { 可靠性，持久性，延迟，寿命，历史 }

3.2 关键 QoS 参数

3.2.1 可靠性（Reliability）

• RELIABLE (可靠传输):
  • 确认机制
  • 保证到达，会重传
  • 适用于控制指令
• BEST_EFFORT (尽力而为):
  • 快速发送
  • 高频数据，允许丢包
  • 适用于传感器

时间模型对比：

• BEST_EFFORT 模式：Msg 1 (丢失), Msg 2(无 ACK)
• RELIABLE 模式：Msg A ✅ ACK, Msg B (丢失) 🔄 Msg B (重传) ✅ ACK

3.2.2 持久性（Durability）

数据生命周期：

• VOLATILE 模式：消息只在发送瞬间存在，订阅者 B 此时加入 ❌ 无法获取之前的消息
• TRANSIENT_LOCAL 模式：💾 消息被缓存，订阅者 B 加入 📨 消息发送接收 📨 重发缓存的消息 ✅ 获取到历史消息

3.2.3 历史（History）

• KEEP_ALL: 保留所有 (Msg 1, Msg 2, Msg 3, Msg 4)
• KEEP_LAST (Depth=N): 只保留最后 N 条 (Msg 2, Msg 3, Msg 4)

3.2.4 截止期（Deadline）

有效时间 = T_deadline 如果消息未在截止期内到达，则认为超时。

• 正常传输：发送消息 -> Deadline -> 接收 (有效)
• 超时传输：发送消息 -> Deadline -> 接收 (超时)

3.2.5 生存时间（Lifespan）

消息的有效期：

• T=0s 消息生成 (内容"温度=25°C")
• T=0.5s 此时被接收 (有效)
• T=1s Lifespan 到期
• T=1.1s 此时被丢弃 (失效)

3.3 QoS 的数学模型

通信成功的概率模型： P_success = P_reliability × P_no_timeout × P_delivery

其中：

• P_reliability - 根据 Reliability 设置
• P_no_timeout - 消息在 Deadline 内到达的概率
• P_delivery - 根据 Durability 设置

四、在 Python 中配置 QoS

4.1 使用预定义的 QoS 配置文件

import rclpy
from rclpy.node import Node
from rclpy.qos import QoSProfile, QoSHistoryPolicy, QoSReliabilityPolicy, QoSDurabilityPolicy
from std_msgs.msg import String

class MyPublisher(Node):
    def __init__(self):
        super().__init__('my_publisher')
        # 方法 1：使用预定义配置（推荐）
        # 系统默认配置
        qos = QoSProfile(depth=10)
        # 传感器数据配置（低延迟，允许丢包）
        qos_sensor = rclpy.qos.sensor_data_qos_profile
        # 系统事件配置（可靠传输）
        qos_reliable = rclpy.qos.qos_profile_system_default
        
        # 创建发布者
        self.pub = self.create_publisher(String, 'topic_name', qos_sensor) # 使用传感器 QoS 配置
        self.get_logger().info('发布者已创建')

def main(args=None):
    rclpy.init(args=args)
    node = MyPublisher()
    rclpy.spin(node)
    rclpy.shutdown()

if __name__ == '__main__':
    main()

4.2 自定义 QoS 配置

from rclpy.qos import (
    QoSProfile, QoSReliabilityPolicy, QoSHistoryPolicy, QoSDurabilityPolicy
)
from rclpy.duration import Duration
from rclpy.node import Node
from std_msgs.msg import String

class QoSCustomizer(Node):
    def __init__(self):
        super().__init__('qos_customizer')
        # 自定义 QoS 配置
        custom_qos = QoSProfile(
            # 可靠性：RELIABLE = 可靠传输，BEST_EFFORT = 尽力而为
            reliability=QoSReliabilityPolicy.RELIABLE,
            # 持久性
            durability=QoSDurabilityPolicy.VOLATILE,
            # 历史：KEEP_LAST = 保留最后 N 条，KEEP_ALL = 保留所有
            history=QoSHistoryPolicy.KEEP_LAST,
            # 历史深度（KEEP_LAST 时的队列大小）
            depth=10,
            # 截止期（消息必须在此时间内到达）
            deadline=Duration(seconds=1),
            # 生存时间（消息的有效期）
            lifespan=Duration(seconds=10),
            # 策略中的缺失时间再发送间隔
            period=Duration(milliseconds=100)
        )
        # 使用自定义 QoS 创建发布者
        self.pub = self.create_publisher(String, 'custom_topic', custom_qos)
        # 使用自定义 QoS 创建订阅者
        self.sub = self.create_subscription(String, 'custom_topic', self.callback, custom_qos)
        self.get_logger().info('自定义 QoS 配置已应用')

    def callback(self, msg):
        self.get_logger().info(f'接收：{msg.data}')

def main(args=None):
    rclpy.init(args=args)
    node = QoSCustomizer()
    rclpy.spin(node)
    rclpy.shutdown()

if __name__ == '__main__':
    main()

4.3 QoS 预定义配置对比

# 不同 QoS 配置的对比
# 1. 传感器数据（高频，允许丢包）
sensor_qos = rclpy.qos.sensor_data_qos_profile
# 特点：BEST_EFFORT, VOLATILE, KEEP_LAST(depth=5)

# 2. 参数（可靠传输）
param_qos = rclpy.qos.qos_profile_parameter_events
# 特点：RELIABLE, TRANSIENT_LOCAL, KEEP_LAST(depth=1000)

# 3. 系统默认
default_qos = rclpy.qos.qos_profile_system_default
# 特点：RELIABLE, VOLATILE, KEEP_LAST(depth=10)

# 4. 服务
service_qos = rclpy.qos.qos_profile_services_default
# 特点：RELIABLE, VOLATILE, KEEP_LAST

五、在 C++ 中配置 QoS

5.1 C++ 版本的 QoS 配置

#include "rclcpp/rclcpp.hpp"
#include "std_msgs/msg/string.hpp"
#include "rclcpp/qos.hpp"

class QoSExample : public rclcpp::Node {
public:
    QoSExample() : Node("qos_example") {
        // 方法 1：使用预定义配置
        auto sensor_qos = rclcpp::SensorDataQoS();
        
        // 方法 2：自定义 QoS
        rclcpp::QoS custom_qos(10); // 队列大小为 10
        custom_qos.reliable(); // 启用可靠传输
        custom_qos.transient_local(); // 启用本地持久性
        
        // 创建发布者（使用自定义 QoS）
        pub_ = this->create_publisher<std_msgs::msg::String>("my_topic", custom_qos);
        
        // 创建订阅者
        sub_ = this->create_subscription<std_msgs::msg::String>("my_topic", sensor_qos,
            [this](const std_msgs::msg::String & msg){
                RCLCPP_INFO(this->get_logger(), "收到：%s", msg.data.c_str());
            });
        
        RCLCPP_INFO(this->get_logger(), "QoS 配置完成");
    }
private:
    rclcpp::Publisher<std_msgs::msg::String>::SharedPtr pub_;
    rclcpp::Subscription<std_msgs::msg::String>::SharedPtr sub_;
};

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

5.2 QoS 兼容性问题

发布者和订阅者的 QoS 必须兼容（即发布者提供的 QoS 必须满足或优于订阅者的要求）：

结论：

• 可靠性规则：Reliable 发布者兼容所有订阅者；Best Effort 发布者只能兼容 Best Effort 订阅者。
• 持久性规则：Transient Local 发布者兼容所有订阅者；Volatile 发布者只能兼容 Volatile 订阅者。

六、QoS 实战案例

6.1 高性能传感器数据传输

场景：激光雷达每秒发送 100 帧扫描数据

from rclcpp.qos import QoSProfile, QoSReliabilityPolicy, QoSHistoryPolicy

# 优化方案：优先保证吞吐量和低延迟
lidar_qos = QoSProfile(
    reliability=QoSReliabilityPolicy.BEST_EFFORT, # 允许丢包
    history=QoSHistoryPolicy.KEEP_LAST, # 只保留最新数据
    depth=1 # 队列只有 1 条
)

# 优势：
# 1. 消息处理快（BEST_EFFORT 无需重传等待）
# 2. 内存占用低（只保留最新数据）
# 3. 延迟最小（深度为 1）

6.2 关键命令传输

场景：发送"停止"命令给机器人

# 优化方案：保证命令一定被执行
command_qos = QoSProfile(
    reliability=QoSReliabilityPolicy.RELIABLE, # 可靠传输
    durability=QoSDurabilityPolicy.TRANSIENT_LOCAL, # 持久化
    history=QoSHistoryPolicy.KEEP_LAST, depth=5 # 保留最后 5 条
)

# 优势：
# 1. 消息不会丢失（RELIABLE）
# 2. 新启动的接收方能收到历史消息（TRANSIENT_LOCAL）
# 3. 关键命令不会被遗漏

6.3 定期状态报告

场景：每秒发送一次机器人状态

status_qos = QoSProfile(
    reliability=QoSReliabilityPolicy.BEST_EFFORT,
    history=QoSHistoryPolicy.KEEP_LAST,
    depth=3 # 保留最后 3 秒的数据
)

# 平衡方案：
# 1. 新数据优先于历史数据（BEST_EFFORT）
# 2. 缓冲 3 条消息防止一瞬间的网络波动
# 3. 内存占用合理

七、网络配置和优化

7.1 DDS 通信的传输层

7.2 配置 DDS 环境变量

# 启用 UDP 单播（默认）
export ROS_SUPER_CLIENT=1

# 启用多播（局域网推荐）
export CYCLONEDDS_URI="multicast"

# 配置网络接口
export CYCLONEDDS_URI="<CycloneDDS><Domain><General><Interfaces><NetworkInterface>eth0</NetworkInterface></Interfaces></General></Domain></CycloneDDS>"

# 设置 IGMP 版本（多播）
export CYCLONEDDS_URI="<CycloneDDS><Domain><General><MulticastRecvNetworkInterfaceAddrs>192.168.1.0/24</MulticastRecvNetworkInterfaceAddrs></General></Domain></CycloneDDS>"

7.3 性能监控和调优

# 监控网络带宽
ros2 topic bw /lidar_scan

# 监控消息延迟
ros2 topic delay /lidar_scan

# 监控消息频率
ros2 topic hz /lidar_scan

# 显示节点和话题的 QoS 配置
ros2 topic info /lidar_scan

八、DDS 配置文件

8.1 FastRTPS/Fast-DDS 配置

创建 fastrtps_profile.xml：

<?xml version="1.0" encoding="UTF-8" ?>
<dds>
    <profiles>
        <!-- 定义一个发布者配置 -->
        <publisher profile_name="fast_pub_profile">
            <rtps>
                <historyMemoryPolicy>DYNAMIC</historyMemoryPolicy>
                <publishMode>
                    <kind>ASYNCHRONOUS</kind>
                </publishMode>
            </rtps>
        </publisher>
        <!-- 定义一个订阅者配置 -->
        <subscriber profile_name="fast_sub_profile">
            <rtps>
                <readerTimes>
                    <initialAcknackDelay>
                        <sec>0</sec>
                        <nanosec>100000000</nanosec>
                    </initialAcknackDelay>
                </readerTimes>
            </rtps>
        </subscriber>
    </profiles>
</dds>

8.2 CycloneDDS 配置

创建 cyclonedds_config.xml：

<?xml version="1.0" encoding="UTF-8" ?>
<CycloneDDS>
    <Domain>
        <!-- 通用配置 -->
        <General>
            <NetworkInterfaceAddress>auto</NetworkInterfaceAddress>
            <AllowMulticast>default</AllowMulticast>
        </General>
        <!-- 调度器配置 -->
        <Internal>
            <ScheduleKind>default</ScheduleKind>
        </Internal>
    </Domain>
</CycloneDDS>

使用配置文件：

export CYCLONEDDS_URI=file:///path/to/cyclonedds_config.xml
ros2 run pkg_name node_name

九、常见问题和故障排查

9.1 问题：消息丢失

诊断：

# 查看话题信息
ros2 topic info /my_topic
# 检查 QoS 配置
ros2 topic info /my_topic --verbose

原因和解决方案：

9.2 问题：延迟过高

诊断：

# 测量延迟
ros2 topic delay /my_topic
# 结果示例：
# min: 15ms, max: 50ms, mean: 25ms

优化方案：

# 1. 减小队列深度
qos = QoSProfile(depth=1) # 从 10 改为 1

# 2. 使用 BEST_EFFORT
qos.reliability = QoSReliabilityPolicy.BEST_EFFORT

# 3. 使用共享内存传输（仅限本机）
export RMW_IMPLEMENTATION=rmw_cyclonedds_cpp
export CYCLONEDDS_URI="<CycloneDDS><Domain><Transport><Shared_Memory/></Transport></Domain></CycloneDDS>"

9.3 问题：发布者和订阅者无法连接

诊断：

# 检查节点状态
ros2 node list
ros2 topic list
# 查看节点详细信息
ros2 node info /publisher_node
ros2 node info /subscriber_node

常见原因：

十、本文要点总结

• DDS 基础：标准化的实时中间件，支持灵活的通信配置
• DDS 实现：
  • FastRTPS（默认，功能完整）
  • CycloneDDS（轻量级，启动快）
  • RTI Connext（企业级）
• QoS 五大参数：
  • Reliability（可靠性）：BEST_EFFORT vs RELIABLE
  • Durability（持久性）：VOLATILE vs TRANSIENT_LOCAL
  • History（历史）：KEEP_LAST vs KEEP_ALL
  • Deadline（截止期）：消息必须在此时间内到达
  • Lifespan（生存时间）：消息的有效期
• QoS 预定义配置：
  • sensor_data_qos_profile（传感器）
  • qos_profile_system_default（系统）
  • qos_profile_services_default（服务）
• 性能优化：
  • 高频传感器数据：BEST_EFFORT + KEEP_LAST(depth=1)
  • 关键命令：RELIABLE + TRANSIENT_LOCAL
  • 定期状态：BEST_EFFORT + KEEP_LAST(depth=3-5)

十一、快速参考

QoS 配置模板

# 传感器数据
qos_sensor = QoSProfile(
    reliability=QoSReliabilityPolicy.BEST_EFFORT,
    history=QoSHistoryPolicy.KEEP_LAST,
    depth=1
)
# 关键命令
qos_command = QoSProfile(
    reliability=QoSReliabilityPolicy.RELIABLE,
    durability=QoSDurabilityPolicy.TRANSIENT_LOCAL,
    depth=5
)
# 系统默认
qos_default = QoSProfile(depth=10)

常用命令

# 查看话题 QoS 信息
ros2 topic info /my_topic -v
# 切换 DDS 实现
export RMW_IMPLEMENTATION=rmw_cyclonedds_cpp
# 测试消息吞吐量
ros2 topic hz /my_topic
# 测试消息延迟
ros2 topic delay /my_topic