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. 创建 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 等级：
- 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. 总结对比
| 特性 | DDS | MQTT |
|---|---|---|
| 架构 | 去中心化（Brokerless） | 中心化（需要 Broker） |
| 实时性 | 高实时性，低延迟 | 针对低带宽、高延迟环境 |
| QoS | 复杂且灵活（22 种 QoS 策略） | 简单（3 种 QoS 等级） |
| 适用场景 | 工业自动化、航空航天、机器人等 | IoT、智能家居、移动应用等 |
| 学习曲线 | 较陡，需要较高专业技能 | 简单，快速上手 |
| 性能 | 高吞吐量、低延迟 | 适合低功耗和高延迟网络 |
总结
- 如果你的项目需要 实时性、高可靠性、复杂 QoS，并且运行在高性能网络中，例如工业自动化或机器人控制，DDS 是更好的选择。
- 如果你的项目需要 轻量、低功耗、简单消息传递，运行在低带宽或不可靠的网络中，例如物联网设备或智能家居，MQTT 是更适合的选择。