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 的核心概念 DomainParticipant：表示 DDS 系统的入口点，属于特定的域。Publisher/Subscriber：分别用于发布和订阅数据。Topic：数据通信的主题，定义通信数据的类型。DataWriter/DataReader：负责数据的写入与读取。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. 创建 DomainParticipant 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; } // 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. 创建 Topic 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; } // 4. 创建 Publisher 和 DataWriter DDSPublisher *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 等级： QoS 0：消息最多发送一次，不保证到达。QoS 1：消息至少发送一次，可能重复。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. 总结对比 特性DDSMQTT架构去中心化（Brokerless）中心化（需要 Broker）实时性高实时性，低延迟针对低带宽、高延迟环境QoS复杂且灵活（22 种 QoS 策略）简单（3 种 QoS 等级）适用场景工业自动化、航空航天、机器人等IoT、智能家居、移动应用等学习曲线较陡，需要较高专业技能简单，快速上手性能高吞吐量、低延迟适合低功耗和高延迟网络

---

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