C++中间件DDS介绍

C++ DDS 库简介

DDS（Data Distribution Service） 是一种用于实时分布式系统通信的中间件标准，由 OMG（Object Management Group） 提出。它是一种发布/订阅（Publish/Subscribe）模式的数据通信框架，广泛应用于嵌入式系统、物联网、航空航天、机器人等领域。

C++ DDS 库是实现 DDS 标准的库，常见的实现包括：

• RTI Connext DDS（Real-Time Innovations 提供）
• OpenDDS（开源实现）
• Eclipse Cyclone DDS（开源实现）
• Fast DDS (以前称为 FastRTPS)（由 eProsima 开发）

这些库提供了丰富的 API，用于创建发布者、订阅者、定义数据类型和 QoS（Quality of Service）等功能。

---

DDS 的核心概念

1. DomainParticipant：表示 DDS 系统的入口点，属于特定的域。
2. Publisher/Subscriber：分别用于发布和订阅数据。
3. Topic：数据通信的主题，定义通信数据的类型。
4. DataWriter/DataReader：负责数据的写入与读取。
5. QoS：定义数据传输的质量服务属性。

---

一个简单的 C++ DDS 代码示例

以下是使用 RTI Connext DDS 的一个简单例子，展示如何发布和订阅一个简单的消息数据类型。请确保你已安装 RTI Connext DDS 或其他 DDS 实现，并正确配置开发环境。

1. 定义数据类型（IDL 文件）

module HelloWorld {struct Msg {long id;string message;};
};

IDL (Interface Definition Language) 文件用于定义消息的数据结构，DDS 会通过工具生成相应的 C++ 类型。

2. 发布者（Publisher）代码

#include <iostream>
#include "HelloWorld/Msg.hpp" // 自动生成的头文件
#include "ndds/ndds_cpp.h"   // RTI Connext DDS 的核心头文件int main() {// 1. 创建 DomainParticipantDDSDomainParticipant *participant = DDSTheParticipantFactory->create_participant(0, DDS_PARTICIPANT_QOS_DEFAULT, nullptr, DDS_STATUS_MASK_NONE);if (participant == nullptr) {std::cerr << "创建 DomainParticipant 失败！" << std::endl;return -1;}// 2. 注册数据类型const char *type_name = HelloWorld::MsgTypeSupport::get_type_name();if (HelloWorld::MsgTypeSupport::register_type(participant, type_name) != DDS_RETCODE_OK) {std::cerr << "注册数据类型失败！" << std::endl;return -1;}// 3. 创建 TopicDDSTopic *topic = participant->create_topic("HelloWorldTopic", type_name, DDS_TOPIC_QOS_DEFAULT, nullptr, DDS_STATUS_MASK_NONE);if (topic == nullptr) {std::cerr << "创建 Topic 失败！" << std::endl;return -1;}// 4. 创建 Publisher 和 DataWriterDDSPublisher *publisher = participant->create_publisher(DDS_PUBLISHER_QOS_DEFAULT, nullptr, DDS_STATUS_MASK_NONE);if (publisher == nullptr) {std::cerr << "创建 Publisher 失败！" << std::endl;return -1;}DDSDataWriter *writer = publisher->create_datawriter(topic, DDS_DATAWRITER_QOS_DEFAULT, nullptr, DDS_STATUS_MASK_NONE);if (writer == nullptr) {std::cerr << "创建 DataWriter 失败！" << std::endl;return -1;}HelloWorld::MsgDataWriter *msg_writer = HelloWorld::MsgDataWriter::narrow(writer);if (msg_writer == nullptr) {std::cerr << "narrow DataWriter 失败！" << std::endl;return -1;}// 5. 发布数据HelloWorld::Msg msg;msg.id = 1;msg.message = DDS_String_dup("Hello, DDS!");if (msg_writer->write(msg, DDS_HANDLE_NIL) != DDS_RETCODE_OK) {std::cerr << "写入数据失败！" << std::endl;} else {std::cout << "消息已发布: " << msg.message << std::endl;}// 清理资源participant->delete_contained_entities();DDSTheParticipantFactory->delete_participant(participant);return 0;
}

---

3. 订阅者（Subscriber）代码

#include <iostream>
#include "HelloWorld/Msg.hpp"
#include "ndds/ndds_cpp.h"class MsgListener : public DDSDataReaderListener {
public:void on_data_available(DDSDataReader *reader) override {HelloWorld::MsgDataReader *msg_reader = HelloWorld::MsgDataReader::narrow(reader);if (msg_reader == nullptr) return;HelloWorld::MsgSeq data_seq;DDS_SampleInfoSeq info_seq;DDS_ReturnCode_t retcode = msg_reader->take(data_seq, info_seq, DDS_LENGTH_UNLIMITED,DDS_ANY_SAMPLE_STATE, DDS_ANY_VIEW_STATE, DDS_ANY_INSTANCE_STATE);if (retcode == DDS_RETCODE_OK) {for (int i = 0; i < data_seq.length(); ++i) {if (info_seq[i].valid_data) {std::cout << "接收到消息: id=" << data_seq[i].id<< ", message=" << data_seq[i].message << std::endl;}}msg_reader->return_loan(data_seq, info_seq);}}
};int main() {DDSDomainParticipant *participant = DDSTheParticipantFactory->create_participant(0, DDS_PARTICIPANT_QOS_DEFAULT, nullptr, DDS_STATUS_MASK_NONE);if (participant == nullptr) {std::cerr << "创建 DomainParticipant 失败！" << std::endl;return -1;}const char *type_name = HelloWorld::MsgTypeSupport::get_type_name();if (HelloWorld::MsgTypeSupport::register_type(participant, type_name) != DDS_RETCODE_OK) {std::cerr << "注册数据类型失败！" << std::endl;return -1;}DDSTopic *topic = participant->create_topic("HelloWorldTopic", type_name, DDS_TOPIC_QOS_DEFAULT, nullptr, DDS_STATUS_MASK_NONE);if (topic == nullptr) {std::cerr << "创建 Topic 失败！" << std::endl;return -1;}DDSSubscriber *subscriber = participant->create_subscriber(DDS_SUBSCRIBER_QOS_DEFAULT, nullptr, DDS_STATUS_MASK_NONE);if (subscriber == nullptr) {std::cerr << "创建 Subscriber 失败！" << std::endl;return -1;}DDSDataReader *reader = subscriber->create_datareader(topic, DDS_DATAREADER_QOS_DEFAULT, new MsgListener(), DDS_STATUS_MASK_ALL);if (reader == nullptr) {std::cerr << "创建 DataReader 失败！" << std::endl;return -1;}std::cout << "等待接收消息..." << std::endl;while (true) {DDS_Duration_t sleep_time = {1, 0};NDDSUtility::sleep(sleep_time);}participant->delete_contained_entities();DDSTheParticipantFactory->delete_participant(participant);return 0;
}

---

DDS和MQTT的区别

DDS（Data Distribution Service） 和 MQTT（Message Queuing Telemetry Transport） 是两种用于分布式系统通信的协议或标准，它们在设计目标、特性和应用场景上有显著的区别。以下从多个方面对比它们：

---

1. 核心概念与设计目标

DDS

• 是一种 实时分布式系统通信中间件标准，由 OMG（Object Management Group） 制定。
• 提供 数据中心式的发布/订阅（Publish/Subscribe）模型，并且支持 点对点通信。
• 设计目标是为 低延迟、高吞吐量、实时性和可靠性 的系统提供支持，主要用于嵌入式系统、工业自动化、国防、航空航天、机器人、汽车等场景。
• 强调 数据优先（Data-Centric），通过 Topic 直接共享数据，并支持复杂的 QoS（Quality of Service） 配置。

MQTT

• 是一种轻量级的 消息队列协议，由 IBM 提出，后成为 OASIS 标准。
• 提供 发布/订阅（Publish/Subscribe）模型，但以 消息传输 为核心。
• 设计目标是为 低带宽、高延迟、不可靠网络（如物联网设备）提供简单可靠的通信方式，广泛应用于物联网 (IoT)、智能家居、移动应用等场景。
• 强调 消息优先（Message-Centric），通过 Broker 中转消息，设备之间并不直接通信。

---

2. 架构对比

DDS

• 分布式架构：DDS 是去中心化的（Brokerless），无需中央服务器。所有参与者（如发布者、订阅者）可以直接通信。
• 数据中心化：基于 Topic 共享数据，支持复杂的数据类型和强大的 QoS 配置。
• 灵活性：支持动态发现（Dynamic Discovery），系统中的节点可以动态加入或离开。

MQTT

• 中心化架构：MQTT 依赖于一个 Broker（中间服务器），所有消息都通过 Broker 转发。
• 消息传输：基于 Topic 组织消息，消息是无状态的，数据类型简单（通常是字符串或 JSON）。
• 设备简单性：客户端实现非常轻量，适用于低功耗设备。

---

3. QoS（服务质量）支持

DDS

• DDS 提供了 22 种 QoS 策略，可以对实时性、可靠性、资源管理、延迟、传输顺序等进行精细控制。例如：
  • Deadline：设置数据传输的时间期限。
  • Latency Budget：定义通信的延迟预算。
  • Reliability：支持可靠传输（可靠模式）或最佳传输（不保证消息到达）。
  • Durability：支持数据的持久化，确保新订阅者可以获得历史数据。
• DDS 的 QoS 机制非常灵活，适用于复杂场景。

MQTT

• MQTT 提供了 3 种简单的 QoS 等级：
  1. QoS 0：消息最多发送一次，不保证到达。
  2. QoS 1：消息至少发送一次，可能重复。
  3. QoS 2：消息保证仅发送一次，且不会重复（最可靠）。
• QoS 机制相对简单，主要适用于低复杂度的 IoT 场景。

---

4. 性能对比

DDS

• 由于 DDS 是去中心化的，通信路径更短，通常具有 低延迟、高吞吐量 的特性。
• 支持 实时性 和 高可靠性，适合对延迟、抖动敏感的场景（如无人机控制、工业机器人）。
• 性能依赖于网络和硬件环境，在高性能网络中表现极为出色。

MQTT

• MQTT 由于采用中心化架构，通信性能受 Broker 服务器 的影响。
• 适合 低带宽网络 和 高延迟环境，但在实时性要求高的场景中表现不如 DDS。
• 消耗资源少，非常适合低功耗设备。

---

5. 使用场景

DDS

• 工业自动化：实时监控和控制（如工厂设备间通信）。
• 航空航天：飞行器传感器数据共享、控制系统。
• 机器人：机器人操作系统（如 ROS 2 的通信层基于 DDS）。
• 汽车：自动驾驶汽车的传感器融合、车辆间通信。
• 国防领域：分布式作战系统、军事仿真。

MQTT

• 物联网 (IoT)：智能家居、智能农业、环境监测。
• 移动应用：即时消息、状态更新。
• 远程监控：设备状态上传、警报通知。
• 能源管理：智能电网监测。
• 智能交通：传感器数据上传、车辆定位。

---

6. 易用性与学习曲线

DDS

• DDS 功能强大，但 API 和配置相对复杂，学习曲线较陡。
• 开发和调试 DDS 应用需要对 QoS 策略和分布式系统有较深入的了解。
• 多数 DDS 实现（如 RTI Connext DDS）提供大量工具和文档，但仍需要较高的专业技能。

MQTT

• MQTT 设计简单，易于上手，开发和部署快速。
• 客户端实现轻量，适合嵌入式和低功耗设备。
• 由于依赖 Broker，部署和管理集中化。

---

7. 开源实现

DDS

• 常见的 DDS 实现包括：
  • RTI Connext DDS（商业版，功能强大）
  • OpenDDS（开源，Apache 2.0 许可证）
  • Eclipse Cyclone DDS（开源，Eclipse 基金会维护）
  • Fast-DDS（以前称为 FastRTPS，开源，eProsima 提供）
• 这些实现通常支持多种编程语言（C++、Java、Python 等）。

MQTT

• 常见的 MQTT Broker：
  • Mosquitto（开源，轻量级）
  • EMQX（开源，高性能，支持分布式部署）
  • HiveMQ（商业版，支持企业级功能）
  • VerneMQ（开源，支持大规模连接）
• MQTT 客户端库也非常丰富，几乎支持所有主流语言。

---

8. 总结对比

特性	DDS	MQTT
架构	去中心化（Brokerless）	中心化（需要 Broker）
实时性	高实时性，低延迟	针对低带宽、高延迟环境
QoS	复杂且灵活（22 种 QoS 策略）	简单（3 种 QoS 等级）
适用场景	工业自动化、航空航天、机器人等	IoT、智能家居、移动应用等
学习曲线	较陡，需要较高专业技能	简单，快速上手
性能	高吞吐量、低延迟	适合低功耗和高延迟网络

---

总结

• 如果你的项目需要 实时性、高可靠性、复杂 QoS，并且运行在高性能网络中，例如工业自动化或机器人控制，DDS 是更好的选择。
• 如果你的项目需要 轻量、低功耗、简单消息传递，运行在低带宽或不可靠的网络中，例如物联网设备或智能家居，MQTT 是更适合的选择。