简介面向希望在 .NET 环境中使用 OpenDDS 的 C# 开发者这份“opendds订阅发布C#封装1.0v”提供了一套发布/订阅通信的封装示例与可复用组件重点解决 C# 应用接入 DDS 中间件、主题与发布者/订阅者管理、UDP 配置及 QoS 参数调整等问题适合实时数据分发、分布式系统联调和教学实践场景。压缩包共 142 个文件约 12.89MB包含 33 个 cs 源码、23 个 dll 库文件以及 exe、config、ini、csproj、sln、pdb 等工程与运行配置便于直接查看封装结构、引用类库并复现发布订阅流程。已有 1085 人学习读者可从中了解 OpenDDS 的 C# 接口组织、订阅端与发布端模块划分、配置文件作用及常见工程组织方式为二次开发和排错提供参考。1. 上位机要跨机收实时数据为什么最后选的是 OpenDDS 而不是自建 TCP 中转在一个测控类项目上三台设备每秒各推两百帧状态数据上位机要同时订阅并使用这些数据。最早用的是自建 TCP 服务器做中转几个客户端连同一个中心节点跨机延迟和重传全压在那个节点上一忙就积压。后来把链路换成 DDS 的发布订阅模型每个节点自己发现对方自己订阅感兴趣的话题没有中心 broker 做转发延迟从几十毫秒降到个位数。OpenDDS 是这个模型下的一个 C 开源实现代价是接入层只有原生接口而上位机、仿真前端、自动化测试脚本基本都在 .NET 里。这份封装的定位就在这把 DomainParticipant、Publisher/Subscriber、DataWriter/DataReader 这些 DDS 实体包成托管对象让订阅端写起来接近一次普通的事件订阅。做分布式仿真、设备集群监控、多机实时同步又不想把整套技术栈切到 C 的团队可以直接照着用。2. 项目骨架与互操作层从 DomainParticipant 到 DataReader 的映射设计DDS 的概念不算多但每一个在 C# 里都得有对应的落点Participant 是进程级的域参与者Topic 是类型化的话题Publisher/Subscriber 是发布订阅的容器DataWriter/DataReader 才是真正收发数据的两端。这个封装把每个 DDS 实体都映射成一个托管类原生对象用IntPtr句柄持有不跨边界暴露 C 指针。理解这个分层后面看代码就不会迷路。2.1 目录结构与工程划分目录大致是这样拆的原生层、托管层和生成代码三块互相不越界OpenDdsNet/ ├── native/ # C 封装层产出 OpenDdsNet.Native.dll │ ├── include/bridge_api.h # 对外导出的 C 风格接口托管层只认这一份 │ ├── src/ │ │ ├── ParticipantBridge.cpp │ │ ├── WriterBridge.cpp │ │ ├── ReaderBridge.cpp │ │ └── ListenerBridge.cpp │ └── CMakeLists.txt ├── managed/ # C# 托管封装 │ ├── OpenDdsNet.csproj │ ├── Domain/DomainParticipant.cs │ ├── Pub/DataWriter.cs │ ├── Sub/DataReader.cs │ ├── Interop/NativeMethods.cs │ └── Qos/QosProfile.cs ├── generated/ # IDL 编译产出的类型代码 ├── samples/ │ ├── PublisherDemo/ │ └── SubscriberDemo/ └── OpenDdsNet.slnnative/include/bridge_api.h是整个项目的契约文件托管层只依赖它不依赖 OpenDDS 的任何头文件。这种做法看起来多了一层但好处很明显OpenDDS 升级或者换编译参数时托管层的NativeMethods.cs基本不用动改动全部收在 C 那一侧。generated/里的代码是产物不是手写代码建议在 CI 或本地构建脚本里固定生成指令别提交半成品进去。2.2 互操作层选型P/Invoke 而不是 C/CLI托管与原生之间有两种常见做法C/CLI 写桥接程序集或者用 P/Invoke 直接调 C 风格导出函数。这个封装选的是后者理由有三个一是 C/CLI 要求每个目标框架都单独编译一套桥接层.NET Framework 和 .NET 6 得分别处理二是原生侧的 DDS 回调线程模型要自己管用 C 导出函数更容易控制生命周期三是团队里只要有人会写 C 就能改不需要额外熟悉 C/CLI 那套语法。导出声明的写法如下每个函数一一对应bridge_api.h里的extern C// Interop/NativeMethods.cs internal static class NativeMethods { private const string Dll OpenDdsNet.Native; // 创建域参与者返回不透明句柄0 表示失败 [DllImport(Dll, CallingConvention CallingConvention.Cdecl)] internal static extern IntPtr DdsParticipantCreate(int domainId, string qosProfile); // 注册 Topic返回 0 成功重复注册同名但不同类型会返回非 0 [DllImport(Dll, CallingConvention CallingConvention.Cdecl)] internal static extern int DdsTopicRegister( IntPtr participant, string topicName, string typeName); // 创建 DataReadercb 是数据到达回调userCtx 透传回托管侧 [DllImport(Dll, CallingConvention CallingConvention.Cdecl)] internal static extern IntPtr DdsReaderCreate( IntPtr participant, string topicName, DataAvailableCallback cb, IntPtr userCtx); [DllImport(Dll, CallingConvention CallingConvention.Cdecl)] internal static extern void DdsReaderDestroy(IntPtr reader); // 回调委托原生线程直接调用务必保持强引用 [UnmanagedFunctionPointer(CallingConvention.Cdecl)] internal delegate void DataAvailableCallback(IntPtr readerHandle, IntPtr userCtx); }几个参数值得单独拎出来说。CallingConvention.Cdecl必须和 C 侧的导出约定保持一致混用StdCall在 x64 上可能碰巧没事在 x86 上就会直接把栈打歪这类翻车现场非常难查。userCtx这个IntPtr看起来多余实际是托管侧做实例分发的关键同一个回调函数会被多个 Reader 复用只有靠userCtx才能知道这次回调属于哪个托管对象。最要紧的是委托本身——DdsReaderCreate传进去的委托对象如果在托管侧只被局部变量引用GC 随时可能把它回收掉之后原生线程一调就是访问已释放内存进程直接退出没有任何异常可抓。所以托管类里必须留一个字段长期持有它这一点后面 3.2 会展开。2.3 QoS 策略在 C# 里怎么表达DDS 的 QoS 是它的核心能力也是最容易被忽视的部分。原生侧的 QoS 是一堆带默认值的策略结构体托管侧如果直接透传调用方每次都得填几十个字段。这个封装的做法是抽一层DataReaderQos/DataWriterQos只暴露真正高频使用的几项其余走原生默认值。QoS 策略C# 封装字段常用取值实际影响ReliabilityReliabilityReliable / BestEffort是否重传丢失样本实时性优先时用 BestEffortHistoryHistoryDepth8 / 32 / 128缓存里保留多少样本决定突发流量的缓冲深度DurabilityDurabilityVolatile / TransientLocal后启动的订阅端能否拿到历史数据DeadlineDeadline零值或毫秒超时未更新会触发回调常用于在线检测OwnershipOwnershipShared / Exclusive多个发布者同 Topic 时是共享还是独占public sealed class DataReaderQos { public ReliabilityKind Reliability { get; set; } ReliabilityKind.Reliable; public int HistoryDepth { get; set; } 32; public DurabilityKind Durability { get; set; } DurabilityKind.Volatile; public TimeSpan Deadline { get; set; } TimeSpan.Zero; }默认给 Reliable 是为了让第一次跑通的人不至于因为丢包而怀疑人生但要注意它和实时性是一对矛盾Reliable 在链路抖动时会做重传重传期间新样本可能被压在后面。我一般会在跑通之后按场景再调高频状态流用 BestEffort 加小 History关键指令下发才用 Reliable。3. 核心链路拆解一次订阅从 DDS 回调到 C# 事件的完整路径发布订阅看起来是「写进去、读出来」两个动作中间其实横跨了原生线程池、序列化边界和托管对象生命周期三层。这一章按调用顺序把链路走一遍原生监听器收到数据 → 序列化成字节 → 进入托管回调 → 反序列化成 C# 对象 → 触发事件。3.1 原生侧监听器与句柄表OpenDDS 的数据到达通知走的是DataReaderListener::on_data_available这个函数运行在 DDS 的接收线程上。封装层要在监听器里把样本取出来序列化后交给托管回调DDS 类型指针绝不能越界。// ListenerBridge.cpp —— DDS 监听器到函数指针的转接 class ManagedForwardListener : public DDS::DataReaderListener { public: ManagedForwardListener(DataAvailableFn cb, void* ctx) : cb_(cb), ctx_(ctx) {} void on_data_available(DDS::DataReader_ptr reader) override { MyTypeSeq dataSeq; DDS::SampleInfoSeq infoSeq; // take 会把样本从 DDS 缓存移走read 则保留可能重复读到 DDS::ReturnCode_t rc reader-take( dataSeq, infoSeq, DDS::LENGTH_UNLIMITED, DDS::ANY_SAMPLE_STATE, DDS::ANY_VIEW_STATE, DDS::ANY_INSTANCE_STATE); if (rc ! DDS::RETCODE_OK) return; for (CORBA::ULong i 0; i dataSeq.length(); i) { // invalid_data 是 dispose / no_writers 之类的状态样本业务层通常不处理 if (!infoSeq[i].valid_data) continue; cb_(ctx_, static_castvoid*(dataSeq[i])); // 由回调内部完成序列化 } } private: DataAvailableFn cb_; void* ctx_; };第一处要注意的是take和read的区别take取走之后缓存里就没了read会保留下一次回调可能又读到同一批。高频场景下用read容易造成重复消费用take又要注意别在循环中途抛异常否则数据被取走却没处理等于丢样本。第二处是valid_data的判断DDS 除了正常数据还会送 dispose、no_writers 这类实例状态变更不判断的话下游会拿到一堆脏数据。第三处是序列化本身——为什么不直接把DDS::DataReader_ptr传给托管层因为托管层一旦持有原生指针就得自己管理引用计数和析构时机稍不留神就是悬垂指针。把数据在原生侧一次性序列化成字节块边界上只传值是最省心的做法代价是一份拷贝开销。3.2 托管侧Topic、Subscriber、DataReader 三段式放到 C# 这一侧封装的顺序是 Participant → Topic → DataReader每一层都持有上一层的句柄。DataReader 是最关键的一层因为它要同时管三件事句柄销毁、回调委托的强引用、事件的线程语义。public sealed class DataReader : IDisposable { private readonly IntPtr _handle; // 关键必须用字段长期持有否则 GC 回收后原生线程回调会崩 private readonly NativeMethods.DataAvailableCallback _cbRef; public event EventHandlerSampleEventArgs? DataAvailable; public DataReader(IntPtr participant, string topicName, DataReaderQos qos) { _cbRef OnNativeData; // 先赋值再取指针传入 _handle NativeMethods.DdsReaderCreate( participant, topicName, _cbRef, IntPtr.Zero); if (_handle IntPtr.Zero) { // 最常见的原因Topic 没注册或 typeName 和注册时不一致 throw new InvalidOperationException( $DataReader 创建失败topic{topicName}); } } private void OnNativeData(IntPtr readerHandle, IntPtr payload) { var sample SampleCodec.DecodeDeviceState(payload); DataAvailable?.Invoke(this, new SampleEventArgs(sample)); } public void Dispose() NativeMethods.DdsReaderDestroy(_handle); }三个细节决定这段代码稳不稳。第一_cbRef OnNativeData这一行必须在DdsReaderCreate之前执行先拿委托的指针再传过去不能写成DdsReaderCreate(..., OnNativeData, ...)那样委托对象只有栈上的瞬时引用随时可能被回收。第二_handle IntPtr.Zero的检查不是形式主义DDS 创建失败时不会抛 C 异常只会返回空指针不检查就会在后续调用里一路空指针下去。第三DataAvailable?.Invoke是在原生线程上执行的这直接决定了订阅方能不能在里面更新 UI——不能这个问题放到第 5 章细说。3.3 类型生成IDL 到 C# 结构的对齐约定DDS 是强类型的传输协议发布端和订阅端对同一个数据结构的理解必须完全一致包括字段顺序、字段类型、字符串长度和 key 的定义。// DeviceState.idl topic struct DeviceState { key long deviceId; long timestampMs; double temperature; string status; };key标识的是实例键DDS 靠它区分不同的数据实例做 dispose 和实例级 QoS 时都依赖这个字段。C# 侧对应结构必须按同样的顺序声明字符串要处理好编码和长度[Serializable] public sealed class DeviceState { public int DeviceId; // 对应 key参与实例区分 public int TimestampMs; public double Temperature; [MarshalAs(UnmanagedType.LPStr, SizeConst 64)] public string Status; // 长度上限要和 IDL 侧约定一致 }最容易出错的是double和字符串。double的字节序在 x86 和 ARM 上如果序列化层自己实现很容易一边大端一边小端表现出来的现象是数值完全不对却又不是报错。字符串则涉及长度前缀和终止符如果原生侧按定长数组序列化、托管侧按变长读就会读出乱码或者越界。稳妥的做法是优先使用 IDL 生成代码而不是手写一定要手写就用一次 round-trip 单元测试把字段逐一对一遍别等到联调时才发现对不上。4. 跑起来编译顺序、配置文件与双进程验证封装层的构建比普通 C# 库麻烦一点因为它依赖已经编译好的 OpenDDS 原生库。踩过最多的一次坑是只编了托管工程运行时提示找不到OpenDdsNet.Native.dll或者它的依赖OpenDDS_Dcps.dll这两个是不同层级的问题要分开排查。4.1 编译环境与依赖顺序需要的环境大致是Visual Studio 2019 或 2022 带 C 桌面开发工作负载、CMake 3.15 以上、.NET Framework 4.7.2 或 .NET 6、一套编译好的 OpenDDS 原生库。原生库的编译不在这个封装的范围内环境变量OPENDDS_ROOT指向它的安装目录即可。# 1. 配置原生封装层指定 OpenDDS 根目录 cmake -S native -B build/native \ -DOPENDDS_ROOT$OPENDDS_ROOT \ -DCMAKE_BUILD_TYPERelease # 2. 编译原生 DLL产物是 OpenDdsNet.Native.dll 及其依赖 cmake --build build/native --config Release # 3. 先确认原生产物存在再编托管层顺序不能反 dotnet build managed/OpenDdsNet.csproj -c Release编译顺序必须严格是先原生后托管托管工程里只用DllImport声明编译期不校验 DLL 是否存在所以反着来也能编过但一运行就炸。另外OPENDDS_ROOT路径里有空格是常见坑CMake 传进去会被拆成两个参数表现是链接阶段找不到库遇到就从路径上排除空格别去改 CMakeLists 绕。4.2 运行时配置发现模式与网卡绑定DDS 靠发现机制找到对方OpenDDS 支持两种模式一种是中心式的 DCPSInfoRepo另一种是 RTPS 去中心化发现。这个封装默认走 RTPS因为多机部署时不用额外维护一个中心进程少一个单点故障。[domain] id42 # 同一域内的节点才能互相发现多机部署时务必一致 [transport] configrtps [repo] # rtps_discovery 不需要额外启动中心进程 modertps_discoverydomain id不一致是最隐蔽的问题之一两边进程都跑得好好的日志也没有报错就是一条数据都收不到。RTPS 发现默认会监听所有网卡多网卡机器上经常出现「本机互通、跨机不通」原因是它在一个错误的网段上广播。遇到这种情况要显式指定网卡或 IP别依赖自动选择。4.3 最小可运行示例发布端与订阅端发布端把结构体写进 Topic写法接近一次普通的队列写入using var participant new DomainParticipant(42, default); using var writer new DataWriterDeviceState(participant, DeviceState); for (int i 0; i 100; i) { writer.Write(new DeviceState { DeviceId 1, TimestampMs Environment.TickCount, Temperature 36.5 i * 0.1, Status OK }); Thread.Sleep(100); // 模拟设备 10Hz 上报 }订阅端更短注册一个事件然后靠Console.ReadLine()把进程挂住using var participant new DomainParticipant(42, default); using var reader new DataReaderDeviceState(participant, DeviceState); reader.DataAvailable (s, e) Console.WriteLine($[{e.Sample.TimestampMs}] id{e.Sample.DeviceId} $temp{e.Sample.Temperature:F2} {e.Sample.Status}); Console.ReadLine(); // 保持进程存活原生回调线程一直在跑验证顺序有讲究先启动订阅端再启动发布端。默认的 Durability 是 Volatile意思是发布端写出去时如果订阅端还没上线这些样本就永久错过了。这不是 bug是 DDS 的语义设计。要解决就两条路要么用 Volatile 但保证订阅端先起要么把 Durability 改成 TransientLocal 让发布端保留一段历史。前者适合实时状态流后者适合配置下发类的场景按业务选。5. 避坑与排查C# 封装 OpenDDS 时最容易出问题的几处这一章是按真实调试顺序整理的每条都按「现象、原因、解决」写遇到类似症状可以对着看。5.1 程序退出瞬间崩溃异常代码 0xC0000005现象订阅端跑得好好的点关闭或者using作用域结束时进程直接退出没有托管异常Windows 事件日志里是访问冲突。原因有两个一是回调委托被 GC 回收后原生线程还在调用二是释放顺序错了Participant 先于 Reader 销毁Reader 的监听器还在引着已释放的 Participant。解决委托用字段强引用这件事前面说过释放顺序则要反过来——先 Dispose Reader再 Dispose Writer/Subscriber最后才是 Participant。我在封装里给每个类加了_disposed标志位Dispose里先判断避免重复释放引发二次崩溃。5.2 两个进程都正常订阅端一条数据都收不到现象发布端日志显示写入成功订阅端监听器一次都没进。原因按出现频率排domain id 不一致、topic 名大小写不一致、注册时的 typeName 和创建 Reader 时不一致、多网卡绑定到了错误网段、Durability 是 Volatile 且发布端先起。解决先在订阅端注册on_subscription_matched回调并打日志这个回调会在发现对端发布者时触发能看到匹配事件说明发现层是通的问题在数据层看不到说明问题在网络或发现配置。这一步能把排查范围砍掉一半比盲目改代码高效得多。5.3 数据一多WPF 界面就卡死现象低频时界面正常数据频率上去之后界面失去响应或者抛跨线程访问控件的异常。原因DataAvailable事件是在 DDS 的接收线程上触发的直接在里面更新绑定属性就越了 UI 线程。解决不要直接在事件里做 UI 操作用一个有界队列把样本转到 UI 线程。我在项目里用的是ChannelDeviceState加一个 UI 定时器批量取出好处是天然做了削峰界面刷新频率和数据到达频率解耦数据再快界面也不慌。5.4 高频写入时发布端内存持续上涨现象跑十几分钟后发布端内存曲线一路上扬不回落。原因Reliable 模式下 History 深度给太大或者订阅端处理速度跟不上样本在发布端缓存里堆积。解决把 HistoryDepth 调小到业务能容忍的窗口比如 32纯状态同步改成 BestEffort让它自然丢弃旧样本同时在订阅端的回调里避免重活把耗时操作挪到另一个线程处理回调本身只负责入队。5.5 字符串字段读出来是乱码或 double 值明显不对现象不报错但字段值完全对不上字符串像是编码错乱浮点数像被乘了个奇怪的系数。原因两边对序列化格式的理解不一致包括字节序、字符串长度前缀、内存对齐。解决优先用 IDL 生成代码不要两边各写一套必须手写时写一个只依赖字段的 round-trip 测试把结构体在托管侧序列化、原生侧反序列化、再反过来各跑一遍逐字段比对这个测试五分钟能写完能省掉半天联调。6. 二次开发新增自定义 Topic 与回调降载的改造思路封装跑通之后真正的工作量在于加自己的数据类型。这个封装把类型注册收在一个静态入口里扩展的时候只需要两步把 IDL 加进生成列表重新生成然后在启动阶段注册一次。public static class TopicRegistry { private static readonly ConcurrentDictionarystring, Type Map new(); public static void RegisterT(string topicName) where T : class { var typeName typeof(T).Name; // 原生侧需要 typeName 才能把 DDS 类型注册进 Participant var rc NativeMethods.DdsTopicRegister( CurrentParticipant.Handle, topicName, typeName); if (rc ! 0) throw new InvalidOperationException( $Topic 注册失败{topicName}/{typeName}检查 IDL 是否已生成); Map[topicName] typeof(T); } }注意DdsTopicRegister返回非 0 时要立刻抛不能吞掉。最常见的原因是同一个 Participant 上重复注册了同名但不同类型的 Topic这种错误如果被静默忽略后面创建 Reader 时才会以空句柄的形式暴露出来排查路径会长很多。第二处改造是把事件回调改成有界 Channel 做背压。默认的event是同步调用所有订阅者串行执行只要有一个订阅者慢整条接收链路就被拖住。改成每个 Reader 内部维护一个ChannelDeviceState容量给个几百回调里只做TryWrite满了就丢最旧的业务侧另起消费者读。这样数据到达速度和消费速度彻底解耦也顺带解决了 5.3 的线程问题——消费者可以在任意线程上跑需要更新 UI 时再切回去。第三处是验证方法。每次加完新 Topic我会固定跑三轮第一轮同机回环两个进程都用localhost发现验证类型对齐和基础收发第二轮跨机但先用 BestEffort验证网络和发现配置排除重传干扰第三轮才切回 Reliable 加实际频率压测看 History 深度够不够。这个顺序不能颠倒我早年图省事直接从跨机 Reliable 开始调结果类型不匹配、网卡绑定、QoS 配置三个问题一起冒出来根本分不清哪条日志对应哪个原因。从那以后我每次加新类型都强制先把回环那一轮跑绿再往下走慢是慢一点但省下的时间远多于花掉的。希望帮到你。本文还有配套的精品资源点击获取