加载中...

Published on: 2025年1月22日星期三

Cyclone DDS：轻量级高性能的 DDS 实现

引言

在分布式实时系统中，选择合适的通信中间件对于系统的性能、可靠性和可维护性至关重要。DDS（Data Distribution Service）作为工业级的实时数据分发标准，在机器人、自动驾驶、航空航天等领域得到了广泛应用。然而，DDS 是一个标准，有多种实现可供选择，每种实现都有其独特的特点和适用场景。

Eclipse Cyclone DDS 作为 Eclipse 基金会维护的开源 DDS 实现，以其轻量级、高性能和完全符合 DDS 标准的特点，在资源受限的环境和嵌入式系统中脱颖而出。与 Fast DDS、RTI Connext DDS 等其他实现相比，Cyclone DDS 在保持高性能的同时，提供了更小的资源占用和更简洁的实现，使其成为许多项目的理想选择。

Cyclone DDS 的核心价值在于其平衡了性能、资源消耗和标准符合性。它完全实现了 OMG DDS 标准，确保了与其他 DDS 实现的互操作性，同时通过优化的架构设计，实现了低延迟、高吞吐量的通信性能。更重要的是，Cyclone DDS 的轻量级特性使其特别适合资源受限的环境，如嵌入式系统、边缘计算设备和移动机器人平台。

在 ROS2 生态系统中，Cyclone DDS 作为官方支持的 DDS 实现之一，为 ROS2 节点提供了高效的通信能力。通过 RMW（ROS Middleware）接口，Cyclone DDS 无缝集成到 ROS2 中，使得 ROS2 开发者可以轻松地使用 Cyclone DDS 作为底层通信中间件。这种集成不仅提供了去中心化的节点通信能力，还通过 DDS 的 QoS 策略实现了灵活的通信配置。

本文将带您全面深入地了解 Cyclone DDS 的方方面面，从基础概念到技术架构，从安装配置到实际应用，从性能特点到最佳实践。无论您是刚开始接触 DDS 的新手，还是希望深入了解 Cyclone DDS 技术细节的资深开发者，都能从本文中获得有价值的知识和实践指导。我们将重点关注实用操作，提供详细的技术分析和代码示例，帮助您在实际项目中快速应用 Cyclone DDS 的强大功能。

第一部分：Cyclone DDS 基础概念

什么是 Cyclone DDS

Eclipse Cyclone DDS 是由 Eclipse 基金会维护的开源 DDS 实现，完全符合 OMG DDS 标准。它提供了一个高性能、轻量级的实时数据分发中间件，专门为资源受限的环境和嵌入式系统而优化。Cyclone DDS 的核心目标是提供一个完全符合标准、高性能、易于集成的 DDS 实现，同时保持最小的资源占用。

Cyclone DDS 的历史与发展

Cyclone DDS 最初由 ADLINK Technology 开发，后来捐赠给 Eclipse 基金会，成为 Eclipse IoT 项目的一部分。这一转变不仅确保了项目的长期维护和持续发展，还通过 Eclipse 基金会的开放治理模式，吸引了更多的贡献者和用户。

Cyclone DDS 的发展历程：

2018 年：Cyclone DDS 项目启动，最初由 ADLINK Technology 开发
2019 年：项目捐赠给 Eclipse 基金会，成为 Eclipse IoT 项目的一部分
2020 年：Cyclone DDS 成为 ROS2 官方支持的 DDS 实现之一
持续发展：项目持续更新，不断优化性能和功能

Cyclone DDS 的定位

在 DDS 实现生态系统中，Cyclone DDS 占据了一个独特的位置：

轻量级：相比 Fast DDS 和 RTI Connext DDS，Cyclone DDS 的资源占用更小
高性能：通过优化的实现，提供了低延迟、高吞吐量的通信性能
标准符合：完全符合 OMG DDS 标准，确保与其他实现的互操作性
开源免费：使用 Eclipse Public License 2.0 许可证，完全开源免费

Cyclone DDS 的适用场景

Cyclone DDS 特别适合以下应用场景：

资源受限的环境：嵌入式系统、边缘计算设备、移动机器人平台
ROS2 应用：需要轻量级 DDS 实现的 ROS2 项目
实时系统：对实时性要求高的应用，如机器人控制、传感器数据分发
跨平台应用：需要在多种平台上运行的应用

Cyclone DDS 的核心特点

Cyclone DDS 具有以下几个核心特点，这些特点使其在众多 DDS 实现中脱颖而出：

1. 轻量级设计

Cyclone DDS 的设计哲学是保持实现的简洁和高效。相比其他 DDS 实现，Cyclone DDS 的代码库更小，资源占用更少，这使得它特别适合资源受限的环境。轻量级设计不仅减少了内存占用，还降低了 CPU 使用率，提高了系统的整体性能。

2. 高性能实现

尽管是轻量级实现，Cyclone DDS 并没有牺牲性能。通过优化的架构设计和高效的算法实现，Cyclone DDS 提供了低延迟、高吞吐量的通信性能。在单进程通信场景中，Cyclone DDS 的延迟可以低至 0.022 ms，这对于实时应用至关重要。

3. 完全符合 DDS 标准

Cyclone DDS 完全实现了 OMG DDS 标准，包括 DCPS（Data-Centric Publish-Subscribe）模型和 RTPS（Real-Time Publish-Subscribe）协议。这种标准符合性确保了 Cyclone DDS 与其他 DDS 实现的互操作性，使得使用不同 DDS 实现的系统可以相互通信。

4. 零配置发现

Cyclone DDS 实现了去中心化的自动发现机制，节点可以自动发现和连接，无需手动配置。这种零配置的特性简化了分布式系统的部署和维护，提高了系统的可扩展性。

5. 灵活的传输支持

Cyclone DDS 支持多种传输方式，包括 UDP（默认）、TCP 和共享内存（PSMX）。对于同一机器上的通信，可以使用共享内存传输实现零拷贝，进一步提高性能。

6. 丰富的 QoS 支持

Cyclone DDS 完全支持 DDS 标准定义的所有 QoS 策略，包括可靠性、持久性、历史记录、期限、延迟预算、存活性等。这使得开发者可以根据应用需求精确控制通信行为。

Cyclone DDS 与其他 DDS 实现的对比

理解 Cyclone DDS，需要了解它与其他主要 DDS 实现的区别：

Cyclone DDS vs Fast DDS

特性	Cyclone DDS	Fast DDS
许可证	Eclipse Public License 2.0	Apache 2.0
资源占用	较小	中等
性能	高性能	高性能
共享内存（Windows）	不支持	支持
文档	相对较少	完善
社区支持	Eclipse 基金会	eProsima
适用场景	资源受限环境	通用应用

Cyclone DDS vs RTI Connext DDS

特性	Cyclone DDS	RTI Connext DDS
许可证	开源免费	商业许可证
资源占用	小	大
功能完整性	核心功能	功能最完整
商业支持	社区支持	商业支持
工具支持	基础工具	丰富的工具
适用场景	资源受限、开源项目	商业应用、关键系统

选择建议

选择 Cyclone DDS：如果您的项目需要轻量级实现、资源受限、或需要开源解决方案
选择 Fast DDS：如果您的项目需要完善的文档、Windows 共享内存支持、或 ROS2 默认实现
选择 RTI Connext DDS：如果您的项目需要商业支持、完整的功能集、或关键系统应用

第二部分：Cyclone DDS 技术架构

Cyclone DDS 的架构层次

Cyclone DDS 的架构分为几个关键层次，每一层都有其特定的职责：

1. 应用层（Application Layer）

应用层是用户应用程序与 Cyclone DDS 交互的接口。应用程序通过 DDS API 创建和管理 DDS 实体（如域参与者、发布者、订阅者、主题等），进行数据的发布和订阅操作。

2. DDS API 层（DDS API Layer）

DDS API 层提供了符合 OMG DDS 标准的公共 C API。这一层负责：

实体管理：创建和删除域参与者、发布者、订阅者、主题等实体
读写操作：数据的写入和读取操作
QoS 配置：配置各种 QoS 策略
事件处理：通过 waitsets 和 conditions 实现基于事件的编程

3. 核心实现层（Core Implementation Layer）

核心实现层包含了 DDS 规范的实际实现，包括：

DDSI 核心：DDS Interoperability 核心实现
发现机制：SPDP（Simple Participant Discovery Protocol）和 SEDP（Simple Endpoint Discovery Protocol）
类型系统：数据类型的注册和管理
历史缓存：数据读取者和写入者的历史数据缓存
序列化/反序列化：数据的序列化和反序列化处理
参数列表：QoS 参数的管理

4. 传输层（Transport Layer）

传输层负责节点之间的通信，提供了传输抽象并支持多种传输选项：

UDP 传输：默认传输方式，提供低延迟的通信
TCP 传输：提供可靠的传输，适合需要可靠性的场景
共享内存传输（PSMX）：用于同一机器上的零拷贝通信

数据为中心的架构

Cyclone DDS 采用数据为中心的架构设计，这种设计将数据作为系统的核心，而不是以节点或服务为中心。在这种架构中：

空间解耦（Space Decoupling）

应用程序不需要知道数据生产者或消费者的身份或位置。Cyclone DDS 使用零配置、可互操作的发现协议，自动匹配对相同数据主题感兴趣的数据读取者和写入者。这种设计使得系统可以动态地添加或移除节点，而无需修改其他节点的配置。

时间解耦（Time Decoupling）

通信本质上是异步的，允许数据生产者和消费者（DataWriter 和 DataReader）独立运行，无需同时活动。中间件代表后加入的 DataReader 应用程序管理数据，当它们加入系统时交付数据。这种设计使得系统可以容忍节点的临时离线，提高了系统的可靠性。

双向解耦的优势

这种双向解耦使得应用程序可以随时从任何位置连接或断开系统，简化了数据为中心架构的复杂性和管理，特别是当 DataReader 和 DataWriter 应用程序数量增加时。

类型系统与序列化

Cyclone DDS 支持 DDS 标准的类型系统，包括：

类型注解

Cyclone DDS 支持类型演化注解，这些注解影响序列化格式和类型兼容性：

@final：类型不能演化，提供最高的性能
@appendable：可以在末尾添加新字段，向后兼容
@mutable：提供最大的类型演化灵活性

序列化格式

Cyclone DDS 支持 XCDR1 和 XCDR2 序列化格式：

XCDR1：传统的序列化格式，兼容性好
XCDR2：新的序列化格式，提供更好的性能和类型演化支持

类型注册

在使用自定义数据类型之前，需要注册类型：

#include <dds/dds.h>

// 定义数据类型
typedef struct {
    int32_t id;
    char message[256];
} HelloWorld;

// 注册类型
dds_topic_descriptor_t desc = {
    .name = "HelloWorld",
    .ops = NULL  // 使用默认操作
};

发现机制

Cyclone DDS 实现了去中心化的自动发现机制，包括两个阶段：

1. 参与者发现（Participant Discovery）

使用 SPDP（Simple Participant Discovery Protocol）协议：

每个域参与者定期发送 SPDP 消息，声明自己的存在
SPDP 消息包含域参与者的唯一标识符、IP 地址、端口等信息
域参与者通过接收 SPDP 消息发现其他参与者

2. 端点发现（Endpoint Discovery）

使用 SEDP（Simple Endpoint Discovery Protocol）协议：

在参与者发现后，使用 SEDP 交换数据写入者和数据读取者的信息
SEDP 消息包含主题名称、数据类型、QoS 策略等信息
当数据写入者和数据读取者匹配时，它们建立连接

发现配置

可以通过配置文件自定义发现行为：

<CycloneDDS>
    <Domain>
        <Id>0</Id>
        <Discovery>
            <ParticipantIndex>auto</ParticipantIndex>
            <MaxAutoParticipantIndex>0</MaxAutoParticipantIndex>
        </Discovery>
    </Domain>
</CycloneDDS>

第三部分：安装与配置

系统要求

在安装 Cyclone DDS 之前，需要确保系统满足以下要求：

操作系统支持

Linux（Ubuntu 18.04+、Debian、CentOS 等）
Windows（Windows 10+）
macOS（macOS 10.14+）

编译工具

C 编译器：GCC 7+、Clang 5+、MSVC 2017+
CMake：3.10 或更高版本
构建工具：Make、Ninja 等

依赖库

OpenSSL：用于安全功能（可选）
Bison：用于 IDL 编译器（可选）

安装方法

方法一：通过包管理器安装（Linux）

在 Ubuntu/Debian 系统上，可以通过包管理器安装：

# 更新包列表
sudo apt update

# 安装 Cyclone DDS
sudo apt install cyclonedds

# 安装开发库
sudo apt install libcyclonedds-dev

方法二：从源码编译安装

从源码编译可以获得最新版本和更多控制选项：

# 克隆仓库
git clone https://github.com/eclipse-cyclonedds/cyclonedds.git
cd cyclonedds

# 创建构建目录
mkdir build && cd build

# 配置构建
cmake .. -DCMAKE_INSTALL_PREFIX=/usr/local

# 编译
make -j$(nproc)

# 安装
sudo make install

方法三：通过 ROS2 安装

如果使用 ROS2，Cyclone DDS 可以通过 ROS2 包管理器安装：

# 安装 ROS2 Cyclone DDS RMW
sudo apt install ros-humble-rmw-cyclonedds-cpp

# 或者从源码构建
cd ~/ros2_ws/src
git clone https://github.com/ros2/rmw_cyclonedds.git -b humble
git clone https://github.com/eclipse-cyclonedds/cyclonedds.git
cd ~/ros2_ws
rosdep install --from-paths src -i
colcon build --symlink-install

ROS2 配置

设置默认 RMW

要在 ROS2 中使用 Cyclone DDS，需要设置环境变量：

# 设置 Cyclone DDS 为默认 RMW
export RMW_IMPLEMENTATION=rmw_cyclonedds_cpp

# 永久设置（添加到 ~/.bashrc）
echo "export RMW_IMPLEMENTATION=rmw_cyclonedds_cpp" >> ~/.bashrc
source ~/.bashrc

验证安装

可以通过运行 ROS2 示例节点验证安装：

# 终端 1：运行发布者
ros2 run demo_nodes_cpp talker

# 终端 2：运行订阅者
ros2 run demo_nodes_cpp listener

如果订阅者能够接收到发布者的消息，说明安装成功。

配置文件

Cyclone DDS 支持通过 XML 配置文件进行高级配置。配置文件通常命名为 cyclonedds.xml，可以放在以下位置：

当前工作目录
用户主目录（~/.config/cyclonedds.xml）
系统配置目录（/etc/cyclonedds.xml）
通过环境变量 CYCLONEDDS_URI 指定

配置文件示例

<?xml version="1.0" encoding="UTF-8"?>
<CycloneDDS>
    <Domain>
        <Id>0</Id>
        <General>
            <NetworkInterfaceAddress>auto</NetworkInterfaceAddress>
            <AllowMulticast>true</AllowMulticast>
        </General>
        <Discovery>
            <ParticipantIndex>auto</ParticipantIndex>
            <MaxAutoParticipantIndex>0</MaxAutoParticipantIndex>
            <SPDP>
                <Interfaces>
                    <NetworkInterface name="eth0" priority="0"/>
                </Interfaces>
            </SPDP>
        </Discovery>
        <Tracing>
            <Verbosity>config</Verbosity>
            <OutputFile>cyclonedds.log</OutputFile>
        </Tracing>
    </Domain>
</CycloneDDS>

环境变量配置

也可以通过环境变量进行配置：

# 指定配置文件路径
export CYCLONEDDS_URI=file:///path/to/cyclonedds.xml

# 或者直接使用配置字符串
export CYCLONEDDS_URI="<CycloneDDS><Domain><Id>0</Id></Domain></CycloneDDS>"

第四部分：使用示例

C API 基础示例

发布者示例

#include <dds/dds.h>
#include <stdio.h>
#include <string.h>

// 定义数据类型
typedef struct {
    int32_t id;
    char message[256];
} HelloWorld;

int main(void) {
    dds_entity_t participant;
    dds_entity_t topic;
    dds_entity_t writer;
    HelloWorld sample;
    dds_return_t ret;

    // 创建域参与者
    participant = dds_create_participant(DDS_DOMAIN_DEFAULT, NULL, NULL);
    if (participant < 0) {
        printf("Failed to create participant\n");
        return 1;
    }

    // 创建主题
    topic = dds_create_topic(participant, &DDS_TOPIC_HELLOWORLD,
                              "HelloWorldTopic", NULL, NULL);
    if (topic < 0) {
        printf("Failed to create topic\n");
        dds_delete(participant);
        return 1;
    }

    // 创建数据写入者
    writer = dds_create_writer(participant, topic, NULL, NULL);
    if (writer < 0) {
        printf("Failed to create writer\n");
        dds_delete(participant);
        return 1;
    }

    // 发布数据
    for (int i = 0; i < 10; i++) {
        sample.id = i;
        snprintf(sample.message, sizeof(sample.message),
                 "Hello, Cyclone DDS! Message %d", i);

        ret = dds_write(writer, &sample);
        if (ret != DDS_RETCODE_OK) {
            printf("Failed to write sample\n");
        } else {
            printf("Published: id=%d, message=%s\n",
                   sample.id, sample.message);
        }

        dds_sleepfor(DDS_MSECS(500));
    }

    // 清理资源
    dds_delete(participant);
    return 0;
}

订阅者示例

#include <dds/dds.h>
#include <stdio.h>

// 定义数据类型（与发布者相同）
typedef struct {
    int32_t id;
    char message[256];
} HelloWorld;

int main(void) {
    dds_entity_t participant;
    dds_entity_t topic;
    dds_entity_t reader;
    HelloWorld *sample;
    void *samples[1];
    dds_sample_info_t infos[1];
    dds_return_t ret;

    // 创建域参与者
    participant = dds_create_participant(DDS_DOMAIN_DEFAULT, NULL, NULL);
    if (participant < 0) {
        printf("Failed to create participant\n");
        return 1;
    }

    // 创建主题
    topic = dds_create_topic(participant, &DDS_TOPIC_HELLOWORLD,
                              "HelloWorldTopic", NULL, NULL);
    if (topic < 0) {
        printf("Failed to create topic\n");
        dds_delete(participant);
        return 1;
    }

    // 创建数据读取者
    reader = dds_create_reader(participant, topic, NULL, NULL);
    if (reader < 0) {
        printf("Failed to create reader\n");
        dds_delete(participant);
        return 1;
    }

    // 等待数据
    printf("Waiting for data...\n");
    ret = dds_set_status_mask(reader, DDS_DATA_AVAILABLE_STATUS);
    if (ret != DDS_RETCODE_OK) {
        printf("Failed to set status mask\n");
        dds_delete(participant);
        return 1;
    }

    // 读取数据
    samples[0] = &sample;
    while (true) {
        ret = dds_take(reader, samples, infos, 1, 1);
        if (ret > 0) {
            if (infos[0].valid_data) {
                printf("Received: id=%d, message=%s\n",
                       sample->id, sample->message);
            }
        } else if (ret == DDS_RETCODE_TIMEOUT) {
            printf("Timeout waiting for data\n");
            break;
        } else {
            printf("Error reading data\n");
            break;
        }

        dds_sleepfor(DDS_MSECS(100));
    }

    // 清理资源
    dds_delete(participant);
    return 0;
}

ROS2 使用示例

C++ 发布者节点

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

using namespace std::chrono_literals;

class PublisherNode : public rclcpp::Node {
public:
    PublisherNode() : Node("cyclone_publisher") {
        publisher_ = this->create_publisher<std_msgs::msg::String>(
            "topic", 10);

        timer_ = this->create_wall_timer(
            500ms, std::bind(&PublisherNode::timer_callback, this));
    }

private:
    void timer_callback() {
        auto message = std_msgs::msg::String();
        message.data = "Hello from Cyclone DDS!";
        RCLCPP_INFO(this->get_logger(), "Publishing: '%s'",
                    message.data.c_str());
        publisher_->publish(message);
    }

    rclcpp::TimerBase::SharedPtr timer_;
    rclcpp::Publisher<std_msgs::msg::String>::SharedPtr publisher_;
};

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

C++ 订阅者节点

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

class SubscriberNode : public rclcpp::Node {
public:
    SubscriberNode() : Node("cyclone_subscriber") {
        subscription_ = this->create_subscription<std_msgs::msg::String>(
            "topic", 10,
            std::bind(&SubscriberNode::topic_callback, this,
                      std::placeholders::_1));
    }

private:
    void topic_callback(const std_msgs::msg::String::SharedPtr msg) const {
        RCLCPP_INFO(this->get_logger(), "I heard: '%s'", msg->data.c_str());
    }

    rclcpp::Subscription<std_msgs::msg::String>::SharedPtr subscription_;
};

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

Python 示例

import rclpy
from rclpy.node import Node
from std_msgs.msg import String

class PublisherNode(Node):
    def __init__(self):
        super().__init__('cyclone_publisher')
        self.publisher_ = self.create_publisher(String, 'topic', 10)
        timer_period = 0.5  # seconds
        self.timer = self.create_timer(timer_period, self.timer_callback)
        self.i = 0

    def timer_callback(self):
        msg = String()
        msg.data = f'Hello from Cyclone DDS: {self.i}'
        self.publisher_.publish(msg)
        self.get_logger().info(f'Publishing: "{msg.data}"')
        self.i += 1

class SubscriberNode(Node):
    def __init__(self):
        super().__init__('cyclone_subscriber')
        self.subscription = self.create_subscription(
            String,
            'topic',
            self.listener_callback,
            10)

    def listener_callback(self, msg):
        self.get_logger().info(f'I heard: "{msg.data}"')

def main(args=None):
    rclpy.init(args=args)

    # 使用发布者或订阅者
    # publisher = PublisherNode()
    subscriber = SubscriberNode()

    rclpy.spin(subscriber)
    subscriber.destroy_node()
    rclpy.shutdown()

if __name__ == '__main__':
    main()

自定义 QoS 配置

C API QoS 配置

#include <dds/dds.h>

// 创建自定义 QoS
dds_qos_t *qos = dds_create_qos();

// 配置可靠性
dds_qset_reliability(qos, DDS_RELIABILITY_RELIABLE,
                     DDS_SECS(10));

// 配置持久性
dds_qset_durability(qos, DDS_DURABILITY_TRANSIENT_LOCAL);

// 配置历史记录
dds_qset_history(qos, DDS_HISTORY_KEEP_LAST, 10);

// 使用自定义 QoS 创建数据写入者
dds_entity_t writer = dds_create_writer(participant, topic, qos, NULL);

// 清理 QoS
dds_delete_qos(qos);

ROS2 QoS 配置

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

// 创建自定义 QoS
rclcpp::QoS custom_qos(10);
custom_qos.reliability(RMW_QOS_POLICY_RELIABILITY_RELIABLE);
custom_qos.durability(RMW_QOS_POLICY_DURABILITY_TRANSIENT_LOCAL);
custom_qos.history(RMW_QOS_POLICY_HISTORY_KEEP_LAST);

// 使用自定义 QoS 创建发布者
auto publisher = node->create_publisher<std_msgs::msg::String>(
    "topic", custom_qos);

第五部分：性能优化与最佳实践

性能优化技巧

1. 选择合适的传输方式

UDP：默认传输，提供最低延迟，适合实时数据流
TCP：提供可靠传输，适合需要可靠性的场景
共享内存（PSMX）：同一机器上的通信，零拷贝，最高性能

2. 优化 QoS 配置

实时传感器数据：使用 BEST_EFFORT 可靠性，VOLATILE 持久性
控制命令：使用 RELIABLE 可靠性，TRANSIENT_LOCAL 持久性
状态信息：使用 RELIABLE 可靠性，TRANSIENT_LOCAL 持久性，KEEP_LAST 历史记录

3. 减少历史记录深度

对于只需要最新数据的场景，使用 KEEP_LAST 而不是 KEEP_ALL，并设置较小的深度值：

dds_qset_history(qos, DDS_HISTORY_KEEP_LAST, 1);

4. 使用批量写入

对于需要发布多个样本的场景，使用批量写入可以提高性能：

HelloWorld samples[10];
// ... 填充样本数据 ...
dds_write_multi(writer, samples, 10);

5. 优化网络配置

使用专用网络接口
配置多播地址
优化防火墙规则

调试与故障排除

启用日志

Cyclone DDS 提供了详细的日志功能，可以通过配置文件启用：

<CycloneDDS>
    <Domain>
        <Tracing>
            <Verbosity>trace</Verbosity>
            <OutputFile>cyclonedds.log</OutputFile>
        </Tracing>
    </Domain>
</CycloneDDS>

常见问题

连接失败
- 检查域 ID 是否匹配
- 检查主题名称和数据类型是否匹配
- 检查 QoS 策略是否兼容
- 检查网络连接和防火墙设置
数据丢失
- 检查可靠性 QoS 配置
- 检查历史记录深度
- 检查资源限制
延迟过高
- 检查网络延迟
- 检查 QoS 配置
- 考虑使用共享内存传输（同一机器）

性能分析工具

可以使用以下工具分析 Cyclone DDS 的性能：

Wireshark：网络包分析工具，可以分析 RTPS 协议流量
perf：Linux 性能分析工具
Cyclone DDS 日志：通过配置文件启用详细日志

最佳实践

1. 合理选择 DDS 实现

根据项目需求选择合适的 DDS 实现：

资源受限环境：选择 Cyclone DDS
需要完善文档：选择 Fast DDS
商业应用：考虑 RTI Connext DDS

2. 统一 QoS 配置

在项目中统一 QoS 配置，避免不兼容问题。可以创建 QoS 配置文件或使用代码中的常量定义。

3. 错误处理

始终检查 API 调用的返回值，正确处理错误情况：

dds_return_t ret = dds_write(writer, &sample);
if (ret != DDS_RETCODE_OK) {
    printf("Write failed: %s\n", dds_strretcode(ret));
    // 处理错误
}

4. 资源管理

及时清理不再使用的资源，避免内存泄漏：

dds_delete(writer);
dds_delete(topic);
dds_delete(participant);

5. 测试与验证

在部署到生产环境之前，充分测试系统的性能和可靠性，包括：

延迟测试
吞吐量测试
可靠性测试
压力测试

第六部分：在 ROS2 中的应用

ROS2 与 Cyclone DDS 的集成

ROS2 通过 RMW（ROS Middleware）接口支持多种 DDS 实现，Cyclone DDS 是其中之一。这种集成使得 ROS2 节点可以使用 Cyclone DDS 作为底层通信中间件，实现去中心化的节点通信。

RMW 接口

RMW 接口是 ROS2 与底层中间件之间的抽象层，它定义了标准的接口，使得 ROS2 可以支持不同的 DDS 实现。rmw_cyclonedds_cpp 是 Cyclone DDS 的 RMW 实现，它实现了 RMW 接口，将 ROS2 的 API 调用转换为 Cyclone DDS 的 API 调用。

ROS2 节点通信流程

ROS2 节点创建发布者或订阅者
RMW 层将 ROS2 的请求转换为 DDS 操作
Cyclone DDS 处理实际的通信
数据通过 RTPS 协议在网络上传输

ROS2 QoS 映射

ROS2 的 QoS 配置会映射到 DDS 的 QoS 策略：

ROS2 QoS	DDS QoS
Reliability: RELIABLE	Reliability: RELIABLE
Reliability: BEST_EFFORT	Reliability: BEST_EFFORT
Durability: VOLATILE	Durability: VOLATILE
Durability: TRANSIENT_LOCAL	Durability: TRANSIENT_LOCAL
History: KEEP_LAST	History: KEEP_LAST
History: KEEP_ALL	History: KEEP_ALL

ROS2 QoS 配置示例

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

// 创建可靠、持久化的 QoS
rclcpp::QoS reliable_qos(10);
reliable_qos.reliability(RMW_QOS_POLICY_RELIABILITY_RELIABLE);
reliable_qos.durability(RMW_QOS_POLICY_DURABILITY_TRANSIENT_LOCAL);

// 创建发布者
auto publisher = node->create_publisher<std_msgs::msg::String>(
    "topic", reliable_qos);

ROS2 性能优化

1. 选择合适的 RMW

根据应用需求选择合适的 RMW：

Cyclone DDS：轻量级，资源占用小
Fast DDS：功能完整，文档完善
RTI Connext DDS：商业支持，功能最完整

2. 优化 ROS2 QoS

根据数据类型选择合适的 QoS：

传感器数据：BEST_EFFORT，VOLATILE
控制命令：RELIABLE，TRANSIENT_LOCAL
状态信息：RELIABLE，TRANSIENT_LOCAL

3. 使用共享内存传输

对于同一机器上的 ROS2 节点通信，可以使用共享内存传输提高性能（Fast DDS 支持，Cyclone DDS 在 Linux 上支持）。

第七部分：总结与展望

Cyclone DDS 的优势总结

Cyclone DDS 作为轻量级、高性能的 DDS 实现，具有以下主要优势：

1. 轻量级设计

Cyclone DDS 的资源占用小，特别适合资源受限的环境。这使得它可以在嵌入式系统、边缘计算设备和移动机器人平台上高效运行。

2. 高性能实现

尽管是轻量级实现，Cyclone DDS 提供了低延迟、高吞吐量的通信性能，满足实时应用的需求。

3. 完全符合标准

Cyclone DDS 完全实现了 OMG DDS 标准，确保了与其他 DDS 实现的互操作性。

4. 开源免费

使用 Eclipse Public License 2.0 许可证，完全开源免费，适合各种项目。

5. ROS2 集成

作为 ROS2 官方支持的 DDS 实现之一，Cyclone DDS 与 ROS2 无缝集成，为 ROS2 应用提供了高效的通信能力。

Cyclone DDS 的局限性

尽管 Cyclone DDS 具有许多优势，但也存在一些局限性：

1. 文档相对较少

相比 Fast DDS 和 RTI Connext DDS，Cyclone DDS 的文档相对较少，学习曲线可能较陡。

2. 功能相对较少

Cyclone DDS 专注于核心功能，一些高级功能可能不如商业实现完整。

3. Windows 共享内存支持

Cyclone DDS 在 Windows 上不支持共享内存传输，这可能影响同一机器上的通信性能。

4. 社区支持

虽然 Eclipse 基金会提供了良好的支持，但社区规模相对较小，可能影响问题解决速度。

Cyclone DDS 的未来发展

Cyclone DDS 正在不断发展，未来的发展方向包括：

1. 性能优化

持续优化性能，减少延迟，提高吞吐量，降低资源消耗。

2. 功能扩展

扩展功能，包括更好的安全支持、更多的传输选项、更好的工具支持等。

3. 文档改进

改进文档，提供更多的教程和示例代码，降低学习曲线。

4. 社区发展

通过 Eclipse 基金会的支持，吸引更多的贡献者和用户，扩大社区规模。

学习资源推荐

官方文档

Cyclone DDS 官方文档：https://cyclonedds.io/docs/cyclonedds/latest/
GitHub 仓库：https://github.com/eclipse-cyclonedds/cyclonedds
Eclipse IoT 项目：https://iot.eclipse.org/

ROS2 相关资源

ROS2 DDS 实现文档：https://docs.ros.org/en/rolling/Installation/RMW-Implementations/DDS-Implementations/Working-with-Eclipse-CycloneDDS.html
ROS2 社区：https://discourse.ros.org/

示例代码

Cyclone DDS 示例：https://github.com/eclipse-cyclonedds/cyclonedds/tree/master/examples
ROS2 示例：https://github.com/ros2/examples

结语

Cyclone DDS 作为轻量级、高性能的 DDS 实现，在资源受限的环境和嵌入式系统中发挥着重要作用。通过本文的全面介绍，我们希望您能够深入理解 Cyclone DDS 的核心概念、技术架构、安装配置和实际应用。

无论您是刚开始接触 DDS 的新手，还是希望深入了解 Cyclone DDS 技术细节的资深开发者，Cyclone DDS 都为您提供了一个轻量级、高性能、完全符合标准的实时通信平台。通过合理使用 Cyclone DDS 的 QoS 策略和最佳实践，您可以构建出高性能、高可靠性的分布式实时系统。

随着 Cyclone DDS 技术的不断发展和完善，我们相信它将在更多领域得到应用，特别是在资源受限的环境和嵌入式系统中。希望本文能够帮助您在 Cyclone DDS 的学习和应用中取得成功。

GMR：通用动作重映射器

使用 Dotfile 同步配置

编码器正交编码方法

发表评论

请登录后发表评论

发表评论

请登录后发表评论

暂无评论，快来抢沙发吧！

Published on: 2025年1月22日星期三

Cyclone DDS：轻量级高性能的 DDS 实现

引言

第一部分：Cyclone DDS 基础概念

什么是 Cyclone DDS

Cyclone DDS 的历史与发展

Cyclone DDS 的发展历程：

2018 年：Cyclone DDS 项目启动，最初由 ADLINK Technology 开发
2019 年：项目捐赠给 Eclipse 基金会，成为 Eclipse IoT 项目的一部分
2020 年：Cyclone DDS 成为 ROS2 官方支持的 DDS 实现之一
持续发展：项目持续更新，不断优化性能和功能

Cyclone DDS 的定位

在 DDS 实现生态系统中，Cyclone DDS 占据了一个独特的位置：

轻量级：相比 Fast DDS 和 RTI Connext DDS，Cyclone DDS 的资源占用更小
高性能：通过优化的实现，提供了低延迟、高吞吐量的通信性能
标准符合：完全符合 OMG DDS 标准，确保与其他实现的互操作性
开源免费：使用 Eclipse Public License 2.0 许可证，完全开源免费

Cyclone DDS 的适用场景

Cyclone DDS 特别适合以下应用场景：

资源受限的环境：嵌入式系统、边缘计算设备、移动机器人平台
ROS2 应用：需要轻量级 DDS 实现的 ROS2 项目
实时系统：对实时性要求高的应用，如机器人控制、传感器数据分发
跨平台应用：需要在多种平台上运行的应用

Cyclone DDS 的核心特点

Cyclone DDS 具有以下几个核心特点，这些特点使其在众多 DDS 实现中脱颖而出：

1. 轻量级设计

2. 高性能实现

3. 完全符合 DDS 标准

4. 零配置发现

5. 灵活的传输支持

6. 丰富的 QoS 支持

Cyclone DDS 与其他 DDS 实现的对比

理解 Cyclone DDS，需要了解它与其他主要 DDS 实现的区别：

Cyclone DDS vs Fast DDS

特性	Cyclone DDS	Fast DDS
许可证	Eclipse Public License 2.0	Apache 2.0
资源占用	较小	中等
性能	高性能	高性能
共享内存（Windows）	不支持	支持
文档	相对较少	完善
社区支持	Eclipse 基金会	eProsima
适用场景	资源受限环境	通用应用

Cyclone DDS vs RTI Connext DDS

特性	Cyclone DDS	RTI Connext DDS
许可证	开源免费	商业许可证
资源占用	小	大
功能完整性	核心功能	功能最完整
商业支持	社区支持	商业支持
工具支持	基础工具	丰富的工具
适用场景	资源受限、开源项目	商业应用、关键系统

选择建议

选择 Cyclone DDS：如果您的项目需要轻量级实现、资源受限、或需要开源解决方案
选择 Fast DDS：如果您的项目需要完善的文档、Windows 共享内存支持、或 ROS2 默认实现
选择 RTI Connext DDS：如果您的项目需要商业支持、完整的功能集、或关键系统应用

第二部分：Cyclone DDS 技术架构

Cyclone DDS 的架构层次

Cyclone DDS 的架构分为几个关键层次，每一层都有其特定的职责：

1. 应用层（Application Layer）

2. DDS API 层（DDS API Layer）

DDS API 层提供了符合 OMG DDS 标准的公共 C API。这一层负责：

实体管理：创建和删除域参与者、发布者、订阅者、主题等实体
读写操作：数据的写入和读取操作
QoS 配置：配置各种 QoS 策略
事件处理：通过 waitsets 和 conditions 实现基于事件的编程

3. 核心实现层（Core Implementation Layer）

核心实现层包含了 DDS 规范的实际实现，包括：

DDSI 核心：DDS Interoperability 核心实现
发现机制：SPDP（Simple Participant Discovery Protocol）和 SEDP（Simple Endpoint Discovery Protocol）
类型系统：数据类型的注册和管理
历史缓存：数据读取者和写入者的历史数据缓存
序列化/反序列化：数据的序列化和反序列化处理
参数列表：QoS 参数的管理

4. 传输层（Transport Layer）

传输层负责节点之间的通信，提供了传输抽象并支持多种传输选项：

UDP 传输：默认传输方式，提供低延迟的通信
TCP 传输：提供可靠的传输，适合需要可靠性的场景
共享内存传输（PSMX）：用于同一机器上的零拷贝通信

数据为中心的架构

Cyclone DDS 采用数据为中心的架构设计，这种设计将数据作为系统的核心，而不是以节点或服务为中心。在这种架构中：

空间解耦（Space Decoupling）

时间解耦（Time Decoupling）

双向解耦的优势

类型系统与序列化

Cyclone DDS 支持 DDS 标准的类型系统，包括：

类型注解

Cyclone DDS 支持类型演化注解，这些注解影响序列化格式和类型兼容性：

@final：类型不能演化，提供最高的性能
@appendable：可以在末尾添加新字段，向后兼容
@mutable：提供最大的类型演化灵活性

序列化格式

Cyclone DDS 支持 XCDR1 和 XCDR2 序列化格式：

XCDR1：传统的序列化格式，兼容性好
XCDR2：新的序列化格式，提供更好的性能和类型演化支持

类型注册

在使用自定义数据类型之前，需要注册类型：

#include <dds/dds.h>

// 定义数据类型
typedef struct {
    int32_t id;
    char message[256];
} HelloWorld;

// 注册类型
dds_topic_descriptor_t desc = {
    .name = "HelloWorld",
    .ops = NULL  // 使用默认操作
};

发现机制

Cyclone DDS 实现了去中心化的自动发现机制，包括两个阶段：

1. 参与者发现（Participant Discovery）

使用 SPDP（Simple Participant Discovery Protocol）协议：

每个域参与者定期发送 SPDP 消息，声明自己的存在
SPDP 消息包含域参与者的唯一标识符、IP 地址、端口等信息
域参与者通过接收 SPDP 消息发现其他参与者

2. 端点发现（Endpoint Discovery）

使用 SEDP（Simple Endpoint Discovery Protocol）协议：

在参与者发现后，使用 SEDP 交换数据写入者和数据读取者的信息
SEDP 消息包含主题名称、数据类型、QoS 策略等信息
当数据写入者和数据读取者匹配时，它们建立连接

发现配置

可以通过配置文件自定义发现行为：

<CycloneDDS>
    <Domain>
        <Id>0</Id>
        <Discovery>
            <ParticipantIndex>auto</ParticipantIndex>
            <MaxAutoParticipantIndex>0</MaxAutoParticipantIndex>
        </Discovery>
    </Domain>
</CycloneDDS>

第三部分：安装与配置

系统要求

在安装 Cyclone DDS 之前，需要确保系统满足以下要求：

操作系统支持

Linux（Ubuntu 18.04+、Debian、CentOS 等）
Windows（Windows 10+）
macOS（macOS 10.14+）

编译工具

C 编译器：GCC 7+、Clang 5+、MSVC 2017+
CMake：3.10 或更高版本
构建工具：Make、Ninja 等

依赖库

OpenSSL：用于安全功能（可选）
Bison：用于 IDL 编译器（可选）

安装方法

方法一：通过包管理器安装（Linux）

在 Ubuntu/Debian 系统上，可以通过包管理器安装：

# 更新包列表
sudo apt update

# 安装 Cyclone DDS
sudo apt install cyclonedds

# 安装开发库
sudo apt install libcyclonedds-dev

方法二：从源码编译安装

从源码编译可以获得最新版本和更多控制选项：

# 克隆仓库
git clone https://github.com/eclipse-cyclonedds/cyclonedds.git
cd cyclonedds

# 创建构建目录
mkdir build && cd build

# 配置构建
cmake .. -DCMAKE_INSTALL_PREFIX=/usr/local

# 编译
make -j$(nproc)

# 安装
sudo make install

方法三：通过 ROS2 安装

如果使用 ROS2，Cyclone DDS 可以通过 ROS2 包管理器安装：

# 安装 ROS2 Cyclone DDS RMW
sudo apt install ros-humble-rmw-cyclonedds-cpp

# 或者从源码构建
cd ~/ros2_ws/src
git clone https://github.com/ros2/rmw_cyclonedds.git -b humble
git clone https://github.com/eclipse-cyclonedds/cyclonedds.git
cd ~/ros2_ws
rosdep install --from-paths src -i
colcon build --symlink-install

ROS2 配置

设置默认 RMW

要在 ROS2 中使用 Cyclone DDS，需要设置环境变量：

# 设置 Cyclone DDS 为默认 RMW
export RMW_IMPLEMENTATION=rmw_cyclonedds_cpp

# 永久设置（添加到 ~/.bashrc）
echo "export RMW_IMPLEMENTATION=rmw_cyclonedds_cpp" >> ~/.bashrc
source ~/.bashrc

验证安装

可以通过运行 ROS2 示例节点验证安装：

# 终端 1：运行发布者
ros2 run demo_nodes_cpp talker

# 终端 2：运行订阅者
ros2 run demo_nodes_cpp listener

如果订阅者能够接收到发布者的消息，说明安装成功。

配置文件

Cyclone DDS 支持通过 XML 配置文件进行高级配置。配置文件通常命名为 cyclonedds.xml，可以放在以下位置：

当前工作目录
用户主目录（~/.config/cyclonedds.xml）
系统配置目录（/etc/cyclonedds.xml）
通过环境变量 CYCLONEDDS_URI 指定

配置文件示例

<?xml version="1.0" encoding="UTF-8"?>
<CycloneDDS>
    <Domain>
        <Id>0</Id>
        <General>
            <NetworkInterfaceAddress>auto</NetworkInterfaceAddress>
            <AllowMulticast>true</AllowMulticast>
        </General>
        <Discovery>
            <ParticipantIndex>auto</ParticipantIndex>
            <MaxAutoParticipantIndex>0</MaxAutoParticipantIndex>
            <SPDP>
                <Interfaces>
                    <NetworkInterface name="eth0" priority="0"/>
                </Interfaces>
            </SPDP>
        </Discovery>
        <Tracing>
            <Verbosity>config</Verbosity>
            <OutputFile>cyclonedds.log</OutputFile>
        </Tracing>
    </Domain>
</CycloneDDS>

环境变量配置

也可以通过环境变量进行配置：

# 指定配置文件路径
export CYCLONEDDS_URI=file:///path/to/cyclonedds.xml

# 或者直接使用配置字符串
export CYCLONEDDS_URI="<CycloneDDS><Domain><Id>0</Id></Domain></CycloneDDS>"

第四部分：使用示例

C API 基础示例

发布者示例

#include <dds/dds.h>
#include <stdio.h>
#include <string.h>

// 定义数据类型
typedef struct {
    int32_t id;
    char message[256];
} HelloWorld;

int main(void) {
    dds_entity_t participant;
    dds_entity_t topic;
    dds_entity_t writer;
    HelloWorld sample;
    dds_return_t ret;

    // 创建域参与者
    participant = dds_create_participant(DDS_DOMAIN_DEFAULT, NULL, NULL);
    if (participant < 0) {
        printf("Failed to create participant\n");
        return 1;
    }

    // 创建主题
    topic = dds_create_topic(participant, &DDS_TOPIC_HELLOWORLD,
                              "HelloWorldTopic", NULL, NULL);
    if (topic < 0) {
        printf("Failed to create topic\n");
        dds_delete(participant);
        return 1;
    }

    // 创建数据写入者
    writer = dds_create_writer(participant, topic, NULL, NULL);
    if (writer < 0) {
        printf("Failed to create writer\n");
        dds_delete(participant);
        return 1;
    }

    // 发布数据
    for (int i = 0; i < 10; i++) {
        sample.id = i;
        snprintf(sample.message, sizeof(sample.message),
                 "Hello, Cyclone DDS! Message %d", i);

        ret = dds_write(writer, &sample);
        if (ret != DDS_RETCODE_OK) {
            printf("Failed to write sample\n");
        } else {
            printf("Published: id=%d, message=%s\n",
                   sample.id, sample.message);
        }

        dds_sleepfor(DDS_MSECS(500));
    }

    // 清理资源
    dds_delete(participant);
    return 0;
}

订阅者示例

#include <dds/dds.h>
#include <stdio.h>

// 定义数据类型（与发布者相同）
typedef struct {
    int32_t id;
    char message[256];
} HelloWorld;

int main(void) {
    dds_entity_t participant;
    dds_entity_t topic;
    dds_entity_t reader;
    HelloWorld *sample;
    void *samples[1];
    dds_sample_info_t infos[1];
    dds_return_t ret;

    // 创建域参与者
    participant = dds_create_participant(DDS_DOMAIN_DEFAULT, NULL, NULL);
    if (participant < 0) {
        printf("Failed to create participant\n");
        return 1;
    }

    // 创建主题
    topic = dds_create_topic(participant, &DDS_TOPIC_HELLOWORLD,
                              "HelloWorldTopic", NULL, NULL);
    if (topic < 0) {
        printf("Failed to create topic\n");
        dds_delete(participant);
        return 1;
    }

    // 创建数据读取者
    reader = dds_create_reader(participant, topic, NULL, NULL);
    if (reader < 0) {
        printf("Failed to create reader\n");
        dds_delete(participant);
        return 1;
    }

    // 等待数据
    printf("Waiting for data...\n");
    ret = dds_set_status_mask(reader, DDS_DATA_AVAILABLE_STATUS);
    if (ret != DDS_RETCODE_OK) {
        printf("Failed to set status mask\n");
        dds_delete(participant);
        return 1;
    }

    // 读取数据
    samples[0] = &sample;
    while (true) {
        ret = dds_take(reader, samples, infos, 1, 1);
        if (ret > 0) {
            if (infos[0].valid_data) {
                printf("Received: id=%d, message=%s\n",
                       sample->id, sample->message);
            }
        } else if (ret == DDS_RETCODE_TIMEOUT) {
            printf("Timeout waiting for data\n");
            break;
        } else {
            printf("Error reading data\n");
            break;
        }

        dds_sleepfor(DDS_MSECS(100));
    }

    // 清理资源
    dds_delete(participant);
    return 0;
}

ROS2 使用示例

C++ 发布者节点

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

using namespace std::chrono_literals;

class PublisherNode : public rclcpp::Node {
public:
    PublisherNode() : Node("cyclone_publisher") {
        publisher_ = this->create_publisher<std_msgs::msg::String>(
            "topic", 10);

        timer_ = this->create_wall_timer(
            500ms, std::bind(&PublisherNode::timer_callback, this));
    }

private:
    void timer_callback() {
        auto message = std_msgs::msg::String();
        message.data = "Hello from Cyclone DDS!";
        RCLCPP_INFO(this->get_logger(), "Publishing: '%s'",
                    message.data.c_str());
        publisher_->publish(message);
    }

    rclcpp::TimerBase::SharedPtr timer_;
    rclcpp::Publisher<std_msgs::msg::String>::SharedPtr publisher_;
};

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

C++ 订阅者节点

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

class SubscriberNode : public rclcpp::Node {
public:
    SubscriberNode() : Node("cyclone_subscriber") {
        subscription_ = this->create_subscription<std_msgs::msg::String>(
            "topic", 10,
            std::bind(&SubscriberNode::topic_callback, this,
                      std::placeholders::_1));
    }

private:
    void topic_callback(const std_msgs::msg::String::SharedPtr msg) const {
        RCLCPP_INFO(this->get_logger(), "I heard: '%s'", msg->data.c_str());
    }

    rclcpp::Subscription<std_msgs::msg::String>::SharedPtr subscription_;
};

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

Python 示例

import rclpy
from rclpy.node import Node
from std_msgs.msg import String

class PublisherNode(Node):
    def __init__(self):
        super().__init__('cyclone_publisher')
        self.publisher_ = self.create_publisher(String, 'topic', 10)
        timer_period = 0.5  # seconds
        self.timer = self.create_timer(timer_period, self.timer_callback)
        self.i = 0

    def timer_callback(self):
        msg = String()
        msg.data = f'Hello from Cyclone DDS: {self.i}'
        self.publisher_.publish(msg)
        self.get_logger().info(f'Publishing: "{msg.data}"')
        self.i += 1

class SubscriberNode(Node):
    def __init__(self):
        super().__init__('cyclone_subscriber')
        self.subscription = self.create_subscription(
            String,
            'topic',
            self.listener_callback,
            10)

    def listener_callback(self, msg):
        self.get_logger().info(f'I heard: "{msg.data}"')

def main(args=None):
    rclpy.init(args=args)

    # 使用发布者或订阅者
    # publisher = PublisherNode()
    subscriber = SubscriberNode()

    rclpy.spin(subscriber)
    subscriber.destroy_node()
    rclpy.shutdown()

if __name__ == '__main__':
    main()

自定义 QoS 配置

C API QoS 配置

#include <dds/dds.h>

// 创建自定义 QoS
dds_qos_t *qos = dds_create_qos();

// 配置可靠性
dds_qset_reliability(qos, DDS_RELIABILITY_RELIABLE,
                     DDS_SECS(10));

// 配置持久性
dds_qset_durability(qos, DDS_DURABILITY_TRANSIENT_LOCAL);

// 配置历史记录
dds_qset_history(qos, DDS_HISTORY_KEEP_LAST, 10);

// 使用自定义 QoS 创建数据写入者
dds_entity_t writer = dds_create_writer(participant, topic, qos, NULL);

// 清理 QoS
dds_delete_qos(qos);

ROS2 QoS 配置

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

// 创建自定义 QoS
rclcpp::QoS custom_qos(10);
custom_qos.reliability(RMW_QOS_POLICY_RELIABILITY_RELIABLE);
custom_qos.durability(RMW_QOS_POLICY_DURABILITY_TRANSIENT_LOCAL);
custom_qos.history(RMW_QOS_POLICY_HISTORY_KEEP_LAST);

// 使用自定义 QoS 创建发布者
auto publisher = node->create_publisher<std_msgs::msg::String>(
    "topic", custom_qos);

第五部分：性能优化与最佳实践

性能优化技巧

1. 选择合适的传输方式

UDP：默认传输，提供最低延迟，适合实时数据流
TCP：提供可靠传输，适合需要可靠性的场景
共享内存（PSMX）：同一机器上的通信，零拷贝，最高性能

2. 优化 QoS 配置

实时传感器数据：使用 BEST_EFFORT 可靠性，VOLATILE 持久性
控制命令：使用 RELIABLE 可靠性，TRANSIENT_LOCAL 持久性
状态信息：使用 RELIABLE 可靠性，TRANSIENT_LOCAL 持久性，KEEP_LAST 历史记录

3. 减少历史记录深度

对于只需要最新数据的场景，使用 KEEP_LAST 而不是 KEEP_ALL，并设置较小的深度值：

dds_qset_history(qos, DDS_HISTORY_KEEP_LAST, 1);

4. 使用批量写入

对于需要发布多个样本的场景，使用批量写入可以提高性能：

HelloWorld samples[10];
// ... 填充样本数据 ...
dds_write_multi(writer, samples, 10);

5. 优化网络配置

使用专用网络接口
配置多播地址
优化防火墙规则

调试与故障排除

启用日志

Cyclone DDS 提供了详细的日志功能，可以通过配置文件启用：

<CycloneDDS>
    <Domain>
        <Tracing>
            <Verbosity>trace</Verbosity>
            <OutputFile>cyclonedds.log</OutputFile>
        </Tracing>
    </Domain>
</CycloneDDS>

常见问题

连接失败
- 检查域 ID 是否匹配
- 检查主题名称和数据类型是否匹配
- 检查 QoS 策略是否兼容
- 检查网络连接和防火墙设置
数据丢失
- 检查可靠性 QoS 配置
- 检查历史记录深度
- 检查资源限制
延迟过高
- 检查网络延迟
- 检查 QoS 配置
- 考虑使用共享内存传输（同一机器）

性能分析工具

可以使用以下工具分析 Cyclone DDS 的性能：

Wireshark：网络包分析工具，可以分析 RTPS 协议流量
perf：Linux 性能分析工具
Cyclone DDS 日志：通过配置文件启用详细日志

最佳实践

1. 合理选择 DDS 实现

根据项目需求选择合适的 DDS 实现：

资源受限环境：选择 Cyclone DDS
需要完善文档：选择 Fast DDS
商业应用：考虑 RTI Connext DDS

2. 统一 QoS 配置

在项目中统一 QoS 配置，避免不兼容问题。可以创建 QoS 配置文件或使用代码中的常量定义。

3. 错误处理

始终检查 API 调用的返回值，正确处理错误情况：

dds_return_t ret = dds_write(writer, &sample);
if (ret != DDS_RETCODE_OK) {
    printf("Write failed: %s\n", dds_strretcode(ret));
    // 处理错误
}

4. 资源管理

及时清理不再使用的资源，避免内存泄漏：

dds_delete(writer);
dds_delete(topic);
dds_delete(participant);

5. 测试与验证

在部署到生产环境之前，充分测试系统的性能和可靠性，包括：

延迟测试
吞吐量测试
可靠性测试
压力测试

第六部分：在 ROS2 中的应用

ROS2 与 Cyclone DDS 的集成

RMW 接口

ROS2 节点通信流程

ROS2 节点创建发布者或订阅者
RMW 层将 ROS2 的请求转换为 DDS 操作
Cyclone DDS 处理实际的通信
数据通过 RTPS 协议在网络上传输

ROS2 QoS 映射

ROS2 的 QoS 配置会映射到 DDS 的 QoS 策略：

ROS2 QoS	DDS QoS
Reliability: RELIABLE	Reliability: RELIABLE
Reliability: BEST_EFFORT	Reliability: BEST_EFFORT
Durability: VOLATILE	Durability: VOLATILE
Durability: TRANSIENT_LOCAL	Durability: TRANSIENT_LOCAL
History: KEEP_LAST	History: KEEP_LAST
History: KEEP_ALL	History: KEEP_ALL

ROS2 QoS 配置示例

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

// 创建可靠、持久化的 QoS
rclcpp::QoS reliable_qos(10);
reliable_qos.reliability(RMW_QOS_POLICY_RELIABILITY_RELIABLE);
reliable_qos.durability(RMW_QOS_POLICY_DURABILITY_TRANSIENT_LOCAL);

// 创建发布者
auto publisher = node->create_publisher<std_msgs::msg::String>(
    "topic", reliable_qos);

ROS2 性能优化

1. 选择合适的 RMW

根据应用需求选择合适的 RMW：

Cyclone DDS：轻量级，资源占用小
Fast DDS：功能完整，文档完善
RTI Connext DDS：商业支持，功能最完整

2. 优化 ROS2 QoS

根据数据类型选择合适的 QoS：

传感器数据：BEST_EFFORT，VOLATILE
控制命令：RELIABLE，TRANSIENT_LOCAL
状态信息：RELIABLE，TRANSIENT_LOCAL

3. 使用共享内存传输

对于同一机器上的 ROS2 节点通信，可以使用共享内存传输提高性能（Fast DDS 支持，Cyclone DDS 在 Linux 上支持）。

第七部分：总结与展望

Cyclone DDS 的优势总结

Cyclone DDS 作为轻量级、高性能的 DDS 实现，具有以下主要优势：

1. 轻量级设计

Cyclone DDS 的资源占用小，特别适合资源受限的环境。这使得它可以在嵌入式系统、边缘计算设备和移动机器人平台上高效运行。

2. 高性能实现

尽管是轻量级实现，Cyclone DDS 提供了低延迟、高吞吐量的通信性能，满足实时应用的需求。

3. 完全符合标准

Cyclone DDS 完全实现了 OMG DDS 标准，确保了与其他 DDS 实现的互操作性。

4. 开源免费

使用 Eclipse Public License 2.0 许可证，完全开源免费，适合各种项目。

5. ROS2 集成

作为 ROS2 官方支持的 DDS 实现之一，Cyclone DDS 与 ROS2 无缝集成，为 ROS2 应用提供了高效的通信能力。

Cyclone DDS 的局限性

尽管 Cyclone DDS 具有许多优势，但也存在一些局限性：

1. 文档相对较少

相比 Fast DDS 和 RTI Connext DDS，Cyclone DDS 的文档相对较少，学习曲线可能较陡。

2. 功能相对较少

Cyclone DDS 专注于核心功能，一些高级功能可能不如商业实现完整。

3. Windows 共享内存支持

Cyclone DDS 在 Windows 上不支持共享内存传输，这可能影响同一机器上的通信性能。

4. 社区支持

虽然 Eclipse 基金会提供了良好的支持，但社区规模相对较小，可能影响问题解决速度。

Cyclone DDS 的未来发展

Cyclone DDS 正在不断发展，未来的发展方向包括：

1. 性能优化

持续优化性能，减少延迟，提高吞吐量，降低资源消耗。

2. 功能扩展

扩展功能，包括更好的安全支持、更多的传输选项、更好的工具支持等。

3. 文档改进

改进文档，提供更多的教程和示例代码，降低学习曲线。

4. 社区发展

通过 Eclipse 基金会的支持，吸引更多的贡献者和用户，扩大社区规模。

学习资源推荐

官方文档

Cyclone DDS 官方文档：https://cyclonedds.io/docs/cyclonedds/latest/
GitHub 仓库：https://github.com/eclipse-cyclonedds/cyclonedds
Eclipse IoT 项目：https://iot.eclipse.org/

ROS2 相关资源

ROS2 DDS 实现文档：https://docs.ros.org/en/rolling/Installation/RMW-Implementations/DDS-Implementations/Working-with-Eclipse-CycloneDDS.html
ROS2 社区：https://discourse.ros.org/

示例代码

Cyclone DDS 示例：https://github.com/eclipse-cyclonedds/cyclonedds/tree/master/examples
ROS2 示例：https://github.com/ros2/examples

引言

第一部分：Cyclone DDS 基础概念

什么是 Cyclone DDS

Cyclone DDS 的核心特点

Cyclone DDS 与其他 DDS 实现的对比

第二部分：Cyclone DDS 技术架构

Cyclone DDS 的架构层次

数据为中心的架构

类型系统与序列化

发现机制

第三部分：安装与配置

系统要求

安装方法

ROS2 配置

配置文件

第四部分：使用示例

C API 基础示例

ROS2 使用示例

自定义 QoS 配置

第五部分：性能优化与最佳实践

性能优化技巧

调试与故障排除

最佳实践

第六部分：在 ROS2 中的应用

ROS2 与 Cyclone DDS 的集成

ROS2 QoS 映射

ROS2 性能优化

第七部分：总结与展望

Cyclone DDS 的优势总结

Cyclone DDS 的局限性

Cyclone DDS 的未来发展

学习资源推荐

结语

相关文章

GMR：通用动作重映射器

使用 Dotfile 同步配置

编码器正交编码方法

发表评论

最新评论 (0)

发表评论

评论 (0)

引言

第一部分：Cyclone DDS 基础概念

什么是 Cyclone DDS

Cyclone DDS 的核心特点

Cyclone DDS 与其他 DDS 实现的对比

第二部分：Cyclone DDS 技术架构

Cyclone DDS 的架构层次

数据为中心的架构

类型系统与序列化

发现机制

第三部分：安装与配置

系统要求

安装方法

ROS2 配置

配置文件

第四部分：使用示例

C API 基础示例

ROS2 使用示例

自定义 QoS 配置

第五部分：性能优化与最佳实践

性能优化技巧

调试与故障排除

最佳实践

第六部分：在 ROS2 中的应用

ROS2 与 Cyclone DDS 的集成

ROS2 QoS 映射

ROS2 性能优化

第七部分：总结与展望

Cyclone DDS 的优势总结

Cyclone DDS 的局限性

Cyclone DDS 的未来发展

学习资源推荐

结语

相关文章

GMR：通用动作重映射器

使用 Dotfile 同步配置

编码器正交编码方法

发表评论

最新评论 (0)