DDS通信中間件——DCPS規(guī)范(上)

本篇文章繼續(xù)和大家分享一下對DDS這套規(guī)范的理解。預(yù)期本系列文章將包括以下內(nèi)容陸續(xù)更新：

1. DDS規(guī)范概述

2.DCPS規(guī)范解讀

3. DDS-XTypes與IDL解讀

4. RTPS規(guī)范解讀

5. DDS安全規(guī)范解讀

6. DDS-RPC規(guī)范解讀

7. DDS-TSN規(guī)范解讀

8. DDS-XRCE規(guī)范解讀

DCPS規(guī)范分為上下兩篇，本篇內(nèi)容先講模型概述、接口定義和開發(fā)示例，下篇再來講QoS定義。

1. 概述

DCPS（Data-CentricPublish-Subscribe，以數(shù)據(jù)為中心的發(fā)布/訂閱）規(guī)范是DDS系列規(guī)范最初也是最核心的規(guī)范，在某些場合DDS規(guī)范指的就是DCPS規(guī)范。

2. DCPS-PIM

DCPS的內(nèi)容總結(jié)在下面這張圖中，主要分為三大塊，這篇文章也將按照導(dǎo)圖中的內(nèi)容依次介紹：

• ? 模型描述，介紹模型中涉及的概念；
• ? 接口描述，分模塊定義類并定義接口形式及接口功能描述；
• ? QoS定義，定義22種QoS的功能及其配置方式。

2.1. 模型描述

2.1.1. 實體概念

• 基于主題的發(fā)布/訂閱模型

• 系統(tǒng)的數(shù)據(jù)按照主題來分類；
• 發(fā)布端發(fā)布主題數(shù)據(jù)；
• 訂閱端訂閱主題數(shù)據(jù)；
• 發(fā)布端發(fā)布的主題數(shù)據(jù)后，相同主題的訂閱端都能收到。

• DCPS實體模型定義了三層實體，如下圖所示

• 域參與者（DomainParticipant）

• 這個實體看上去和發(fā)布/訂閱模型沒啥關(guān)系，實際上也沒啥關(guān)系。主要是DDS里面定義域的概念用來隔離通信，相同域內(nèi)實體互相通信，不同域間隔離；
• 這是應(yīng)用參加DDS系統(tǒng)通信的入口實體，創(chuàng)建某個域下的域參與者表示：“我想要參與某個域進行通信（是發(fā)布還是訂閱后面再告訴你）”。

• 發(fā)布者（Publisher）

• 這個最好理解，應(yīng)用創(chuàng)建發(fā)布者表示：“我需要發(fā)布數(shù)據(jù)（具體發(fā)布什么主題數(shù)據(jù)，后面再告訴你）”；

• 訂閱者（Subscriber）
  

• 這個也很好理解，應(yīng)用創(chuàng)建訂閱者表示：“我需要訂閱數(shù)據(jù)（具體訂閱什么主題數(shù)據(jù)，后面再告訴你）”；

• 主題（Topic）

• 主題用來描述一類數(shù)據(jù)，創(chuàng)建主題表示：“我有這么一類數(shù)據(jù)需要交互”；
• 定義主題需要什么信息，后面章節(jié)詳細介紹；

• 數(shù)據(jù)寫者（DataWriter）

• 數(shù)據(jù)寫者關(guān)聯(lián)一個且僅一個主題實體，用來向全局數(shù)據(jù)空間寫指定主題的數(shù)據(jù)；
• 應(yīng)用創(chuàng)建數(shù)據(jù)寫者表示：“我需要發(fā)布XX主題數(shù)據(jù)?！?/li>

• 數(shù)據(jù)讀者（DataReader）

• 數(shù)據(jù)讀者關(guān)聯(lián)一個且僅一個主題實體，用來從全局數(shù)據(jù)空間讀指定主題的數(shù)據(jù)；
• 應(yīng)用創(chuàng)建數(shù)據(jù)讀者表示：“我需要訂閱XX主題數(shù)據(jù)?！?/li>

2.1.2. 數(shù)據(jù)描述

DCPS模型中有幾個重要的概念來描述系統(tǒng)中傳輸?shù)臄?shù)據(jù)，如下圖所示自頂向下分為三層來描述系統(tǒng)中交互的數(shù)據(jù)。

2.2. 接口描述

2.2.1. 基礎(chǔ)模塊

基礎(chǔ)模塊包含了DCPS的接口定義的幾個重要的原則：

實體管理（創(chuàng)建/刪除/查詢）采用工廠模式，父實體為子實體的工廠，層級關(guān)系參見2.2.1節(jié)；

每個實體會關(guān)聯(lián)一組內(nèi)部的狀態(tài)；

狀態(tài)可以通過實體提供的查詢接口查詢獲??；

狀態(tài)可以通過監(jiān)聽器由底層回調(diào)獲取；

狀態(tài)可以通過“條件-等待”的方式同步阻塞獲?。?/p>

每個實體都會關(guān)聯(lián)一系列的QoS配置。

2.2.1.1. 實體狀態(tài)

實體狀態(tài)表示DDS為用戶關(guān)心的底層事件所維護的狀態(tài)，例如數(shù)據(jù)寫者關(guān)聯(lián)成功匹配遠程數(shù)據(jù)讀者、檢測到數(shù)據(jù)讀者數(shù)據(jù)樣本丟失、檢測到主題的類型不兼容等。每個實體會關(guān)聯(lián)一系列的代表該實體“通信狀態(tài)”的狀態(tài)對象。實體關(guān)聯(lián)的狀態(tài)參見下圖。狀態(tài)結(jié)構(gòu)體中包含的數(shù)值可以提供更多關(guān)于該狀態(tài)的信息。

2.2.1.2. 獲取實體狀態(tài)

狀態(tài)分為兩種類型：

• 簡單通信狀態(tài) ：除了表明狀態(tài)是否改變的標識還包含保存當(dāng)前狀態(tài)的對應(yīng)結(jié)構(gòu)體。
• 讀通信狀態(tài)：讀通信狀態(tài)更像一個事件，除了是否發(fā)生以外沒有其他聲明。只有兩個狀態(tài)是讀通信狀態(tài)：#DATA_AVAILABLE_STATUS 和 #DATA_ON_READERS_STATUS 。

下面分別描述獲取實體簡單通信狀態(tài)有三種方式 :

• 同步獲取實體狀態(tài)，參見上圖最左邊圖，用戶想要獲取某個狀態(tài)的當(dāng)前值時，主動調(diào)用該狀態(tài)有關(guān)的實體方法，獲取各實體狀態(tài)的實體方法參見2.2.1.1中的圖；
• 同步等待并獲取實體狀態(tài)，參見上圖中間的圖，用戶設(shè)置等待條件，并調(diào)用DDS接口進行阻塞等待，在DDS底層檢測到該狀態(tài)發(fā)生變化后，解開阻塞，用戶再通過方法1獲取該狀態(tài)；
• 異步回調(diào)實體狀態(tài)，參見上圖右邊的圖，該方式為通過監(jiān)聽器異步回調(diào)狀態(tài) ，用戶為實體設(shè)置相應(yīng)的監(jiān)聽器（再進行其他的業(yè)務(wù)邏輯），DDS底層檢測到該狀態(tài)變化時，通過回調(diào)監(jiān)聽器的相應(yīng)方法，用戶即可獲取該狀態(tài)的值，各實體狀態(tài)關(guān)聯(lián)的回調(diào)方法參見參見2.2.1.1中的圖。

2.2.1.3. 條件-等待

DDS使用條件（ Condition ）以及等待（ WaitSet ）來實現(xiàn)同步等待模型，其中每個條件均有一個觸發(fā)狀態(tài)，不同類型的狀態(tài)的觸發(fā)條件不一樣，可以將一個或者多個條件加入到等待集合中，即允許用戶同時等待多個狀態(tài)的集合中的某個狀態(tài)觸發(fā)。條件的類型及其相關(guān)的操作參見，使用條件等待的詳細例子參見：

（向左滑動）

2.2.1.4. 監(jiān)聽器

DDS通過監(jiān)聽器提供底層狀態(tài)事件異步回調(diào)的機制，每個實體狀態(tài)在相應(yīng)的實體監(jiān)聽器中均有相應(yīng)的回調(diào)函數(shù)與之對應(yīng)，實體回調(diào)函數(shù)與實體狀態(tài)的對應(yīng)關(guān)系參見 2.2.1.1中的圖 。下圖顯示了DDS中各個監(jiān)聽器的繼承關(guān)系，即父實體的監(jiān)聽器繼承于子類的監(jiān)聽器，這就意味著父實體監(jiān)聽器能夠獲取所有子實體的狀態(tài)變化。

2.2.2. 域模塊

在發(fā)布訂閱模型中，理論上每個計算機上的中間件都需要掌握全局的節(jié)點信息，以提供分布式的數(shù)據(jù)傳輸服務(wù)。但在實際應(yīng)用中，可能存在需要隔離的情況，例如網(wǎng)絡(luò)中的節(jié)點數(shù)量過于龐大，掌握全局信息會付出巨大的代價，或者從屬于不同分區(qū)的小組之間不希望互相干擾。為了解決隔離的問題，DDS引入了域的概念。每個參與通信的計算機都可以加入一個或多個域，相同域內(nèi)的節(jié)點可以互相通信，不同域之間的節(jié)點則不會有數(shù)據(jù)交互。DDS中使用 DomainId_t 來唯一標識一個域。域參與者實體是DDS的入口，用戶創(chuàng)建一個域參與者表示該程序想要在指定的域中交互數(shù)據(jù)。域模型示意圖參見下圖，一個應(yīng)用程序可以在多個域中交互數(shù)據(jù)。

2.2.2.1. 域參與者工廠

域參與者實體的工廠為全局單例的域參與者工廠( DomainParticipantFactory )，域參與者工廠的主要功能參見下表。

（向左滑動）

2.2.2.2. 域參與者

域參與者( DomainParticipant )，主要接口的功能參見下表，此外，域參與者內(nèi)部還實現(xiàn)發(fā)現(xiàn)協(xié)議、根據(jù)線程模型配置初始化線程、創(chuàng)建定時器、申請并創(chuàng)建網(wǎng)絡(luò)傳輸需要的資源，并監(jiān)聽對應(yīng)的網(wǎng)絡(luò)端口，負責(zé)處理網(wǎng)絡(luò)報文等，域參與者所需的資源較多，內(nèi)部邏輯復(fù)雜，因而在一個應(yīng)用中應(yīng)盡可能的減少域參與者的創(chuàng)建。

2.2.3. 主題模塊

2.2.4. 發(fā)布模塊

2.2.4.1. 發(fā)布者

2.2.4.2. 數(shù)據(jù)寫者

數(shù)據(jù)寫者主要的功能描述參見下表，數(shù)據(jù)寫者提供強類型安全的接口，該接口由DDS編譯器根據(jù)IDL中定義的類型自動生成，其中類型無關(guān)的接口參見 DataWriter 、類型相關(guān)的接口參見 ExampleDataWriter 接口說明，典型的發(fā)布過程如下 。

（向左滑動）

2.2.5. 訂閱模塊

2.2.5.1. 訂閱者

（向左滑動）

2.2.5.2. 數(shù)據(jù)讀者

數(shù)據(jù)讀者的類型無關(guān)功能描述參見下表，數(shù)據(jù)讀者提供強類型安全的接口，該接口由DDS編譯器根據(jù)IDL中定義的類型自動生成，其中類型無關(guān)的接口參見DataReader 、類型相關(guān)的接口參見 ExampleDataReader 接口說明。

（向左滑動）

2.2.5.3. 獲取主題數(shù)據(jù)

當(dāng)數(shù)據(jù)樣本到達訂閱端時，DDS底層會根據(jù)主題匹配信息分發(fā)給不同的數(shù)據(jù)讀者去處理，當(dāng)完成處理（資源限制等QoS配置）時，數(shù)據(jù)讀者將把該數(shù)據(jù)樣本存儲在底層的隊列中，并通知用戶，再等待用戶通過接口來取出該數(shù)據(jù)樣本，DDS通知用戶數(shù)據(jù)到達的方式有兩種：異步監(jiān)聽方式以及同步等待方式，用戶通過數(shù)據(jù)讀者的讀取數(shù)據(jù)接口訪問底層隊列中的數(shù)據(jù)樣本，這些接口的摘要信息參見下表，詳細信息參見相應(yīng)的接口說明。用戶可以通過多種方式訪問底層存儲的隊列數(shù)據(jù)：

1. 提供樣本深拷貝以及零拷貝（僅提供數(shù)據(jù)樣本在底層隊列中的指針或者引用）兩種方式；

2. 指定最大獲取數(shù)量；

3. 訪問處于指定狀態(tài)的樣本，樣本狀態(tài)由 SampleStateKind 、 ViewStateKind 、 InstanceStateKind 三個狀態(tài)定義；

4. 按照數(shù)據(jù)實例訪問，包括指定數(shù)據(jù)實例以及按數(shù)據(jù)實例順序；

5. 按照樣本的順序依次訪問；

6. 讀取/取出兩種訪問方法，其中讀取操作不從底層隊列中刪除，而取出操作則從底層維護的樣本中刪除。

（向左滑動）

2.3. QoS定義

下一期再繼續(xù)分享。

3. PSM-IDL

因為上面定義的是平臺/技術(shù)無關(guān)的模型，在這一章OMG用IDL給出了DCPS的一種定義，所謂PIM就是只規(guī)定形式和行為，具體數(shù)據(jù)結(jié)構(gòu)在PSM中給出，這是OMG DDS規(guī)范的一般套路，打個比方，在PIM中會定義 InstanceHandle_t 這個結(jié)構(gòu)的含義是用來唯一標識實體或者數(shù)據(jù)實例，但是不會規(guī)定如何來實現(xiàn)，在PSM中會定義 InstanceHandle_t 就是8個字節(jié)或者16個字節(jié)。

4. 開發(fā)流程

5. 代碼示例

5.1. 發(fā)布端

#include"DomainParticipantFactory.h"
#include"DomainParticipant.h"
#include"DefaultQos.h"
#include"Publisher.h"
#include"DataWriter.h"
#include"Topic.h"
#include"Foo.h"
#include"FooDataWriter.h"
#include"FooTypeSupport.h"
#include"ZRSleep.h"
#include
intmain()
{
//域
DomainId_tdomainId=11;
DomainParticipantFactory*factory=DomainParticipantFactory::get_instance();
//域參與者
DomainParticipant*participant=factory->create_participant(
domainId,
DOMAINPARTICIPANT_QOS_DEFAULT,
NULL,STATUS_MASK_NONE);
if(NULL==participant)
{
printf("createparticipantfailed.\n");
return-1;
}
//注冊類型
constChar*type_name=FooTypeSupport::get_instance()->get_type_name();
if(type_name==NULL)
{
return-1;
}
ReturnCode_trtn=FooTypeSupport::get_instance()->register_type(participant,type_name);
if(rtn!=RETCODE_OK)
{
printf("registertypefailed.\n");
return-1;
}
//創(chuàng)建主題
Topic*topic=participant->create_topic("example",type_name,TOPIC_QOS_DEFAULT,NULL,STATUS_MASK_NONE);
if(topic==NULL)
{
printf("createtopicfailed.\n");
return-1;
}
//創(chuàng)建發(fā)布端
Publisher*publisher=participant->create_publisher(
PUBLISHER_QOS_DEFAULT,
NULL,STATUS_MASK_NONE);
if(publisher==NULL)
{
printf("createpublisherfailed.\n");
return-1;
}
//創(chuàng)建數(shù)據(jù)寫者
DataWriterQoswriter_qos;
publisher->get_default_datawriter_qos(writer_qos);
writer_qos.history.depth=5;
DataWriter*writer=publisher->create_datawriter(
topic,writer_qos,NULL,STATUS_MASK_NONE);
FooDataWriter*_writer=dynamic_cast(writer);
if(writer==NULL)
{
printf("createdatawriterfailed.\n");
return-1;
}
//創(chuàng)建數(shù)據(jù)樣本
Foosample;
while(true)
{
//在此處修改數(shù)據(jù)樣本的值

//發(fā)布數(shù)據(jù)樣本
_writer->write(sample,HANDLE_NIL_NATIVE);
ZRSleep(1000);
}
return0;
}

（向左滑動）

5.2. 訂閱端

#include"DomainParticipantFactory.h"
#include"DomainParticipant.h"
#include"DefaultQos.h"
#include"Subscriber.h"
#include"DataReader.h"
#include"Topic.h"
#include"DataReaderListener.h"
#include"Foo.h"
#include"FooDataReader.h"
#include"FooTypeSupport.h"
#include"ZRSleep.h"
#include
//繼承DataReaderListener
classMylistener:publicDataReaderListener
{
//回調(diào)函數(shù)
voidon_data_available(DataReader*the_reader)
{
printf("receiveddata.\n");
FooDataReader*foo_reader=dynamic_cast(the_reader);
if(foo_reader==NULL)
{
printf("castreaderfailed.\n");
return;
}
FooSeqdata_values;
SampleInfoSeqsample_infos;
ReturnCode_trtn;
rtn=foo_reader->take(data_values,
sample_infos,
MAX_INT32_VALUE,
ANY_SAMPLE_STATE,
ANY_VIEW_STATE,
ANY_INSTANCE_STATE);
if(RETCODE_ERROR==rtn)
{
printf("takefailed.\n");
return;
}
if(RETCODE_NO_DATA==rtn)
{
printf("nodata.\n");
return;
}
for(inti=0;ilength();i++)
{
if(sample_infos[i].valid_data)
{
FooPrintData(&data_values[i]);
}
}
foo_reader->return_loan(data_values,sample_infos);
}
};
intmain()
{
//域
DomainId_tdomainId=11;
DomainParticipantFactory*factory=DomainParticipantFactory::get_instance();
//域參與者
DomainParticipant*participant=factory->create_participant(
domainId,
DOMAINPARTICIPANT_QOS_DEFAULT,
NULL,STATUS_MASK_NONE);
if(NULL==participant)
{
printf("createparticipantfailed.\n");
return-1;
}
//注冊類型
constChar*type_name=FooTypeSupport::get_instance()->get_type_name();
if(NULL==type_name)
{
return-1;
}
ReturnCode_trtn=FooTypeSupport::get_instance()->register_type(participant,type_name);
if(RETCODE_OK!=rtn)
{
printf("registertypefailed.\n");
return-1;
}
//創(chuàng)建主題
Topic*topic=participant->create_topic("example",type_name,TOPIC_QOS_DEFAULT,NULL,STATUS_MASK_NONE);
if(NULL==topic)
{
printf("createtopicfailed.\n");
return-1;
}
//創(chuàng)建訂閱端
Subscriber*subscriber=participant->create_subscriber(
SUBSCRIBER_QOS_DEFAULT,
NULL,STATUS_MASK_NONE);
if(NULL==subscriber)
{
printf("createsubscriberfailed.\n");
return-1;
}
//監(jiān)聽器
Mylistener*listener=newMylistener;
//創(chuàng)建數(shù)據(jù)讀者
DataReaderQosreader_qos;
subscriber->get_default_datareader_qos(reader_qos);
reader_qos.history.depth=5;
DataReader*reader=subscriber->create_datareader(
topic,reader_qos,listener,STATUS_MASK_ALL);
FooDataReader*_reader=dynamic_cast(reader);
if(NULL==reader)
{
printf("createdatareaderfailed.\n");
return-1;
}
while(true)
{
ZRSleep(1000);
}
return0;
}

（向左滑動）