我目前用的是基于ros版本的apollo,在apollo內(nèi)部有兩種消息傳輸機制：

• 1.一種是基于ros的回調(diào)函數(shù)的訂閱<詳細介紹>來實現(xiàn)各個模塊之間的數(shù)據(jù)通信；
• 2.還有一種就是事件的通信機制,主要用于感知模塊內(nèi)部的通信,這種通信是基于進程間的內(nèi)存共享來實現(xiàn)消息的傳遞.

下面我主要介紹第二種消息傳遞的方式：
先看一下事件的實際應用:

發(fā)布事件:

void AsyncFusionSubnode::PublishDataAndEvent(const double timestamp, const std::string &device_id,const SharedDataPtr<FusionItem> &data) {CommonSharedDataKey key(timestamp, device_id);bool fusion_succ = fusion_data_->Add(key, data);if (!fusion_succ) {AERROR << "fusion shared data addkey failure";}ADEBUG << "adding key in fusion shared data " << key.ToString();for (size_t idx = 0; idx < pub_meta_events_.size(); ++idx) {const EventMeta &event_meta = pub_meta_events_[idx];Event event;event.event_id = event_meta.event_id;event.timestamp = timestamp;event.reserve = device_id;event_manager_->Publish(event); // 把事件添加進隊列.} }

??????? 事件發(fā)布的函數(shù)中要注意兩點:一是fusion_data_->Add(key,data)就是把要傳遞的信息data與key綁定在一起,然后用fusion_data_添加,第二個是event_manager_->Publish(event)把事件添加進隊列.

先看一下Event的結(jié)構(gòu):

struct Event {EventID event_id = 0;double timestamp = 0.0;std::string reserve;// TODO(Yangguang Li):double local_timestamp = 0.0; // local timestamp to compute process delay.Event() { local_timestamp = TimeUtil::GetCurrentTime(); }std::string to_string() const {std::ostringstream oss;oss << "event_id: " << event_id<< " timestamp: " << GLOG_TIMESTAMP(timestamp)<< " reserve: " << reserve;return oss.str();} };

??????? 仔細觀察會發(fā)現(xiàn)發(fā)布的event并沒有包含任何data的信息,只是實例化了一個event,包含id timestamp device_id,那這個data信息是怎么傳遞出去的呢?
???????再回過頭來看這個與data綁定在一起的key是如何生成的,CommonSharedDataKey key(timestamp, device_id);

struct CommonSharedDataKey {CommonSharedDataKey() = default;CommonSharedDataKey(const double ts, const std::string &id): timestamp(ts), device_id(id) {}virtual std::string ToString() const {return device_id +(boost::format("%ld") %static_cast<int64_t>(timestamp * FLAGS_stamp_enlarge_factor)).str();}double timestamp = 0.0;std::string device_id = ""; }

變量定義：

• device_id: 是std::string類型的變量,某個傳感器的name;
• timespace: 即時間戳。因為每一刻的時間戳都不同,所以不同時刻發(fā)出去的事件key都是獨一無二的.

看到這里是不是有點明白了：

• 1.data本身沒有傳遞,只是這個data根據(jù)device_id和時間戳生成了一個只屬于這個data的鑰匙key；
• 2.然后把這個能生成key的device_id和時間戳添加到對應的事件隊列中；
• 3.接收的時候先按id接受event,然后獲取事件中的device_id和時間戳來生成key,然后通過key拿到data.

---

我們再繼續(xù)深入,考慮一個問題：

?????? 一個節(jié)點會一直不停的發(fā)布事件,這么多事件是如何有序的被另一個節(jié)點接受然后解析,并且被解析過后的事件是如何處理的?

上文提到過一個東西"事件隊列",

using EventQueue = FixedSizeConQueue<Event>;

?????? 感知內(nèi)部會每一個event安排一個queue,這個隊列的最大size是5(可以考慮一下為什么這個size要設(shè)置成5),當然了這個隊列是自己重新改造的符合線程安全的隊列,那么我們總共有多少個事件的? 接觸過apollo的應該更容易理解,apollo有一個config文件,這個文件中每一個edges就是一個事件,感知這個模塊其實就是一個有向無環(huán)圖,串聯(lián)這些節(jié)點的就是這些edges,我們通過讀取config文件來初始化這個各個節(jié)點,各個邊,各個sharedData,在初始化edges時,會初始化一個map,如下:

EventQueueMap event_queue_map_;
這個map的key值是事件ID,value值是一個事件隊列,先根據(jù)每一個事件ID為每個事件new一個專屬隊列,隊列最大size是5 ：

event_queue_map_[event_pb.id()].reset(new EventQueue(5));

事件隊列準備好之后,接下來就是要使用它了。

• 發(fā)布事件就是按照事件id往這個隊列push元素；
• 訂閱事件就是從這個隊列pop元素。

這樣一來,事件通信機制的基本框架就搭好了。

訂閱事件:

bool AsyncFusionSubnode::SubscribeEvents(const EventMeta &event_meta,std::vector<Event> *events) const {Event event;// blocking call for each of these eventswhile (event_manager_->Subscribe(event_meta.event_id, &event, true)) {ADEBUG << "starting subscribing event " << event_meta.event_id;// events->push_back(event);}// only obtain latest event from a sensor queueif (event.event_id != 0 && event.timestamp != 0.0) {events->push_back(event);}return true; }

搞懂了事件之后,要明白就兩個節(jié)點之間通過事件通信還要明白另一個東西,sharedData
??????看sharedData之前,先理一下總體思路:按照我們之前的分析,A節(jié)點需要發(fā)布信息,它不會把這個數(shù)據(jù)量巨大結(jié)構(gòu)體發(fā)給另一個節(jié)點,而是通過底層的事件隊列來完成的.明確我們要發(fā)出的信息,一個事件ID,一個時間戳,一個Name,發(fā)出的同時用時間戳和Name生成一個唯一Key來指向這個數(shù)據(jù)量巨大的結(jié)構(gòu)體.那么節(jié)點B在接受信息的時候只要能復現(xiàn)這個key,是不是就能拿到這個數(shù)據(jù)量巨大的結(jié)構(gòu)體了。

---

明白了總體思路,就來看一下具體的實現(xiàn):

這個是我們在發(fā)布信息的時候把data跟生成的key綁定在一起：

bool fusion_succ = fusion_data_->Add(key, data);

template <class M> bool CommonSharedData<M>::Add(const std::string &key,const SharedDataPtr<M> &data) {MutexLock lock(&mutex_);auto ret = data_map_.emplace(SharedDataPair(key, data));if (!ret.second) {AWARN << "Duplicate key: " << key;return false;}const uint64_t timestamp = ::time(NULL);data_added_time_map_.emplace(DataKeyTimestampPair(key, timestamp));++stat_.add_cnt;return true; }template <class M> bool CommonSharedData<M>::Add(const CommonSharedDataKey &key,const SharedDataPtr<M> &data) {// update latest_timestamp for SharedDatalatest_timestamp_ = key.timestamp;return Add(key.ToString(), data); }

????? 其實綁定也很簡單,就是在common_shared_data.cc中維護了一個map(typedef std::unordered_map<std::string, SharedDataPtr<M>> SharedDataMap;), 它的 key值是string,value值是一個智能指針,這個指針指向的就是我們數(shù)據(jù)量巨大的結(jié)構(gòu)體.
我們把這個key值,跟指向data結(jié)構(gòu)體的指針,尾插到這個map中. 所以我們用一個sharedData類型對象,只要能得到key,是不是就能拿到這個數(shù)據(jù)量巨大的結(jié)構(gòu)體了.當然這個結(jié)構(gòu)體可以是任何東西,在感知內(nèi)部,這個結(jié)構(gòu)體往往是某種傳感器一幀的障礙物信息.

總結(jié)

以上是生活随笔為你收集整理的【自动驾驶】10.百度Apollo平台 事件通信机制的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。