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

本篇文章繼續和大家分享一下對DDS這套規范的理解。預期本系列文章將包括以下內容陸續更新：

1. DDS規范概述

2.DCPS規范解讀

3. DDS-XTypes與IDL解讀

4. RTPS規范解讀

5. DDS安全規范解讀

6. DDS-RPC規范解讀

7. DDS-TSN規范解讀

8. DDS-XRCE規范解讀

DCPS規范分為上下兩篇，本篇內容先講模型概述、接口定義和開發示例，下篇再來講QoS定義。

1. 概述

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

2. DCPS-PIM

DCPS的內容總結在下面這張圖中，主要分為三大塊，這篇文章也將按照導圖中的內容依次介紹：

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

2.1. 模型描述

2.1.1. 實體概念

• 基于主題的發布/訂閱模型

• 系統的數據按照主題來分類；
• 發布端發布主題數據；
• 訂閱端訂閱主題數據；
• 發布端發布的主題數據后，相同主題的訂閱端都能收到。

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

• 域參與者（DomainParticipant）

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

• 發布者（Publisher）

• 這個最好理解，應用創建發布者表示：“我需要發布數據（具體發布什么主題數據，后面再告訴你）”；

• 訂閱者（Subscriber）
  

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

• 主題（Topic）

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

• 數據寫者（DataWriter）

• 數據寫者關聯一個且僅一個主題實體，用來向全局數據空間寫指定主題的數據；
• 應用創建數據寫者表示：“我需要發布XX主題數據。”

• 數據讀者（DataReader）

• 數據讀者關聯一個且僅一個主題實體，用來從全局數據空間讀指定主題的數據；
• 應用創建數據讀者表示：“我需要訂閱XX主題數據。”

2.1.2. 數據描述

DCPS模型中有幾個重要的概念來描述系統中傳輸的數據，如下圖所示自頂向下分為三層來描述系統中交互的數據。

2.2. 接口描述

2.2.1. 基礎模塊

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

實體管理（創建/刪除/查詢）采用工廠模式，父實體為子實體的工廠，層級關系參見2.2.1節；

每個實體會關聯一組內部的狀態；

狀態可以通過實體提供的查詢接口查詢獲取；

狀態可以通過監聽器由底層回調獲取；

狀態可以通過“條件-等待”的方式同步阻塞獲取；

每個實體都會關聯一系列的QoS配置。

2.2.1.1. 實體狀態

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

2.2.1.2. 獲取實體狀態

狀態分為兩種類型：

• 簡單通信狀態 ：除了表明狀態是否改變的標識還包含保存當前狀態的對應結構體。
• 讀通信狀態：讀通信狀態更像一個事件，除了是否發生以外沒有其他聲明。只有兩個狀態是讀通信狀態：#DATA_AVAILABLE_STATUS 和 #DATA_ON_READERS_STATUS 。

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

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

2.2.1.3. 條件-等待

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

（向左滑動）

2.2.1.4. 監聽器

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

2.2.2. 域模塊

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

2.2.2.1. 域參與者工廠

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

（向左滑動）

2.2.2.2. 域參與者

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

2.2.3. 主題模塊

2.2.4. 發布模塊

2.2.4.1. 發布者

2.2.4.2. 數據寫者

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

（向左滑動）

2.2.5. 訂閱模塊

2.2.5.1. 訂閱者

（向左滑動）

2.2.5.2. 數據讀者

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

（向左滑動）

2.2.5.3. 獲取主題數據

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

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

2. 指定最大獲取數量；

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

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

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

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

（向左滑動）

2.3. QoS定義

下一期再繼續分享。

3. PSM-IDL

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

4. 開發流程

5. 代碼示例

5.1. 發布端

#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;
}
//創建主題
Topic*topic=participant->create_topic("example",type_name,TOPIC_QOS_DEFAULT,NULL,STATUS_MASK_NONE);
if(topic==NULL)
{
printf("createtopicfailed.\n");
return-1;
}
//創建發布端
Publisher*publisher=participant->create_publisher(
PUBLISHER_QOS_DEFAULT,
NULL,STATUS_MASK_NONE);
if(publisher==NULL)
{
printf("createpublisherfailed.\n");
return-1;
}
//創建數據寫者
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;
}
//創建數據樣本
Foosample;
while(true)
{
//在此處修改數據樣本的值

//發布數據樣本
_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
{
//回調函數
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;
}
//創建主題
Topic*topic=participant->create_topic("example",type_name,TOPIC_QOS_DEFAULT,NULL,STATUS_MASK_NONE);
if(NULL==topic)
{
printf("createtopicfailed.\n");
return-1;
}
//創建訂閱端
Subscriber*subscriber=participant->create_subscriber(
SUBSCRIBER_QOS_DEFAULT,
NULL,STATUS_MASK_NONE);
if(NULL==subscriber)
{
printf("createsubscriberfailed.\n");
return-1;
}
//監聽器
Mylistener*listener=newMylistener;
//創建數據讀者
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;
}

（向左滑動）