ROS機器人操作系統(tǒng)的實現(xiàn)原理（上）

本文介紹ROS機器人操作系統(tǒng)（Robot Operating System）的實現(xiàn)原理，從最底層分析ROS代碼是如何實現(xiàn)的。

1、序列化

把通信的內(nèi)容（也就是消息message）序列化是通信的基礎(chǔ)，所以我們先研究序列化。

盡管筆者從事機器人學(xué)習(xí)和研發(fā)很長時間了，但是在研究ROS的過程中，“序列化”這個詞還是這輩子第一次聽到。

所以可想而知很多人在看到“把一個消息序列化”這樣的描述時是如何一臉懵逼。

但其實序列化是一個比較常見的概念，你雖然不知道它但一定接觸過它。

下面我們先介紹“序列化”的一些常識，然后解釋ROS里的序列化是怎么做的？

1.1什么是序列化？

“序列化”（Serialization ）的意思是將一個對象轉(zhuǎn)化為字節(jié)流。

這里說的對象可以理解為“面向?qū)ο蟆崩锏哪莻€對象，具體的就是存儲在內(nèi)存中的對象數(shù)據(jù)。

與之相反的過程是“反序列化”（Deserialization ）。

雖然掛著機器人的羊頭，但是后面的介紹全部是計算機知識，跟機器人一丁點關(guān)系都沒有，序列化就是一個純粹的計算機概念。

序列化的英文Serialize就有把一個東西變成一串連續(xù)的東西之意。

形象的描述，數(shù)據(jù)對象是一團面，序列化就是將面團拉成一根面條，反序列化就將面條捏回面團。

另一個形象的類比是我們在對話或者打電話時，一個人的思想轉(zhuǎn)換成一維的語音，然后在另一個人的頭腦里重新變成結(jié)構(gòu)化的思想，這也是一種序列化。

面對序列化，很多人心中可能會有很多疑問。

首先，為什么要序列化？或者更具體的說，既然對象的信息本來就是以字節(jié)的形式儲存在內(nèi)存中，那為什么要多此一舉把一些字節(jié)數(shù)據(jù)轉(zhuǎn)換成另一種形式的、一維的、連續(xù)的字節(jié)數(shù)據(jù)呢？

如果我們的程序在內(nèi)存中存儲了一個數(shù)字，比如25。那要怎么傳遞25這個數(shù)字給別的程序節(jié)點或者把這個數(shù)字永久存儲起來呢？

很簡單，直接傳遞25這個數(shù)字（的字節(jié)表示，即0X19，當(dāng)然最終會變成二進制表示11001以高低電平傳輸存儲）或者直接把這個數(shù)字（的字節(jié)表示）寫進硬盤里即可。

所以，對于本來就是連續(xù)的、一維的、一連串的數(shù)據(jù)（例如字符串），序列化并不需要做太多東西，其本質(zhì)是就是由內(nèi)存向其它地方拷貝數(shù)據(jù)而已。

所以，如果你在一個序列化庫里看到memcpy函數(shù)不用覺得奇怪，因為你知道序列化最底層不過就是在操作內(nèi)存數(shù)據(jù)而已（還有些庫使用了流的ostream.rdbuf()->sputn函數(shù)）。

可是實際程序操作的對象很少是這么簡單的形式，大多數(shù)時候我們面對的是包含不同數(shù)據(jù)類型（int、double、string）的復(fù)雜數(shù)據(jù)結(jié)構(gòu)（比如vector、list），它們很可能在內(nèi)存中是不連續(xù)存儲的而是分散在各處。比如ROS的很多消息都包含向量。

數(shù)據(jù)中還有各種指針和引用。而且，如果數(shù)據(jù)要在運行于不同架構(gòu)的計算機之上的、由不同編程語言所編寫的節(jié)點程序之間傳遞，那問題就更復(fù)雜了，它們的字節(jié)順序endianness規(guī)定有可能不一樣，基本數(shù)據(jù)類型（比如int）的長度也不一樣（有的int是4個字節(jié)、有的是8個字節(jié)）。

這些都不是通過簡單地、原封不動地復(fù)制粘貼原始數(shù)據(jù)就能解決的。這時候就需要序列化和反序列化了。

所以在程序之間需要通信時（ROS恰好就是這種情況），或者希望保存程序的中間運算結(jié)果時，序列化就登場了。

另外，在某種程度上，序列化還起到統(tǒng)一標(biāo)準(zhǔn)的作用。

我們把被序列化的東西叫object（對象），它可以是任意的數(shù)據(jù)結(jié)構(gòu)或者對象：結(jié)構(gòu)體、數(shù)組、類的實例等等。

把序列化后得到的東西叫archive，它既可以是人類可讀的文本形式，也可以是二進制形式。

前者比如JSON和XML，這兩個是網(wǎng)絡(luò)應(yīng)用里最常用的序列化格式，通過記事本就能打開閱讀；

后者就是原始的二進制文件，比如后綴名是bin的文件，人類是沒辦法直接閱讀一堆的0101或者0XC9D23E72的。

序列化算是一個比較常用的功能，所以大多數(shù)編程語言（比如C++、Python、Java等）都會附帶用于序列化的庫，不需要你再去造輪子。

以C++為例，雖然標(biāo)準(zhǔn)STL庫沒有提供序列化功能，但是第三方庫Boost提供了[ 2 ]谷歌的protobuf也是一個序列化庫，還有Fast-CDR，以及不太知名的Cereal，Java自帶序列化函數(shù)，python可以使用第三方的pickle模塊實現(xiàn)。

總之，序列化沒有什么神秘的，用戶可以看看這些開源的序列化庫代碼，或者自己寫個小程序試試簡單數(shù)據(jù)的序列化，例如這個例子，或者這個，有助于更好地理解ROS中的實現(xiàn)。

1.2ROS中的序列化實現(xiàn)

理解了序列化，再回到ROS。我們發(fā)現(xiàn)，ROS沒有采用第三方的序列化工具，而是選擇自己實現(xiàn)，代碼在roscpp_core項目下的roscpp_serialization中，見下圖。這個功能涉及的代碼量不是很多。

為什么ROS不使用現(xiàn)成的序列化工具或者庫呢？可能ROS誕生的時候（2007年），有些序列化庫可能還不存在（protobuf誕生于2008年），更有可能是ROS的創(chuàng)造者認(rèn)為當(dāng)時沒有合適的工具。

1.2.1serialization.h

核心的函數(shù)都在serialization.h里，簡而言之，里面使用了C語言標(biāo)準(zhǔn)庫的memcpy函數(shù)把消息拷貝到流中。

下面來看一下具體的實現(xiàn)。

序列化功能的特點是要處理很多種數(shù)據(jù)類型，針對每種具體的類型都要實現(xiàn)相應(yīng)的序列化函數(shù)。

為了盡量減少代碼量，ROS使用了模板的概念，所以代碼里有一堆的template。

從后往前梳理，先看Stream這個結(jié)構(gòu)體吧。在C++里結(jié)構(gòu)體和類基本沒什么區(qū)別，結(jié)構(gòu)體里也可以定義函數(shù)。

Stream翻譯為流，流是一個計算機中的抽象概念，前面我們提到過字節(jié)流，它是什么意思呢？

在需要傳輸數(shù)據(jù)的時候，我們可以把數(shù)據(jù)想象成傳送帶上連續(xù)排列的一個個被傳送的物體，它們就是一個流。

更形象的，可以想象磁帶或者圖靈機里連續(xù)的紙帶。在文件讀寫、使用串口、網(wǎng)絡(luò)Socket通信等領(lǐng)域，流經(jīng)常被使用。例如我們常用的輸入輸出流：

cout<<"helllo"; 由于使用很多，流的概念也在演變。想了解更多可以看這里。

struct Stream
{
  // Returns a pointer to the current position of the stream
  inline uint8_t* getData() { return data_; }
  // Advances the stream, checking bounds, and returns a pointer to the position before it was advanced.
  // \\throws StreamOverrunException if len would take this stream past the end of its buffer
  ROS_FORCE_INLINE uint8_t* advance(uint32_t len)
{
    uint8_t* old_data = data_;
    data_ += len;
    if (data_ > end_)
    {
      // Throwing directly here causes a significant speed hit due to the extra code generated for the throw statement
      throwStreamOverrun();
    }
    return old_data;
  }
  // Returns the amount of space left in the stream
  inline uint32_t getLength() { return static_cast<uint32_t>(end_ - data_); }
  
protected:
  Stream(uint8_t* _data, uint32_t _count) : data_(_data), end_(_data + _count) {}


private:
  uint8_t* data_;
  uint8_t* end_;
};