★ FEATURED ARTICLE 简介面向需要理解实时数据分发与DDS标准的开发者这套OpenDDS示例程序以清晰注释的方式展示数据发布订阅的核心机制。资源基于OMG制定的DDS规范通过DDSDemo工程演示Publisher、Subscriber、Topic、DataWriter/DataReader的协作并涉及QoS策略、IDL接口定义、代码生成、传输层配置和自动发现协议等关键环节适合希望快速上手OpenDDS并建立中间件开发认知的入门及进阶读者。包内共172个文件以CPP源码、H头文件、IDL定义、VC工程文件vcxproj/sln为主辅以构建日志、可执行程序与动态库压缩包约199.85MB。已有3256人学习查看资源在工程结构上按发布端、订阅端、类型支持等模块拆分便于对照阅读。学习者可借助注释与工程示例逐步掌握OpenDDS应用编写、QoS定制及代码生成流程获得从配置到运行的完整参考。1. 别把分布式通信写死OpenDDS示例程序让你第一次跑通发布订阅当进程分布在两台甚至多台机器上互相传一个结构体数据时最常见的做法是写TCP长连接再自定义封包、粘包、心跳协议。一开始能用等订阅端从1个变成5个发布端还得维护所有对端地址代码就开始失控。OpenDDS示例程序要解决的正是这个场景它演示了基于DDS标准的数据分发如何落地用最简单的消息定义加两个可执行程序把“发布端写入数据、订阅端收到数据”整条链路跑通。更难得的是这份示例代码注释写得非常细几乎每一行都在解释“为什么这里要这么调”。适合谁看刚接触DDS、想把第一个发布订阅程序跑起来的开发者以及看过一堆OMG规范术语但不知道从哪下手的人。顺着注释读一遍源码DDS里的Domain、Topic、QoS这些抽象概念都能对应到实际代码行上而不是停留在PPT里。2. 技术栈与架构拆解从IDL契约到数据回路的完整链路2.1 凭什么选OpenDDSDDS开源实现里的“稳重派”DDS不是某个中间件的名字而是一套OMG发布的分布式数据分发规范。市面上开源实现里OpenDDS、FastDDS、CycloneDDS各占一方。OpenDDS的特点是它构建在ACE框架之上发布订阅之外的能力几乎没有重造轮子全部复用ACE的跨平台抽象所以代码结构对搞过多年C服务端的老手来说非常顺手编译时也有一整套MPC工程体系兜底不像某些实现那样需要你额外学一套私有构建系统。我拆这份示例时最大的感受是它没有把“简单”牺牲成“玩具”。示例里发布端、订阅端两个程序完整地用到了类型注册、QoS策略、监听器回调这几个DDS核心机制。你可以先把它当一个最小可运行demo等哪天要上生产再把这里的写法平移到大工程里骨架不用换。2.2 工程骨架示例项目的目录结构与构建体系一份典型的OpenDDS示例工程目录结构通常是这样example/ ├── Messenger.idl # 消息类型定义发布订阅双方共用的“表结构” ├── Publisher/ │ ├── Publisher.cpp # 发布端主程序 │ └── Publisher.mpc # MPC工程描述文件 ├── Subscriber/ │ ├── Subscriber.cpp # 订阅端主程序 │ ├── Subscriber.mpc │ └── DataReaderListener.cpp # 数据到达后的回调处理 └── README.md拿到手先别急着打开IDE先看两个.mpc文件和Messenger.idl。.mpc是MPC构建系统用的工程描述文件声明了源码文件列表、依赖的库和头文件路径。OpenDDS用MPC来生成Makefile或Visual Studio工程好处是Linux和Windows两套构建从同一份描述生成不会出现“Windows能编、Linux编不过”的维护地狱。Messenger.idl则是整套程序的数据契约。DDS里消息类型不是用JSON或protobuf定的而是用IDL语言描述结构体再通过IDL编译器自动生成C类型和序列化代码。后面章节会细看这个文件它决定了发布端能写什么、订阅端能读什么。构建的具体步骤放在第4章讲因为所有OpenDDS程序第一步面对的都是环境变量问题。2.3 数据回路一个消息从写入到被订阅端读完的完整链路光看目录还理解不了DDS。建议你先在脑子里建立这样一条数据链路发布端进程创建 DomainParticipant参与者在某个Domain里创建一个Topic主题再通过Publisher子实体创建DataWriter数据写入器。与此同时订阅端进程在同一个Domain里给同一个Topic创建DataReader数据读取器。两端的QoS策略通过内置的发现协议彼此比对匹配成功后发布端调用write()把数据交给底层的RTPS协议栈数据经过组播或单播送达订阅端进程订阅端的Listener回调被触发里面通过take_next_sample()把数据取出来。这条回路里发布端和订阅端不需要知道彼此的IP和端口不需要先建连接再传数据。中间所有连接管理、断线重连、QoS协商都由OpenDDS透明处理。这也正是DDS和传统RPC或消息队列最大的差异点通信双方只认Topic和数据类型不认地址。示例里注释写得好的地方就是在每个步骤都标了“当前这行代码在链路里对应哪一段”。我读下来的感觉是与其先啃规范不如先在代码里标出这六个实体。3. 核心代码逐段拆解IDL契约、发布端写入、订阅端回调3.1 IDL消息定义用IDL给通信双方立下数据契约先看示例里最核心的文件Messenger.idl#include dds/DdsDcpsInfrastructure.idl #include dds/DdsDcpsPublication.idl #include dds/DdsDcpsSubscription.idl #include dds/DdsDcpsTopic.idl #include dds/DdsDcpsTypeSupportExt.idl module Messenger { topic struct Message { long id; string content; }; };这段IDL定义了一个名为Messenger::Message的结构体整个工程里发布端和订阅端都只认这一个类型。topic注解告诉OpenDDS的IDL编译器这个结构体要作为DDS主题的消息类型使用编译器会为它生成DataWriter、DataReader和TypeSupport相关的代码。定义完IDL后在工程构建时OpenDDS会执行两步代码生成先由TAO的IDL编译器处理这个文件生成C类型基础代码消息结构体的序列化与反序列化逻辑再由OpenDDS自己的IDL编译器生成DDS相关的TypeSupport代码。最终你会得到MessageC.h、MessageC.cpp、MessageTypeSupportImpl.h这类文件。这一步你可以理解成protobuf的proto文件编译生成pb.h/pb.cc但DDS生成的代码里额外包含了RTPS协议传输所需的所有类型元数据。订阅端发现发布端之后通过元数据判断“这个Topic上的数据是什么类型、QoS能不能匹配”所以IDL里多了一个字段发布端和订阅端都必须同步重新生成代码否则两端类型签名不一致数据根本传不起来。3.2 发布端注册类型、创建Writer一行write把数据送出去发布端主程序Publisher.cpp的核心代码大概长这样#include MessageTypeSupportImpl.h // 初始化DDS运行时从命令行参数里读取Domain等配置 DDS::DomainParticipant_var participant TheParticipantFactoryWithArgs(argc, argv)-create_participant( domain_id, PARTICIPANT_QOS_DEFAULT, 0, 0); if (CORBA::is_nil(participant)) { std::cerr create_participant failed std::endl; return 1; } // 注册消息类型告诉DomainParticipant本程序要用Messenger::Message Messenger::MessageTypeSupport_var ts new Messenger::MessageTypeSupportImpl(); ts-register_type(participant, Messenger::Message); // 构建Topic类型签名必须与上面注册时一致 DDS::Topic_var topic participant-create_topic(MessageTopic, Messenger::Message, TOPIC_QOS_DEFAULT, 0, 0); // 创建发布者 DDS::Publisher_var pub participant-create_publisher(PUBLISHER_QOS_DEFAULT, 0, 0); // 创建数据写入器 DDS::DataWriter_var dw pub-create_datawriter(topic, DATAWRITER_QOS_DEFAULT, 0, 0); Messenger::MessageDataWriter_var writer Messenger::MessageDataWriter::_narrow(dw); // 填充数据并写入 Messenger::Message msg; msg.id 1; msg.content Hello, OpenDDS; writer-write(msg, DDS::HANDLE_NIL);这段代码的每一行都和前面说的链路对应。create_participant里的domain_id默认是0只有两端在同一个Domain ID下才会互相发现你可以把它理解为“广播频道号”不同频道的进程即使在同一台机器上也不互通。register_type传入的类型名Messenger::Message必须和create_topic的第二个参数一致这是OpenDDS匹配数据类型的钥匙。发布端和订阅端如果类型名不一致发现阶段就会因为type mismatch而协商失败。重点是最后的write()调用。消息写入后OpenDDS会根据QoS决定投递策略如果是RELIABLE_RELIABILITY_QOS底层会做缓存和重发如果是BEST_EFFORT_RELIABILITY_QOS则直接往网络上丢不做可靠性保证。示例默认用DATAWRITER_QOS_DEFAULT也就是可靠传输。3.3 订阅端Listener回调里把数据take出来订阅端这边关键在Listener实现和主程序的建链流程。先看Listenerclass DataReaderListenerImpl : public DDS::DataReaderListener { void on_data_available(DDS::DataReader_ptr reader) override { // 将通用DataReader指针转成针对Messenger::Message的专用Reader Messenger::MessageDataReader_var dr Messenger::MessageDataReader::_narrow(reader); Messenger::Message msg; DDS::SampleInfo si; // 循环取出所有到达的数据一次取一个样本 while (dr-take_next_sample(msg, si) DDS::RETCODE_OK) { if (si.valid_data) { std::cout Received: id msg.id , content msg.content std::endl; } } } };on_data_available是OpenDDS回调Listener里最常用的方法只要DataReader缓存队列里有新数据到达这个方法就会被触发。注意它是在OpenDDS的接收线程里被调用的所以回调里不要做耗时操作否则会阻塞后续数据的接收这是初学时最容易忽略的事。take_next_sample每次从队列里取一个样本到msg和si中。SampleInfo里的valid_data字段很重要当它等于false说明这条样本表示的不是真实数据而是DDS生命周期事件比如“订阅端知道数据写完了”。示例里用si.valid_data过滤避免把空样本当业务数据处理。主程序里创建订阅端实体时要把这个Listener传给create_datareaderDDS::Subscriber_var sub participant-create_subscriber(SUBSCRIBER_QOS_DEFAULT, 0, 0); DDS::DataReaderListener_var listener new DataReaderListenerImpl(); DDS::DataReader_var dr sub-create_datareader(topic, DATAREADER_QOS_DEFAULT, listener, DDS::DATA_AVAILABLE_STATUS);create_datareader的第三个参数传进Listener实例第四个参数DDS::DATA_AVAILABLE_STATUS声明了“我对哪些状态变化感兴趣”这里表示订阅端要监听数据可用状态。传DDS::ALL_DATA_STATE或省略监听时代码不会报错但数据到了也不会被回调触发这是容易翻车的地方。4. 编译运行与避坑指南环境变量、启动顺序、QoS不匹配全记录4.1 编译环境先把DDS_ROOT和ACE_ROOT这对兄弟安顿好OpenDDS的构建体系依赖ACE和TAO这三者的版本必须匹配否则编译到一半会冒出大量模板报错。拿到示例工程后先确认你的机器上已经有编译好的ACE库再设置环境变量export ACE_ROOT/opt/ACE_wrappers export DDS_ROOT/opt/OpenDDS export LD_LIBRARY_PATH$ACE_ROOT/lib:$DDS_ROOT/lib:$LD_LIBRARY_PATHACE_ROOT指向ACE源码根目录DDS_ROOT指向OpenDDS根目录LD_LIBRARY_PATH让运行时能同时找到libACE.so和libOpenDDS.so。接着在$DDS_ROOT下执行./configure makeconfigure会检查ACE版本、检测编译器是否支持C11并根据检测结果生成MPC所需的配置头文件make会依次构建OpenDDS核心库和内置工具。如果make到了中间某个模块失败八成是ACE版本对不上或者configure时缺了某个系统依赖。示例工程本身也要走MPC流程先到示例目录里生成Makefile再编译cd example $ACE_ROOT/bin/mwc.pl -type make make这里mwc.pl是MPC的命令行入口它根据example.mwc文件里的工程列表生成Makefiles。一条血泪经验如果系统里同时装了多个OpenDDS版本千万别在环境变量上偷懒我遇到过configure时用的DDS_ROOT和mwc.pl时用的不一致最后生成出来的Makefile把两个版本的头文件混在一起编报错位置毫无规律最后花了半天才意识到是两套库在打架。4.2 运行验证先单机跑通再考虑跨机部署OpenDDS示例支持两种运行模式传统DCPSInfoRepo集中发现方式以及现代RTPS分布式发现方式。RTPS模式更接近DDS标准不需要单独启动发现服务适合跨机部署。先看单机RTPS模式怎么跑./publisher -DCPSConfigFile rtps.ini ./subscriber -DCPSConfigFile rtps.ini注意启动顺序先启动订阅端再启动发布端。DDS的发现协议是异步的两端各自通过SPDP协议在网络上发声明。如果发布端先发数据订阅端还没完成发现匹配数据就丢了。示例默认配置下订阅端不缓存错过的数据所以先起subscriber能让数据送达万无一失。rtps.ini的内容通常是[common] DCPSGlobalTransportConfigTheRTPSConfig [transport/the_rtps_transport] transport_typertps_udp use_multicast0use_multicast0表示关闭组播、改用单播。跨机部署时有些网络环境禁用组播这个开关能救你一命。同一台机器上跑两个进程时单播也好使端口由OpenDDS自动分配。要是想用老式的集中发现模式则先启动DCPSInfoRepo再带上它的IOR地址运行两端$DDS_ROOT/bin/DCPSInfoRepo -o repo.ior ./subscriber -DCPSInfoRepo file://repo.ior ./publisher -DCPSInfoRepo file://repo.iorDCPSInfoRepo模式里所有进程先向信息仓库注册Topic再由仓库撮合发布订阅关系。它对网络组播环境更友好但多了一个单点依赖生产上通常用RTPS。4.3 避坑指南常见错误现象、原因与处理这里把拆示例过程中常见的问题按“现象、原因、解决”整理出来。踩坑记录一找不到dds/DdsDcpsInfrastructure.idl或一堆.h文件缺失现象IDL编译或C编译时预处理器报错找不到OpenDDS自带的头文件。原因几乎是百分百的DDS_ROOT或ACE_ROOT没设、设错或者IDL编译器没在PATH里。解决重新检查环境变量确认$DDS_ROOT下存在dds目录并且$DDS_ROOT/bin在PATH里装了两个OpenDDS时确保IDL编译器与库属于同一版本。踩坑记录二订阅端启动后发布端打印“no writer matched”或数据永远收不到现象发布端write()返回成功但订阅端Listener一直没有回调控制台也不打印任何数据。原因分两种一是两端Domain ID不一致二是QoS策略不兼容。常见的是订阅端QoS里把reliability.kind设成了BEST_EFFORT而发布端写的是RELIABLE这两者匹配时会触发on_incompatible_qos回调但不会报异常。解决把两端QoS打一致最省事的方法是用DATAREADER_QOS_DEFAULT和DATAWRITER_QOS_DEFAULT口对口对齐。踩坑记录三改完IDL后重编数据还是按旧格式解析现象给Message结构体加了个新字段发布端和订阅端都重新编译了但订阅端打印出来的还是旧字段。原因OpenDDS在build目录里缓存了IDL生成代码只编Publisher和Subscriber子目录不会触发TypeSupport重新生成。解决删掉生成目录里的MessageC.*和MessageTypeSupport.*或者整体执行一次make clean make。从那以后我每次改完IDL都强制全量编译绝不只编单个子工程。踩坑记录四本机能通两台机器没反应但程序都正常启动现象在虚拟机里单机测试一切正常部署到两套物理机上就你看不见我、我看不见你。原因跨机时双方端口和路由不通RTPS默认使用预留端口进行发现某网段防火墙拦截了UDP组播包。解决在 rtps.ini 里把use_multicast改成0并给rtps_udp配置明确的本机网卡地址和端口范围然后检查防火墙是否放行该UDP端口段。5. 二次开发把示例工程改成你自己的DDS应用5.1 改造思路换消息类型、加Topic、调QoS示例跑通后最自然的改造是换消息类型。在Messenger.idl里把结构体改成你自己的业务模型比如定位系统里传坐标topic struct Position { long device_id; double longitude; double latitude; double altitude; };改完重新走IDL编译流程Publisher和Subscriber里对应的_narrow和take_next_sample代码也替换成新生成类型。这里有个小技巧如果新增字段用基本类型IDL生成代码的改动是纯自动化的不用手写序列化逻辑。如果业务需要多路独立数据流在同一个IDL里加第二个topic结构体再在两端各自创建一个新Topic即可。注意每个Topic必须用独立的const char*主题名类型名可以复用同一个结构体但生产上我一般一个业务流一个结构体避免多个Topic共享类型导致代码难维护。QoS调整是示例里最值得动的一处。比如让订阅端永不离线还能补收历史数据把reader的durability设成TRANSIENT_LOCAL_DURABILITY_QOSDDS::DataReaderQos qos; sub-get_default_datareader_qos(qos); qos.durability.kind DDS::TRANSIENT_LOCAL_DURABILITY_QOS;这个QoS的含义是即便订阅端晚于发布端才加入Domain只要发布端缓存里有历史样本订阅端也能收到。代价是发布端内存开销变大示例里默认不开启适合需要补数据的场景。5.2 验证方法先用QoS测试验证改造确实生效改完别急着看业务数据对不对先用最笨的方法验证链路把发布端content字段改成带自增序号的字符串订阅端打印出来。序号连续说明可靠传输和队列没有丢数据如果发现跳号检查订阅端回调里是不是做了耗时操作。验证QoS是否生效有个很直观的试法先停掉订阅端让发布端连续写几十条数据再启动订阅端。如果开启了TRANSIENT_LOCAL_DURABILITY_QOS订阅端应该能收到启动前发布的缓存数据如果收不到说明有效期策略或缓存深度没配置对。这个方法成本低但能把DDS生命周期和缓存逻辑一次性检验出来。如果你要动手改的是“让示例程序支持多个消息接收者”可以把子目录复制两份修改domain_id与content前缀再在另一台终端启动两个订阅端看看DDS天然支持one-to-many发布端不需要改任何代码。这几步验证下来示例才算真正被你拿下了。希望帮到你。本文还有配套的精品资源点击获取
阅读完成 · 觉得有帮助？
