OpenDDS简介

Don Busch，首席软件工程师兼合作伙伴
Object Computing，Inc.（OCI）

介绍
分布式实时应用程序有时以数据为中心而不是以服务为中心，这意味着分布式系统中参与者的主要目标是分发应用程序数据，而不是访问共享服务。应用程序数据的提供者和/或使用者的集合在设计时可能是未知的，并且可能会在应用程序的整个生命周期内发生变化。通常，以发布/订阅通信模型而不是请求/响应模型最有效地实现以数据为中心的范例。

用于实时系统的OMG 数据分发服务（DDS）解决了以数据为中心的分布式应用程序的性能要求和实时性要求。DDS增加了分布式实时系统开发人员可以使用的发布/订阅选项的范围。为了方便起见，使用OMG接口定义语言（IDL）定义DDS接口。但是，大多数细节留给实现，最重要的是如何在发布者和订阅者之间进行数据传输。DDS实现者决定将底层数据通过TCP，UDP，UDP多播，共享内存等从发布者移动到订阅者的基础通信机制。使用CORBA或IIOP协议不需要DDS规范的实现。将数据从发布者传输到订阅者。

OpenDDS是OMG数据分发服务规范的开源C ++实现。OpenDDS包含基于文件的配置机制。通过配置文件，OpenDDS用户可以配置发布者或订阅者的传输，调试输出，内存分配，DCPSInfoRepo代理进程的位置以及许多其他设置。《OpenDDS开发人员指南》的“配置”一章中介绍了完整的配置设置集。

在本文中，我们涵盖以下主题：

OMG DDS的OpenDDS实现
DDS架构
股票报价员示例
IDL类型
发行人
订户
订户的听众
建立发布者和订阅者
配置股票报价器
通过TCP传输运行股票报价器
通过UDP传输运行股票报价器
摘要
参考资料

OMG DDS的OpenDDS实现

OpenDDS利用可插拔的传输体系结构，通过应用程序开发人员选择的传输和封送实施实现数据传输。从概念上讲，该体系结构借鉴了TAO的可插入协议框架。OpenDDS当前支持TCP和UDP点对点传输以及不可靠和可靠的多播，并使用高性能的封送处理实现。

这种可插拔的传输体系结构允许DDS用户基于所需的传输以及应用程序部署的同质或异质性质来优化DDS安装。可以做出这些选择而不会影响应用程序代码本身。

封送处理代码由专用的OpenDDS IDL编译器生成。单个单独的DCPS信息存储库（DCPSInfoRepo）流程充当中央票据交换所，将发布者和订阅者联系在一起。在幕后，OpenDDS使用CORBA与DCPSInfoRepo流程进行通信， 以关联发布者和订阅者。发布者和订阅者之间的数据传输直接在发布和订阅过程之间进行。OpenDDS为RB和在发送或接收DDS数据时发生的非CORBA I / O创建自己的线程。

DDS架构

OMG数据分发服务规范将DDS分为两个单独的体系结构层。较低的层是以数据为中心的发布和订阅（DCPS）层，其中包含到发布/订阅通信机制的类型安全接口。上层是数据本地重构层（DLRL），它使应用程序开发人员可以在DCPS层之上构建本地对象模型，从而使应用程序免受DCPS知识的影响。每一层都有自己的概念和使用模式集，因此可以分别讨论这两层的概念和术语。

以数据为中心的发布和订阅-DCPS

DCPS层负责有效地将数据从发布者分发到感兴趣的订阅者。它 在发送方使用发布者和数据写入器 ，在接收方使用订户和数据读取器的概念来实现。DCPS层由一个或多个数据域组成，每个数据域都包含一组通过DDS进行通信的参与者（发布者和订阅者）。每个实体（即发布者或订阅者）都属于一个域。每个进程在其所属的每个数据域中都有一个域参与者。

在任何数据域中，数据都是由主题标识的，该主题是特定于类型的域段，允许发布者和订阅者明确地引用数据。在域中，主题将唯一的主题名称，数据类型和一组服务质量（QoS）策略与数据本身相关联。每个主题仅与一种数据类型相关联，尽管许多不同的主题可以发布相同的数据类型。发布者的行为由与特定数据源的发布者，数据写入者和主题元素相关联的QoS策略确定。同样，订户的行为由与订户，数据读取器和特定数据接收器的主题元素相关联的QoS策略确定。

有关DCPS术语的更多信息，请参见《OpenDDS开发人员指南》。

DDS规范定义了许多服务质量（QoS）策略，应用程序可使用这些策略来指定其可靠性，资源使用，容错以及对服务的其他要求。参与者指定他们从服务中需要的行为；服务决定如何实现这些行为。这些策略可以应用于各种DCPS实体（主题，数据写入器，数据读取器，发布者，订阅者和域参与者），尽管并非所有策略对所有类型的实体都有效。

订阅者和发布者通过报价请求范例协作指定QoS策略。发布者向所有订阅者提供一套QoS策略。订户请求其所需的QoS策略集。然后，DDS实现会尝试将请求的策略与提供的策略进行匹配。如果策略一致，则发布和订阅将匹配。

OpenDDS支持全套DCPS服务质量（QoS）策略，包括：

QoS政策	描述
活泼	控制活动性检查，以确保系统中预期的实体仍处于活动状态
可靠性	确定是否允许该服务删除样本
历史	控制实例的值发生变化，然后将其传达给所有订阅服务器，该实例发生了什么情况
资源限制	控制服务可用于满足其他QoS要求的资源

有关服务质量策略的更完整列表和更详细的服务质量定义，请参阅对象管理组的DDS白皮书简介附录A。

数据本地重建层-DLRL

数据本地重建层（DLRL）是DCPS之上的面向对象层。DLRL对象是具有一个或多个共享属性的本机语言（即C ++）对象。每个DLRL类都映射到一个或多个DCPS主题。每个共享属性值都映射到主题数据类型的字段，并且其值通过DCPS分布在整个应用程序中。DLRL参与者通过修改DLRL对象将数据与应用程序的其余部分进行通信，从而发布有关主题的数据样本。DLRL共享属性可以是简单的值或结构，对另一个DLRL对象的引用或这些对象的集合（列表，映射）。DLRL支持复杂的对象图和DLRL对象之间的复杂关系。

开发人员负责确定DCPS实体如何映射到DLRL对象。使用IDL值类型在OMG接口定义语言（IDL）中指定模型。映射在概念上类似于对象关系数据库映射，后者将对象模型映射到关系数据库表。我们认为每个DCPS主题都类似于关系数据库表，每个样本都作为该表中的一行。DDS规范具有从DCPS到DLRL的默认映射。或者，开发人员可以选择通过XML映射文件指定自己的自定义映射。

OpenDDS当前未实现DLRL。

目录

OpenDDS股票报价器示例

我们的示例说明了通过DDS DCPS层发布和订阅数据样本。该示例包含两个DCPS主题，都与股市有关。

股票报价发布者将股票报价样本发布给感兴趣的订阅者；每个报价均包含证券的股票代码，其价值和时间戳。报价在整个交易日中定期发布，因为买卖交易会影响证券的基础价值。另外，证券交易所事件发布者发布与证券交易所有关的重要事件，即，何时交易所开放，关闭，何时暂停交易或恢复交易。

我们的订户同时订阅股票报价和股票交易所事件。订户打印其所看到的每个报价的代码符号和值。当订阅者收到表明当天股票交易所已经关闭的事件时，它将正常关闭。因此，“封闭”证券交易所事件的接收是订户停止预期股票报价样本的信号。

我们将演示如何使用相同的发布者和订阅者代码通过TCP和UDP传输进行通信。传输配置隔离在一组配置文件中，使我们无需更改任何代码即可切换传输。

目录

IDL类型

首先，我们在IDL中定义已发布的DDS数据类型：

#include "orbsvcs/TimeBase.idl"
module StockQuoter
{
#pragma DCPS_DATA_TYPE "StockQuoter::Quote"
#pragma DCPS_DATA_KEY "StockQuoter::Quote ticker"struct Quote {string ticker;string exchange;string full_name;double value;TimeBase::TimeT timestamp;};#pragma DCPS_DATA_TYPE "StockQuoter::ExchangeEvent"
#pragma DCPS_DATA_KEY "StockQuoter::ExchangeEvent exchange"enum ExchangeEventType { TRADING_OPENED,TRADING_CLOSED,TRADING_SUSPENDED,TRADING_RESUMED };struct ExchangeEvent {string exchange;ExchangeEventType event;TimeBase::TimeT timestamp;};
};

我们发布两种数据类型：每个股票报价的报价类型，以及用于指示何时打开，关闭证券交易所以及何时暂停或恢复交易的ExchangeEvent类型。该DCPS_DATA_TYPE编译标记的类型与DDS使用。的 DCPS_DATA_KEY每种类型的定义是针对每个唯一标识符 的实例中的数据类型的。我们报价类型的关键是股票的股票代码。一整天，我们希望为每个股票代号发布许多值或样本。每个股票代号的已发布样本集属于同一实例。在我们的示例中，我们将发布两个股票代号，并因此发布两个实例：SPY（标准普尔存托凭证，即S＆P 500）和MDY（S＆P中盘存托凭证，即S＆P中盘400）。

接下来，我们使用OpenDDS的opendds_idl编译器编译IDL 以生成 类型支持代码。类型支持代码包括生成的DCPS 数据写入器和数据读取器 C ++类以及其他IDL代码。DDS使用类型安全的接口进行发布和订阅。类型安全的接口具有几个优点：首先，在编译时更容易捕获编程错误；第二，当在编译时已知封送数据类型时，可以使生成的封送代码非常高效。第三，我们可以避免any在数据传输中使用低效类型，例如CORBA 。

生成股票报价器的IDL类型的类型支持代码的命令如下：

$ DDS_ROOT / bin / opendds_idl StockQuoter.idl

此命令生成以下文件：

StockQuoterTypeSupport.idl
StockQuoterTypeSupportImpl.h
StockQuoterTypeSupportImpl.cpp

但是，我们不需要opendds_idl手动运行编译器。稍后，我们将使用Make Project Creator（MPC）项目为我们自动化构建步骤。

接下来，我们使用TAO的IDL编译器来编译所有三个IDL文件- StockQuoter.idl我们手动编写的 文件，以及由生成的类型支持文件opendds_idl。

tao_idl -I $ DDS_ROOT -I $ TAO_ROOT / orbsvcs StockQuoter.idl
tao_idl -I $ DDS_ROOT -I $ TAO_ROOT / orbsvcs StockQuoterTypeSupport.idl

目录

发行人

接下来，我们编写一个发布者，以通过DDS发布股票报价和股票交易所事件。首先，我们包括由opendds_idl编译器生成的两个类型支持头文件。

#include "StockQuoterTypeSupportImpl.h"

我们还包括DCPS发布者，服务参与者和QoS标头文件。

#include "dds/DCPS/Service_Participant.h"
#include "dds/DCPS/Marked_Default_Qos.h"
#include "dds/DCPS/PublisherImpl.h"
#include "ace/streams.h"
#include "orbsvcs/Time_Utilities.h"

以下常量用于我们的域，类型名称和主题名称。每种类型均在单独的主题上发布。订户必须为其域，类型名称和主题名称使用相同的值。

// constants for Stock Quoter domain Id, types, and topic
DDS::DomainId_t QUOTER_DOMAIN_ID = 1066;
const char* QUOTER_QUOTE_TYPE = "Quote Type";
const char* QUOTER_QUOTE_TOPIC = "Stock Quotes";
const char* QUOTER_EXCHANGE_EVENT_TYPE = "Exchange Event Type";
const char* QUOTER_EXCHANGE_EVENT_TOPIC = "Stock Exchange Events";

在发布证券交易所事件（即打开，关闭，暂停或恢复）时，我们还将发布该事件适用的证券交易所的名称。

const char* STOCK_EXCHANGE_NAME = "Test Stock Exchange";

这是获取当前日期和时间的简单辅助方法。

TimeBase::TimeT get_timestamp()
{TimeBase::TimeT retval;ACE_hrtime_t t = ACE_OS::gethrtime ();ORBSVCS_Time::hrtime_to_TimeT (retval, t);return retval;
}

发布者的源代码文件的其余部分包含main()。我们输入发布者的main()

int main (int argc, char *argv[])
{DDS::DomainParticipantFactory_var dpf =DDS::DomainParticipantFactory::_nil();DDS::DomainParticipant_var participant =DDS::DomainParticipant::_nil();try{

首先，我们创建一个域参与者。DDS发布者可以在多个独立域上发布，但是我们的示例仅在一个域上发布。我们使用TheDomainParticipantFactoryWithArgs宏将命令行参数传递到DCPS中，并获得单例域参与者工厂。我们使用域参与者的默认服务质量策略为“ Quote”域创建一个域参与者。QUOTER_DOMAIN_ID传递给工厂的值在发布者和订阅者中必须相同。

// Initialize, and create a DomainParticipant

dpf = TheParticipantFactoryWithArgs(argc, argv);

participant = dpf->create_participant(
QUOTER_DOMAIN_ID,
PARTICIPANT_QOS_DEFAULT,
DDS::DomainParticipantListener::_nil());

if (CORBA::is_nil (participant.in ()))
{
cerr << “create_participant failed.” << endl;
ACE_OS::exit(1);
}

然后，我们通过域参与者使用默认的服务质量值创建发布者。PublisherListener 当某些与发布相关的事件发生时，我们可以附加一个DCPS调用的。但是，我们不在乎那些事件，因此我们附加了一个nil侦听器。

  // Create a publisher for the two topics// (PUBLISHER_QOS_DEFAULT is defined in// Marked_Default_Qos.h)DDS::Publisher_var pub =participant->create_publisher(PUBLISHER_QOS_DEFAULT,DDS::PublisherListener::_nil());if (CORBA::is_nil (pub.in ())){cerr << "create_publisher failed." << endl;ACE_OS::exit(1);}

通过DCPS进行发布涉及三个步骤。首先，我们为发布的数据样本注册每种类型。我们的示例发布了两种IDL类型的示例Quote和ExchangeEvent。其次，我们创建一个或多个发布主题。每个主题只能绑定一种类型。因此，我们为两种类型的每种类型创建一个主题。第三，我们为每个主题创建一个数据编写器，并通过该数据编写器发布示例。

我们首先向域参与者注册IDL Quote类型，并为Quote类型传递生成的QuoteTypeSupportImpl类的实例。我们用于报价类型的名称（存储在常量值中QUOTER_QUOTE_TYPE）必须与订阅服务器上使用的名称匹配。创建主题时，我们指定此类型名称，从而使DCPS能够在以后为该主题创建适当类型的数据写入器。

 // Register the Quote typeStockQuoter::QuoteTypeSupport_var quote_servant= new StockQuoter::QuoteTypeSupportImpl();if (DDS::RETCODE_OK !=quote_servant->register_type(participant.in (),QUOTER_QUOTE_TYPE)){cerr << "register_type for " << QUOTER_QUOTE_TYPE<< " failed." << endl;ACE_OS::exit(1);}

然后，我们使用生成的ExchangeEventTypeSupportImpl类以相同的方式向域参与者注册IDL ExchangeEvent类型。我们的DCPS域参与者可以发布有关Quote或ExchangeEvent类型的主题。

 // Register the ExchangeEvent typeStockQuoter::ExchangeEventTypeSupport_var exchange_evt_servant= new StockQuoter::ExchangeEventTypeSupportImpl();if (DDS::RETCODE_OK !=exchange_evt_servant->register_type(participant.in (),QUOTER_EXCHANGE_EVENT_TYPE)){cerr << "register_type for "<< QUOTER_EXCHANGE_EVENT_TYPE<< " failed." << endl;ACE_OS::exit(1);}

我们为报价样本创建一个主题，指示报价数据类型的主题名称和注册名称，并使用默认的服务质量设置。同样，引用主题和类型名称必须在发布者和订阅者上匹配。

// Get QoS to use for our two topics
// Could also use TOPIC_QOS_DEFAULT instead
DDS::TopicQos default_topic_qos;
participant->get_default_topic_qos(default_topic_qos);

// Create a topic for the Quote type…
DDS::Topic_var quote_topic =
participant->create_topic (QUOTER_QUOTE_TOPIC,
QUOTER_QUOTE_TYPE,
default_topic_qos,
DDS::TopicListener::_nil());

if (CORBA::is_nil (quote_topic.in ()))
{
cerr << “create_topic for "
<< QUOTER_QUOTE_TOPIC
<< " failed.” << endl;
ACE_OS::exit(1);
}

同样，我们为ExchangeEvent示例创建一个主题，指示主题名称和ExchangeEvent类型的注册名称，并使用默认的服务质量设置。同样，证券交易所事件主题和类型名称必须在发布者和订阅者上匹配。

  // .. and another topic for the Exchange Event typeDDS::Topic_var exchange_evt_topic =participant->create_topic (QUOTER_EXCHANGE_EVENT_TOPIC,QUOTER_EXCHANGE_EVENT_TYPE,default_topic_qos,DDS::TopicListener::_nil());if (CORBA::is_nil (exchange_evt_topic.in ())){cerr << "create_topic for "<< QUOTER_EXCHANGE_EVENT_TOPIC<< " failed."<< endl;ACE_OS::exit(1);}

我们创建了两个数据编写器，每个主题一个。我们传入上面创建的主题；该主题知道其类型。每个数据编写者都与一个发布者正好相关联，并就一个主题进行发布。后来，我们的发布者通过将数据样本写入每个数据写入器来发布每个主题。以下代码为“股票报价”主题创建数据编写器。

  // Get QoS to use for our two DataWriters// Could also use DATAWRITER_QOS_DEFAULTDDS::DataWriterQos dw_default_qos;pub->get_default_datawriter_qos (dw_default_qos);// Create a DataWriter for the Quote topicDDS::DataWriter_var quote_base_dw =pub->create_datawriter(quote_topic.in (),dw_default_qos,DDS::DataWriterListener::_nil());if (CORBA::is_nil (quote_base_dw.in ())){cerr << "create_datawriter for "<< QUOTER_QUOTE_TOPIC<< " failed." << endl;ACE_OS::exit(1);}StockQuoter::QuoteDataWriter_var quote_dw= StockQuoter::QuoteDataWriter::_narrow(quote_base_dw.in());if (CORBA::is_nil (quote_dw.in ())){cerr << "QuoteDataWriter could not be narrowed"<< endl;ACE_OS::exit(1);}

然后，我们为“证券交易所事件”主题创建一个数据编写器。同样，我们传入上面创建的主题，该主题知道其类型。

 // Create a DataWriter for the Exchange Event topicDDS::DataWriter_var exchange_evt_base_dw =pub->create_datawriter(exchange_evt_topic.in (),dw_default_qos,DDS::DataWriterListener::_nil());if (CORBA::is_nil (exchange_evt_base_dw.in ())){cerr << "create_datawriter for "<< QUOTER_EXCHANGE_EVENT_TOPIC<< " failed." << endl;ACE_OS::exit(1);}StockQuoter::ExchangeEventDataWriter_var exchange_evt_dw =StockQuoter::ExchangeEventDataWriter::_narrow(exchange_evt_base_dw.in());if (CORBA::is_nil (exchange_evt_dw.in ())){cerr << "ExchangeEventDataWriter could not "<< "be narrowed"<< endl;ACE_OS::exit(1);}

我们可以选择注册每个数据实例。注册每个数据实例将在编写该实例的样本时稍微改善延迟。

发布者可以在每个数据实例上发布许多数据样本。数据实例由唯一键标识。对于Quote类型，我们将ticker 其标识为IDL类型定义中的关键字段。具有相同键值的每个Quote数据样本均被视为同一数据实例的一部分。换句话说，在股票代码“ SPY”上发布的每个Quote示例都是同一实例的一部分。

我们有两个Quote实例，分别是股票代码“ SPY”（标准普尔存托凭证，即S＆P 500）和“ MDY”（标准普尔中型存托凭证，即S＆P Midcap 400），以及一个ExchangeEvent实例，用于“测试证券交易所”。 ”。我们向适当的数据写入器注册每个实例。实际调用了注册方法，_cxx_register因为它register是C ++中的保留字。

 // Register the Exchange Event and the two// Quoted securities (SPY and MDY) with the// appropriate data writerStockQuoter::Quote spy;spy.ticker = CORBA::string_dup("SPY");DDS::InstanceHandle_t spy_handle =quote_dw->_cxx_register(spy);StockQuoter::Quote mdy;mdy.ticker = CORBA::string_dup("MDY");DDS::InstanceHandle_t mdy_handle =quote_dw->_cxx_register(mdy);StockQuoter::ExchangeEvent ex_evt;ex_evt.exchange = STOCK_EXCHANGE_NAME;DDS::InstanceHandle_t exchange_handle =exchange_evt_dw->_cxx_register(ex_evt);

最后，我们发布。首先，我们发布TRADING_OPENED有关“证券交易所事件”主题的事件。

  // Publish...StockQuoter::ExchangeEvent opened;opened.exchange = STOCK_EXCHANGE_NAME;opened.event = StockQuoter::TRADING_OPENED;opened.timestamp = get_timestamp();cout << "Publishing TRADING_OPENED" << endl;DDS::ReturnCode_t ret =exchange_evt_dw->write(opened, exchange_handle);if (ret != DDS::RETCODE_OK){ACE_ERROR ((LM_ERROR,ACE_TEXT("(%P|%t)ERROR: OPEN write returned %d.\n"),ret));}

然后，我们在“股票报价”主题上发布“ SPY”和“ MDY”实例的几个股票报价数据样本。我们简单地循环，每次增加一点报价符号的报价，以模拟在真正美好的一天中的活跃交易。

ACE_Time_Value quarterSecond( 0, 250000 );

for ( int i = 0; i < 20; ++i )
{
//
// SPY
//

StockQuoter::Quote spy_quote;
spy_quote.exchange = STOCK_EXCHANGE_NAME;
spy_quote.ticker = CORBA::string_dup(“SPY”);
spy_quote.full_name =
CORBA::string_dup(“S&P Depository Receipts”);
spy_quote.value = 1200.0 + 10.0*i;
spy_quote.timestamp = get_timestamp();

cout << "Publishing SPY Quote: "
<< spy_quote.value << endl;

ret = quote_dw->write(spy_quote, spy_handle);

if (ret != DDS::RETCODE_OK)
{
ACE_ERROR ((
LM_ERROR,
ACE_TEXT("(%P|%t)ERROR: SPY write returned %d.\n"),
ret));
}

ACE_OS::sleep( quarterSecond );

//
// MDY
//

StockQuoter::Quote mdy_quote;
mdy_quote.exchange = STOCK_EXCHANGE_NAME;
mdy_quote.ticker = CORBA::string_dup(“MDY”);
mdy_quote.full_name =
CORBA::string_dup(“S&P Midcap Depository Receipts”);
mdy_quote.value = 1400.0 + 10.0*i;
mdy_quote.timestamp = get_timestamp();

cout << "Publishing MDY Quote: "
<< mdy_quote.value << endl;

ret = quote_dw->write(mdy_quote, mdy_handle);

if (ret != DDS::RETCODE_OK)
{
ACE_ERROR ((
LM_ERROR,
ACE_TEXT("(%P|%t)ERROR: MDY write returned %d.\n"),
ret));
}

ACE_OS::sleep( quarterSecond );
}

最后，我们TRADING_CLOSED在“证券交易所事件”主题上发布事件，以指示该证券交易所当天关闭。

```cppStockQuoter::ExchangeEvent closed;closed.exchange = STOCK_EXCHANGE_NAME;closed.event = StockQuoter::TRADING_CLOSED;closed.timestamp = get_timestamp();cout << "Publishing TRADING_CLOSED" << endl;ret = exchange_evt_dw->write(closed, exchange_handle);if (ret != DDS::RETCODE_OK){ACE_ERROR ((LM_ERROR,ACE_TEXT("(%P|%t)ERROR: CLOSED write returned %d.\n"),ret));}cout << "Exiting..." << endl;} catch (CORBA::Exception& e) {cerr << "Exception caught in main.cpp:" << endl<< e << endl;ACE_OS::exit(1);}
最后，我们在离开之前先清理自己。// Cleanuptry {if (!CORBA::is_nil (participant.in ())) {participant->delete_contained_entities();}if (!CORBA::is_nil (dpf.in ())) {dpf->delete_participant(participant.in ());}} catch (CORBA::Exception& e) {cerr << "Exception caught in cleanup."<< endl<< e << endl;ACE_OS::exit(1);}TheServiceParticipant->shutdown ();return 0;
}

这就完成了发布者的C ++代码。目录## 订户我们的订户订阅股票报价和证券交易所事件，从发布者那里接收数据样本。我们使用发布者的`TRADING_CLOSED`事件来表示当天的交易已经结束，从而触发了订阅者的正常关闭。订户中的许多代码与发布者中的代码相似。我们以与发布者中相同的方式获得域参与者，注册类型等。主要区别在于订阅者是被动的，等待接收样本，而发布者是主动的。订阅服务器使用侦听器对象从发布服务器接收样本。首先，我们包括由dcps_ts.pl 脚本生成的两个类型支持头文件。我们还将这些文件包含在发布者中。但是，我们还包括两个侦听器头文件，每种发布类型一个。当在相关主题上发布数据样本时，DDS会调用侦听器。```cpp
#include "StockQuoterTypeSupportImpl.h"
#include "ExchangeEventDataReaderListenerImpl.h"

我们还包括DCPS订户，服务参与者和QoS标头文件。

#include "dds/DCPS/Service_Participant.h"
#include "dds/DCPS/Marked_Default_Qos.h"
#include "dds/DCPS/SubscriberImpl.h"
#include "dds/DCPS/BuiltinTopicUtils.h"
#include "ace/streams.h"
#include "orbsvcs/Time_Utilities.h"

以下常量用于我们的域，类型名称和主题名称。这些名称必须与发布者使用的域，类型名称和主题名称匹配。

// constants for Stock Quoter domain Id, types, and topic
// (same as publisher)
DDS::DomainId_t QUOTER_DOMAIN_ID = 1066;
const char* QUOTER_QUOTE_TYPE = "Quote Type";
const char* QUOTER_QUOTE_TOPIC = "Stock Quotes";
const char* QUOTER_EXCHANGE_EVENT_TYPE = "Exchange Event Type";
const char* QUOTER_EXCHANGE_EVENT_TOPIC = "Stock Exchange Events";

订户的源代码文件的其余部分包含main()。我们输入订户的main()。

int main (int argc, char *argv[])
{DDS::DomainParticipantFactory_var dpf =DDS::DomainParticipantFactory::_nil();DDS::DomainParticipant_var participant =DDS::DomainParticipant::_nil();try {

就像在发布者中一样，我们创建域参与者。指定的域与发布者的域匹配。

 // Initialize, and create a DomainParticipant// (same code as publisher)dpf = TheParticipantFactoryWithArgs(argc, argv);participant = dpf->create_participant(QUOTER_DOMAIN_ID,PARTICIPANT_QOS_DEFAULT,DDS::DomainParticipantListener::_nil());if (CORBA::is_nil (participant.in ())){cerr << "create_participant failed." << endl;ACE_OS::exit(1);}

然后，我们通过域参与者使用默认的服务质量策略值创建一个订户。SubscriberListener 当某些与订阅相关的事件发生时，我们可以附加一个DCPS调用的。但是，我们不在乎那些事件，因此我们将其设置为零。这几乎与我们致电时在发布商中所做的相同create_publisher。

// Create a subscriber for the two topics
// (SUBSCRIBER_QOS_DEFAULT is defined
// in Marked_Default_Qos.h)
DDS::Subscriber_var sub =
participant->create_subscriber(
SUBSCRIBER_QOS_DEFAULT,
DDS::SubscriberListener::_nil());

if (CORBA::is_nil (sub.in ()))
{
cerr << “create_subscriber failed.” << endl;
ACE_OS::exit(1);
}

与发布者一样，我们必须向域参与者注册IDL Quote和ExchangeEvent类型，才能订阅有关这些类型的主题。

// Register the Quote type
// (same code as publisher)
StockQuoter::QuoteTypeSupport_var quote_servant
= new StockQuoter::QuoteTypeSupportImpl();

if (DDS::RETCODE_OK !=
quote_servant->register_type(participant.in (),
QUOTER_QUOTE_TYPE))
{
cerr << “register_type for " << QUOTER_QUOTE_TYPE
<< " failed.” << endl;
ACE_OS::exit(1);
}

// Register the ExchangeEvent type
// (same code as publisher)
StockQuoter::ExchangeEventTypeSupport_var exchange_evt_servant
= new StockQuoter::ExchangeEventTypeSupportImpl();

if (DDS::RETCODE_OK !=
exchange_evt_servant->register_type(
participant.in (),
QUOTER_EXCHANGE_EVENT_TYPE))
{
cerr << “register_type for "
<< QUOTER_EXCHANGE_EVENT_TYPE
<< " failed.” << endl;
ACE_OS::exit(1);
}

与发布者一样，我们为股票报价创建一个主题，指示主题名称和报价类型的注册名称，并使用默认的服务质量设置。同样，股票报价主题名称必须在发布者和订阅者上匹配。

// Get QoS to use for our two topics
// Could also use TOPIC_QOS_DEFAULT instead
// (same code as publisher)
DDS::TopicQos default_topic_qos;
participant->get_default_topic_qos(default_topic_qos);

// Create a topic for the Quote type…
// (same code as publisher)
DDS::Topic_var quote_topic =
participant->create_topic (QUOTER_QUOTE_TOPIC,
QUOTER_QUOTE_TYPE,
default_topic_qos,
DDS::TopicListener::_nil());

if (CORBA::is_nil (quote_topic.in ()))
{
cerr << “create_topic for "
<< QUOTER_QUOTE_TOPIC
<< " failed.” << endl;
ACE_OS::exit(1);
}

同样，我们为ExchangeEvent示例创建一个主题，指示主题名称和ExchangeEvent类型的注册名称，并使用默认的服务质量设置。同样，证券交易所事件主题名称必须在发布者和订阅者上匹配。

  // .. and another topic for the Exchange Event type// (same code as publisher)DDS::Topic_var exchange_evt_topic =participant->create_topic (QUOTER_EXCHANGE_EVENT_TOPIC,QUOTER_EXCHANGE_EVENT_TYPE,default_topic_qos,DDS::TopicListener::_nil());if (CORBA::is_nil (exchange_evt_topic.in ())){cerr << "create_topic for "<< QUOTER_EXCHANGE_EVENT_TOPIC<< " failed."<< endl;ACE_OS::exit(1);}

在发布者上，我们创建了两个数据编写器，每个主题一个。在订户上，我们将创建两个数据读取器，每个主题一个。每个数据读取器只有一个订户，并订阅一个主题。我们还在每个数据读取器上附加了一个侦听器，以接收发布的数据样本的通知。这是发布者和订阅者代码不同的地方。

以下代码为“股票报价”主题创建一个侦听器。侦听器是一个本地CORBA对象，实现了DDS::DataReaderListenerIDL接口。我们使用OpenDDS的便捷servant_to_reference功能模板来获取接口类型的引用。

 // Create DataReaders and DataReaderListeners for the// Quote and ExchangeEvent// Create a Quote listenerQuoteDataReaderListenerImpl quote_listener_servant;DDS::DataReaderListener_var quote_listener =::OpenDDS::DCPS::servant_to_reference(&quote_listener_servant);if (CORBA::is_nil (quote_listener.in ())){cerr << "Quote listener is nil." << endl;ACE_OS::exit(1);}

我们为“证券交易所事件”主题创建第二个侦听器。

// Create an ExchangeEvent listener
ExchangeEventDataReaderListenerImpl exchange_evt_listener_servant;DDS::DataReaderListener_var exchange_evt_listener =::OpenDDS::DCPS::servant_to_reference(&exchange_evt_listener_servant);if (CORBA::is_nil (exchange_evt_listener.in ()))
{cerr << "ExchangeEvent listener is nil." << endl;ACE_OS::exit(1);
}

最后，我们为两个主题中的每个主题创建一个数据读取器。首先，我们为“股票报价”主题创建一个数据读取器，并附加上面创建的相关侦听器。

 // Create the Quote DataReader// Get the default QoS// Could also use DATAREADER_QOS_DEFAULTDDS::DataReaderQos dr_default_qos;sub->get_default_datareader_qos (dr_default_qos);DDS::DataReader_var quote_dr =sub->create_datareader(quote_topic.in (),dr_default_qos,quote_listener.in ());

然后，我们为“证券交易所事件”主题创建一个数据读取器，并附加上面创建的另一个侦听器。

// Create the ExchangeEvent DataReader

DDS::DataReader_var exchange_evt_dr =
sub->create_datareader(exchange_evt_topic.in (),
dr_default_qos,
exchange_evt_listener.in ());

OpenDDS产生它自己的线程来处理来自发布者的传入事件。因此，订户中没有事件循环代码。但是，在准备关闭整个订户进程之前，我们必须确保不允许主线程退出。因此，我们循环处理股票报价和证券交易所事件，直到TRADING_CLOSED 在“证券交易所事件”主题上收到该事件为止。本质上，我们希望接收已发布的数据样本，直到证券交易所告诉我们它已经关闭。该sleep调用使我们每秒检查一次，以避免消耗过多的CPU。

// Wait for events from the Publisher; shut
// down when “close” received
cout << “Subscriber: waiting for events” << endl;
while ( ! exchange_evt_listener_servant.
is_exchange_closed_received() )
{
ACE_OS::sleep(1);
}

收到TRADING_CLOSED事件后，我们可以正常退出循环。

 cout << "Received CLOSED event from publisher; "<< " exiting..."<< endl;} catch (CORBA::Exception& e) {cerr << "Exception caught in main.cpp:" << endl<< e << endl;ACE_OS::exit(1);}

最后，我们在离开之前先清理自己。

 // Cleanuptry {if (!CORBA::is_nil (participant.in ())) {participant->delete_contained_entities();}if (!CORBA::is_nil (dpf.in ())) {dpf->delete_participant(participant.in ());}} catch (CORBA::Exception& e) {cerr << "Exception caught in cleanup."<< endl<< e << endl;ACE_OS::exit(1);}TheServiceParticipant->shutdown ();return 0;
}

目录

订阅者的“股票报价”和“证券交易所事件”侦听器

“股票行情”数据阅读器和“股票交易所事件”数据阅读器均附带一个侦听器。每当从发布者接收到数据样本时，DDS框架就会调用这些侦听器。我们已决定两个数据读取器中的每一个都应具有自己的侦听器，尽管如果我们对侦听器进行编码以处理两种数据类型，则可以为两个数据读取器使用单个侦听器。

每个侦听器都实现DDS::DataReaderListenerIDL接口。我们在上面的订户代码中同时使用了a QuoteDataReaderListenerImpl和an ExchangeEventDataReaderListenerImpl，但尚未定义这些类。我们现在将这样做。

首先，我们为Quote类型的数据读取器编写一个侦听器。该侦听器类实现DDS::DataReaderListener IDL接口，该接口重写了七个纯虚方法。它是IDL接口的CORBA本地对象实现，继承自IDL接口的生成类。

侦听器类必须重写所有七个方法，包括侦听器的实现为空的方法。但是，为简单起见，我们将仅显示该on_data_available 方法，当有新的Quote数据样本可用时将调用该方法。其他六个方法具有空的实现。我们还将使用默认的构造函数和析构函数。

#include "StockQuoterTypeSupportC.h"
#include "StockQuoterTypeSupportImpl.h"
#include "dds/DCPS/Service_Participant.h"
#include "dds/DdsDcpsSubscriptionS.h"
#include "ace/streams.h"class QuoteDataReaderListenerImpl: public virtual OpenDDS::DCPS::LocalObject<DDS::DataReaderListener>
{
public:// DDS calls on_data_available on the listener for each// received Quote sample.virtual void on_data_available(DDS::DataReader_ptr reader)throw (CORBA::SystemException){try{

我们首先将数据读取器参数的值缩小为Quote样本的适当类型。

StockQuoter::QuoteDataReader_var quote_dr =StockQuoter::QuoteDataReader::_narrow(reader);if (CORBA::is_nil (quote_dr.in ())){cerr << "QuoteDataReaderListenerImpl:: "<< "on_data_available:"<< " _narrow failed." << endl;ACE_OS::exit(1);}

然后，我们从数据读取器中获取下一个Quote示例。请注意QuoteDataReader接口的类型安全。

  StockQuoter::Quote quote;DDS::SampleInfo si;DDS::ReturnCode_t status =quote_dr->take_next_sample(quote, si) ;

收到报价样本后，我们只需打印其内容即可。

if (status == DDS::RETCODE_OK) {cout << "Quote: ticker    = " << quote.ticker.in()<< endl<< "       exchange  = " << quote.exchange.in()<< endl<< "       full name = " << quote.full_name.in()<< endl<< "       value     = " << quote.value<< endl<< "       timestamp = " << quote.timestamp<< endl;cout << "SampleInfo.sample_rank = "<< si.sample_rank << endl;}else if (status == DDS::RETCODE_NO_DATA){cerr << "ERROR: reader received DDS::RETCODE_NO_DATA!"<< endl;}else{cerr << "ERROR: read Quote: Error: "<<  status << endl;}

当报价样本超出范围时，堆栈将清除报价样本的内存。

    } catch (CORBA::Exception& e) {cerr << "Exception caught in read:"<< endl << e << endl;ACE_OS::exit(1);}}

我们没有在DDS::DataReaderListener接口中显示其他方法的实现，但是即使它们的实现为空，我们也必须重写它们。

 // must also override://    on_requested_deadline_missed//    on_requested_incompatible_qos//    on_liveliness_changed//    on_subscription_match//    on_sample_rejected//    on_sample_lost
};

接下来，我们为ExchangeEvent类型的数据读取器编写一个侦听器。基本结构与相同QuoteDataReaderListenerImpl。

#include "ExchangeEventDataReaderListenerImpl.h"
#include "StockQuoterTypeSupportImpl.h"
#include "dds/DCPS/Service_Participant.h"
#include "dds/DdsDcpsSubscriptionS.h"
#include "ace/streams.h"
#include "ace/Synch.h"class ExchangeEventDataReaderListenerImpl: public virtual OpenDDS::DCPS::LocalObject<DDS::DataReaderListener>
{
public:

我们将该is_exchange_closed_received方法添加到数据读取器中，以便订户的主程序可以找出何时TRADING_CLOSED 收到证券交易所事件。此方法在互斥锁的保护下检查布尔值。收到股票交易事件on_data_available时，布尔值由侦听器的方法设置TRADING_CLOSED。

// app-specificCORBA::Boolean is_exchange_closed_received(){ACE_Guard<ACE_Mutex> guard(this->lock_);return this->is_exchange_closed_received_;}

DDS on_data_available在侦听器上为每个接收到的ExchangeEvent示例进行调用。

 virtual void on_data_available(DDS::DataReader_ptr reader)throw (CORBA::SystemException){try{

与中的一样QuoteDataReaderListenerImpl，我们首先将数据读取器参数的值缩小为适当的类型，在这种情况下为ExchangeEventDataReader。

StockQuoter::ExchangeEventDataReader_var exchange_evt_dr =
StockQuoter::ExchangeEventDataReader::_narrow(reader);

if (CORBA::is_nil (exchange_evt_dr.in ())) {
cerr << “ExchangeEventDataReaderListenerImpl:: "
<< “on_data_available:”
<< " _narrow failed.”
<< endl;
ACE_OS::exit(1);
}

然后，我们从数据读取器中获取下一个ExchangeEvent示例。注意类型安全。

StockQuoter::ExchangeEvent exchange_evt;
DDS::SampleInfo si;

DDS::ReturnCode_t status =
exchange_evt_dr->take_next_sample(exchange_evt, si) ;

收到ExchangeEvent示例后，我们只需打印其内容即可。

if (status == DDS::RETCODE_OK) {
cout << "ExchangeEvent: exchange = "
<< exchange_evt.exchange.in() << endl;

switch ( exchange_evt.event ) {case StockQuoter::TRADING_OPENED:cout << "TRADING_OPENED" << endl;break;

收到TRADING_CLOSED事件后，我们将设置一个标志，指示当天该证券交易所已经关闭。

case StockQuoter::TRADING_CLOSED: {
cout << “TRADING_CLOSED” << endl;

    ACE_Guard<ACE_Mutex> guard(this->lock_);this->is_exchange_closed_received_ = 1;break;}case StockQuoter::TRADING_SUSPENDED:cout << "TRADING_SUSPENDED" << endl;break;case StockQuoter::TRADING_RESUMED:cout << "TRADING_RESUMED" << endl;break;default:cerr << "ERROR: reader received unknown "<< "ExchangeEvent: "<< exchange_evt.event<< endl;
}cout << "timestamp = "<< exchange_evt.timestamp<< endl;cout << "SampleInfo.sample_rank = "<< si.sample_rank<< endl;

}
else if (status == DDS::RETCODE_NO_DATA)
{
cerr << "ERROR: reader received "
<< “DDS::RETCODE_NO_DATA!”
<< endl;
}
else
{
cerr << "ERROR: read ExchangeEvent: Error: "
<< status
<< endl;
}

超出范围时，堆栈会清除ExchangeEvent示例。

    } catch (CORBA::Exception& e) {cerr << "Exception caught in read:" << endl<< e << endl;ACE_OS::exit(1);}}// must also override://    on_requested_deadline_missed//    on_requested_incompatible_qos//    on_liveliness_changed//    on_subscription_match//    on_sample_rejected//    on_sample_lost

我们添加了两个私有类属性，以跟踪TRADING_CLOSED事件并使用锁保护该值。

private:CORBA::Boolean is_exchange_closed_received_;ACE_Mutex lock_;
};

这样就完成了订户的C ++代码。

目录

建立发布者和订阅者

我们使用MPC，即Make Project Creator，为发布者和订阅者生成构建文件。MPC提供了一种简单的语法，并且能够为GNU Make，Visual C ++和许多其他构建系统生成构建文件。有关MPC的更多信息，请参见OCI的MPC页面，网址为http://www.objectcomputing.com/products/mpc。

我们创建两个文件来构建我们的股票报价器，一个工作区文件和一个项目文件。我们的工作区文件只是告诉MPC在哪里可以找到MPC dcps和dcpsexe基础项目文件，我们将在以后使用它们。

//
// file StockQuoter.mwc
//workspace {cmdline += -relative DDS_ROOT=$DDS_ROOT
}

接下来，我们创建一个包含三个项目的MPC文件-一个包含IDL和TypeSupport文件的Common项目，一个Publisher，一个Subscriber。这三个项目中的每一个都继承自dcps或的dcpsexe基础项目（位于）$DDS_ROOT。首先，我们创建一个名为StockQuoterCommon的库来保存由TAO IDL和opendds_idl编译器生成的代码。

//
// file StockQuoter.mpc
//project(*Common) : dcps {sharedname = StockQuoterCommonlibout = .includes += $(TAO_ROOT)/orbsvcsidlflags += -I$(TAO_ROOT)/orbsvcsidlflags += -Wb,export_macro=StockQuoterCommon_Exportidlflags += -Wb,export_include=StockQuoterCommon_Export.hdcps_ts_flags += --export=StockQuoterCommon_Exportdynamicflags = STOCKQUOTERCOMMON_BUILD_DLL

一个dcps项目有一个新的部分TypeSupport_Files。本部分执行opendds_idl 脚本以从我们的DDS数据类型生成TypeSupport文件。在这里，我们指示包含DDS数据类型的IDL文件，并指示从中生成的TypeSupport文件。

 TypeSupport_Files {StockQuoter.idl}

我们的IDL_Files部分包含原始IDL文件以及上一部分生成的TypeSupport IDL文件。

IDL_Files {StockQuoterTypeSupport.idlStockQuoter.idl}

在Header_Files和Source_Files部分包含opendds_idl-生成TypeSupport实现文件。MPC会自动添加生成的IDL存根和框架，因此我们不需要手动添加它们。

 Header_Files {StockQuoterTypeSupportImpl.h}Source_Files {StockQuoterTypeSupportImpl.cpp}
}

我们的发布者从上方使用StockQuoterCommon库，并添加publisher.cpp包含发布者的源文件main()。

project(*Publisher) : dcpsexe, svc_utils {after += *Commonexename   = publisherincludes += $(TAO_ROOT)/orbsvcslibs += StockQuoterCommondynamicflags = STOCKQUOTERCOMMON_HAS_DLLTypeSupport_Files {}IDL_Files {}Header_Files {}Source_Files {publisher.cpp}Documentation_Files {README.txtdomain_ids}
}

我们的订户还使用StockQuoterCommon库，添加了一个subscriber.cpp包含订户的main()和两个侦听器的源文件。

project(*Subscriber) : dcpsexe {after += *Commonexename   = subscriberincludes += $(TAO_ROOT)/orbsvcslibs += StockQuoterCommondynamicflags = STOCKQUOTERCOMMON_HAS_DLLTypeSupport_Files {}IDL_Files {}Header_Files {QuoteDataReaderListenerImpl.h}Source_Files {QuoteDataReaderListenerImpl.cppsubscriber.cpp}Documentation_Files {README.txtdomain_ids}
}

我们使用此MPC文件为我们的构建系统生成构建文件。例如，要生成GNU Makefile，我们执行

$ ACE_ROOT / bin / mwc.pl -type gnuace StockQuoter.mwc

为了生成Visual C ++ 7.1解决方案文件，我们执行

perl％ACE_ROOT％/ bin / mwc.pl -type vc71 StockQuoter.mwc

然后，我们构建项目。

目录

配置股票报价器

OpenDDS包含基于文件的配置机制。有了它，OpenDDS用户可以配置发布者或订阅者的传输，DCPSInfoRepo过程的位置以及许多其他设置。配置文件的语法类似于Windows INI文件的语法。它包含几个部分，而这些部分又包含类似属性的条目。基本语法如下：

[section1-name]
Attribute1=value1
Attribute2=value2[section2-name]
Attribute1=value1
Attribute2=value2

《OpenDDS开发人员指南》的“配置”一章中介绍了完整的配置设置集。

我们基于TCP的示例dds_tcp_conf.ini对发布者和订阅者都使用一个配置文件：

dds_tcp_conf.ini # [common] # Debug Level DCPSDebugLevel=0 # IOR of DCPSInfoRepo process. DCPSInfoRepo=corbaloc::localhost:12345/DCPSInfoRepo # Sets the global transport configuration (used by default in the # process to config1, defined below DCPSGlobalTransportConfig=config1 # Transport configuration named config1, contains a single transport # instance named tcp1 (defined below) [config/config1] transports=tcp1 # Transport instance named tcp1, of type "tcp". Uses defaults for # all configuration paramaters. [transport/tcp1] transport_type=tcp

请注意，有三个部分，[common]，[config/config1]，和[transport/tcp1]。本[common] 节包含适用于整个过程的配置值。在此配置文件中，我们指定调试级别，该DCPSInfoRepo过程的对象引用以及全局传输配置。在这里，我们的DCPSInfoRepo 进程正在监听回送（127.0.0.1）接口，这意味着我们已将其配置为仅对在同一主机上运行的DDS进程可用。要使其在网络上可用，请使用IP地址或网络主机名代替localhost。我们已经指定config1 作为我们的全局传输配置，这意味着具有该名称的传输配置将被我们过程中所有未明确指定其他传输配置的读取器和写入器使用。

本[config/config1]节定义了名称为的传输配置config1。该transports选项指定 tcp1为此配置中包括的唯一传输实例。

本[transport/tcp1]节定义了一个名为的传输实例， tcp1并将其传输类型指定为tcp。如OpenDDS文档中所述，此部分还可以用于通过许多配置选项来配置传输。

目录

通过TCP传输运行股票报价器

要运行该示例，我们必须启动一个DCPSInfoRepo过程，并至少启动一个发布者和一个订阅者。要启动DCPSInfoRepo，我们使用以下命令行：

$ DDS_ROOT / bin / DCPSInfoRepo -ORBListenEndpoints iiop：// localhost：12345

我们的DCPSInfoRepo进程侦听端口12345。该端口与我们在DCPSInfoRepo上面的传输配置文件中的对象引用中指定的端口匹配。此DCPSInfoRepo 进程正在监听回送（127.0.0.1）接口，这意味着我们已将其配置为仅对在同一主机上运行的DDS进程可用。同样，要使其在网络上可用，请使用IP地址或网络主机名代替localhost。

我们有两个订阅者和一个发布者：

订户-DCPSConfigFile dds_tcp_conf.ini订户-DCPSConfigFile dds_tcp_conf.ini发布者-DCPSConfigFile dds_tcp_conf.ini

我们使用-DCPSConfigFile命令行参数来指示我们在上面创建的配置文件的名称。请注意，每个订阅者和发布者都使用相同的传输配置文件。

上面的命令行用于运行DCPSInfoRepo，带有内置主题的发布者和订阅者，默认情况下是内置主题。我们也可以关闭内置主题来运行这些过程。该-NOBITS 所使用的DCPSInfoRepo关闭内置的话题和"-DCPSBit 0" 所使用的其它应用程序DDS。命令行如下：

$ DDS_ROOT / bin / DCPSInfoRepo -NOBITS -ORBListenEndpoints iiop：// localhost：12345订户-DCPS位0 -DCPSConfigFile dds_tcp_conf.ini订户-DCPS位0 -DCPSConfigFile dds_tcp_conf.ini发布者-DCPSBit 0 -DCPSConfigFile dds_tcp_conf.ini

发行者为SPY和MDY股票代号发布20个股票报价，每个订阅者都收到它们。发布者完成后，它将发布“ TRADING_CLOSED”消息，这将导致订阅者退出。

目录

通过UDP传输运行股票报价器

我们可以使用相同的代码库，通过简单地运行配置文件来运行UDP传输上的示例，该配置文件定义了一个指定UDP传输实例的全局传输配置。

这是dds_udp_conf.ini文件：

dds_udp_conf.ini # [common] # Debug Level DCPSDebugLevel=0 # IOR of DCPSInfoRepo process. DCPSInfoRepo=corbaloc::localhost:12345/DCPSInfoRepo # Sets the global transport configuration (used by default in the # process to config1, defined below DCPSGlobalTransportConfig=config1 # Transport configuration named config1, contains a single transport # instance named udp1 (defined below) [config/config1] transports=udp1 # Transport instance named udp1, of type "udp". Uses defaults for # all configuration paramaters.[transport/udp1]
transport_type=udp

然后，我们DCPSInfoRepo像以前一样开始该过程：

$ DDS_ROOT / bin / DCPSInfoRepo -ORBListenEndpoints iiop：// localhost：12345

我们使用新的传输配置文件启动两个订户和发布者：

订户-DCPSConfigFile dds_udp_conf.ini订户-DCPSConfigFile dds_udp_conf.ini发布者-DCPSConfigFile dds_udp_conf.ini

我们还可以在关闭内置主题的情况下运行每个流程。命令行如下：

$ DDS_ROOT / bin / DCPSInfoRepo -NOBITS -ORBListenEndpoints iiop：// localhost：12345订户-DCPS位0 -DCPSConfigFile dds_udp_conf.ini订户-DCPS位0 -DCPSConfigFile dds_udp_conf.ini发布者-DCPSBit 0 -DCPSConfigFile dds_udp_conf.ini

和以前一样，发布者为SPY和MDY报价器符号发布20个股票报价，每个订阅者都收到它们。发布者完成后，它将再次发布“ TRADING_CLOSED”消息，这将导致订阅者退出。唯一的区别是我们用UDP传输代替了TCP传输。更改运输方式无需更改代码。

目录

摘要

用于实时系统 的OMG数据分发服务（DDS） 是针对高性能，类型安全，发布和订阅通信中间件的规范。DDS解决了以数据为中心的应用程序，即那些对应用程序数据进行分发非常重要的应用程序。

OpenDDS是OMG数据分发服务规范的一个开源实现，为用户提供了一个有效的发布和订阅框架，并具有开源软件开发模型的优点。

OpenDDS包含基于文件的配置机制。通过配置文件，OpenDDS用户可以配置发布者或订阅者的传输，调试输出，内存分配，DCPSInfoRepo 代理进程的位置以及许多其他设置。在示例中我们已经显示，可以在不进行任何代码更改的情况下换出OpenDDS应用程序的基础传输。

参考资料

示例代码位于示例/ DCPS / IntroductionToOpenDDS的OpenDDS源代码分发中
用于实时系统的OMG数据分发服务（DDS）（https://www.omg.org/spec/DDS/）
OMG DDS门户（https://www.omg.org/spec/DDS/）
OpenDDS主页（http://www.opendds.org）
TAO开发人员指南主页（http://www.theaceorb.com/product/index.html
《 OpenDDS开发人员指南》（http://download.objectcomputing.com/OpenDDS/OpenDDS-latest.pdf）
MPC（https://github.com/objectcomputing/MPC）

OpenDDS官网文献：https://opendds.org/about/articles/Article-Intro.html