以数据为中心的实时分发技术研究与应用

胡佳贝，邢 浩，吕睿娟

（中国航空工业集团公司西安航空计算技术研究所，陕西 西安 710068）

0 引 言

近年来，随着分布式信息系统的快速发展，很多领域对于系统的实时性和可靠性提出了更高的要求，尤其是航空领域、作战领域以及航海领域等。对于这些分布式硬实时系统，如果数据的共享和传递不能按照预期到达，则会造成重大灾难或事故。因此，对于实时数据分发系统的研究至关重要。传统的客户/服务器（Client/Server system，C/S）模式以对象为中心，由于其信息传递过程存在耦合度高、扩展性低、效率低等缺陷，已不足以满足当前许多信息系统对于实时性的要求[1]。对象管理组织（Object Management Group，OMG）在以数据为中心的发布/订阅（Data Centric Publish-Subscriber，DCPS）模型基础上，制定了一套数据分发服务（Data Distribute Servic，DDS）标准[2]。该标准可以更好地满足当前对于实时性要求高的系统的需求。

本文首先研究了两种数据分发模型，其次在DDS开源实现OpenDDS的基础上，实现了一个以数据为中心的实时分发系统。该系统可以作为分布式复杂实时分发系统的雏形，在此基础上进行扩展所形成的复杂实时分发系统可以被广泛应用到多个领域。

1 数据分发模型

一般情况下，常用的数据分发模型可以分为传统的C/S和DCPS两种，具体内容如下[3]。

1.1 传统的C/S模型

传统的C/S通信模型如图1所示，可以看到，该模型以处理对象为中心，通常情况下，由客户端和服务器端建立链接，客户端主动发起请求，服务器端处理请求并作出响应[4]。一个客户端可以与多个服务器端进行交互，同时一个服务端可以为多个客户端提供服务。常用的Web服务采用的就是C/S通信模型。

然而，从图1中同样也可以看出，客户端和服务器端存在较高的耦合度，这样会限制整个系统的灵活性和扩展性，而且该模型的通信过程较为复杂，使得信息的传递和共享效率会降低。如果在实时分布式系统中使用此模型，随着运行时间的增加，一旦服务器端连接的客户节点数目增加，就会增加服务器端的工作负荷，这样服务器端很容易出现故障，最终造成整个系统无法运转[5]。因此，该模型还存在单点失效问题，不适用于对实时性和可靠性要求较高的分布式信息系统中。

1.2 DCPS模型

在OMG提出的DDS规范中，将DDS API的接口进行了分层，分别为数据本地重构层（Data Local Reconstruction Layer，DLRL） 和 DCPS[6]。 其 中，DLRL位于DCPS的上层，对于DCPS提供的服务进行了映射和封装，这样便于上层应用程序去使用。

实际上，DDS规范中最重要的内容是DCPS。该模型是一个与平台无关的数据模型，其通信模型如图2所示[7]。主要功能是将发布端发送的信息高效地传递给对该信息感兴趣的订阅端。DCPS构建了一个全局数据空间（Global Data Space，GDS），对于每个分布式应用节点，既可以作为发布者角色往GDS中写数据，也可以作为订阅者角色读取GDS中感兴趣的数据，同时也可以两种角色兼得[8]。每当发布者将数据发送到GDS时，DDS中间件就会快速地将数据传递给对该数据感兴趣的所有订阅者[9]。因此，发布者与订阅者就不需要知道彼此的存在，这便形成了一种低耦合的连接方式，而且任何分布式应用节点可以随时接入到GDS中，从而增加了整个系统的灵活性和扩展性。

DDS规范中包括的实体对象有域参与者（Domain Participant）、主题（Topic）、发布者（Publisher）、订阅者（Subscriber）、数据读者（DataReader）以及数据写者（DataWriter），彼此之间的关系如图3所示。

DDS通过引入域的概念将通信空间划分为不同的平面，处于同一域或者同一个平面内的实体之间才能相互通信，处于不同域或不同平面内的实体之间不能通信。每个域中可以包含多个域参与者，每个域参与者也可以包含多个发布者或多个订阅者，每个发布者可以包含多个数据写者，每个订阅者也可以包含多个数据读者[10]。主题作为发布者和订阅者进行通信的桥梁，并且每个主题关联特定的数据类型。在每一次的发布订阅过程中，由DDS中间件通过检测发布者和订阅者之间的主题是否相匹配，如果匹配，则建立通信链接，反之不会建立链接。同时，在整个DDS的通信过程中，可以根据需要通过相关的QoS来设置数据传输过程中的质量。因此，DPCS保证了正确的数据在正确的时间内到达正确的地方。

2 实时分发系统的设计与实现

以DDS的开源实现OpenDDS为例，在不同操作系统之间设计并实现了一个简单的实时发布系统，设计思路如图4所示。

2.1 发布端程序实现过程

2.1.1 创建域参与者Participant对象

发布端应用程序首先需要创建一个域参与者工厂对象dpf，由域参与者工厂对象dpf调用create_participant()方法创建域参与者对象participant。在本示例中，设置的域ID值为10，所有的DDS实体对象均采用默认的QoS，具体代码为：

DDS::DomainParticipantFactory_var dpf = ThePartici pantFactoryWithArgs(argc, argv);

DDS::DomainParticipant_var participant=dpf->create_participant(10, PARTICIPANT_QOS_DEFAULT,0, OpenDDS::DCPS::DEFAULT_STATUS_MASK);

2.1.2 注册发布的数据类型

接着采用register_type()方法注册要发布的数据类型。此类型是事先定义好的数据结构，具体代码为：

Messenger::MessageTypeSupport_var ts =new Messen ger::MessageTypeSupportImpl;

ts->register_type(participant, “Message”);

2.1.3 创建主题Topic对象

利用域参与者对象participant调用create_topic()方法来创建要发布的主题对象topic，主题对象topic需要与主题名称、数据类型进行关联，具体代码为：

DDS::Topic_var topic =participant->create_topic(“Movie List”,”Message”, TOPIC_QOS_DEFAULT,

0, OpenDDS::DCPS::DEFAULT_STATUS_MASK);

2.1.4 创建发布者Publisher对象和数据写者Writer对象

利用域参与者对象participant调用create_publisher()方法创建发布者对象publisher，由发布者对象publisher调用create_datawriter()方法创建相关的写者对象writer，该对象和之前创建的topic对象进行关联，具体代码为：

DDS::Publisher_var publisher =participant->create_publisher(PUBLISHER_QOS_DEFAULT, 0,OpenDDS::DCPS::DEFAULT_STATUS_MASK);

DDS::DataWriter_var writer =publisher->create_datawriter(topic, DATAWRITER_QOS_DEFAULT, 0,OpenDDS::DCPS::DEFAULT_STATUS_MASK);

Messenger::MessageDataWriter_var message_writer=Messenger::MessageDataWriter::_narrow(writer);

2.2 订阅端程序实现过程

2.2.1 创建域参与者Participant对象

订阅端应用程序首先也需要创建一个域参与者工程对象dpf，由域参与者工厂对象dpf调用create_participant()方法创建域参与者对象participant。在本示例中，设置的域ID值为10，所有的DDS实体对象均采用默认的QoS，具体代码为：

DDS::DomainParticipantFactory_var dpf = ThePartici pantFactoryWithArgs(argc, argv);

DDS::DomainParticipant_var participant=dpf->create_participant(10, PARTICIPANT_QOS_DEFAULT,0, OpenDDS::DCPS::DEFAULT_STATUS_MASK);

2.2.2 注册订阅的数据类型

接着采用register_type()方法注册要订阅的数据类型。此类型是事先定义好的数据结构，具体代码为：

Messenger::MessageTypeSupport_var ts =new Messen ger::MessageTypeSupportImpl;

ts->register_type(participant, “Message”);

2.2.3 创建主题Topic对象

利用域参与者对象participant调用create_topic()方法来创建要发布的主题对象topic，主题对象topic需要与主题名称、数据类型进行关联，具体代码为：

p a r t i c i p a n t->c r e a t e_t o p i c(“M o v i e List”,”Message”, TOPIC_QOS_DEFAULT, 0,OpenDDS::DCPS::DEFAULT_STATUS_MASK);

2.2.4 创建订阅者Subscriber对象

利用域参与者对象participant调用create_subscriber ()方法创建订阅者对象subscriber，具体代码为：

DDS::Subscriber_var subscriber =participant->create_subscriber(SUBSCRIBER_QOS_DEFAULT, 0,OpenDDS::DCPS::DEFAULT_STATUS_MASK);

2.2.5 创建监听器Listener对象和数据读者DataReader对象

由订阅者对象subscriber调用create_datareader()方法创建相关的写者对象reader，该对象和之前创建的topic对象进行关联，同时，在创建写者对象reader对象的需要绑定一个监听器对象listener，以便于对发布端发布的数据进行异步监听和检测，具体代码如下：

DDS::DataReaderListener_var listener(new DataReaderListenerImpl);

DDS::DataReader_var reader =subscriber->create_datareader(topic, reader_qos, listener,OpenDDS::DCPS::DEFAULT_STATUS_MASK);

Messenger::MessageDataReader_var message_reader=Messenger::MessageDataReader::_narrow(reader);

3 结 论

本文首先将传统C/S模型和DCPS模型进行对比，前者具有较高的耦合度与扩展性低等缺点，而DDS规范中的DCPS模型具有耦合度低和灵活性高等优势，可以很好地满足目前分布式系统对实时性与可靠性的需求。其次基于DDS的开源实现OpenDDS，设计并实现了一个简单的实时分发系统，介绍了应用程序的开发过程。后续可以在此基础上进一步扩展，将其应用到更多的领域及更复杂的场景。