ROS中的序列化实现

理解了序列化，再回到ROS。我们发现，ROS没有采用第三方的序列化工具，而是选择自己实现，代码在roscpp_core项目下的roscpp_serialization中，见下图。这个功能涉及的代码量不是很多。

为什么ROS不使用现成的序列化工具或者库呢？可能ROS诞生的时候（2007年），有些序列化库可能还不存在（protobuf诞生于2008年），更有可能是ROS的创造者认为当时没有合适的工具。

1.2.1 serialization.h

核心的函数都在serialization.h里，简而言之，里面使用了C语言标准库的memcpy函数把消息拷贝到流中。

下面来看一下具体的实现。

序列化功能的特点是要处理很多种数据类型，针对每种具体的类型都要实现相应的序列化函数。

为了尽量减少代码量，ROS使用了模板的概念，所以代码里有一堆的template。

从后往前梳理，先看Stream这个结构体吧。在C++里结构体和类基本没什么区别，结构体里也可以定义函数。

Stream翻译为流，流是一个计算机中的抽象概念，前面我们提到过字节流，它是什么意思呢？

在需要传输数据的时候，我们可以把数据想象成传送带上连续排列的一个个被传送的物体，它们就是一个流。

更形象的，可以想象磁带或者图灵机里连续的纸带。在文件读写、使用串口、网络Socket通信等领域，流经常被使用。例如我们常用的输入输出流：

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_;
};

注释表明Stream是个基类，输入输出流IStream和OStream都继承自它。

Stream的成员变量data_是个指针，指向序列化的字节流开始的位置，它的类型是uint8_t。

在Ubuntu系统中，uint8_t的定义是typedef unsigned char uint8_t;

所以uint8_t就是一个字节，可以用size_of()函数检验。data_指向的空间就是保存字节流的。

输出流类OStream用来序列化一个对象，它引用了serialize函数，如下。

struct OStream : public Stream
{
  static const StreamType stream_type = stream_types::Output;
  OStream(uint8_t* data, uint32_t count) : Stream(data, count) {}
  /* Serialize an item to this output stream*/
  template< typename T >
  ROS_FORCE_INLINE void next(const T& t)
{
    serialize(*this, t);
  }
  template< typename T >
  ROS_FORCE_INLINE OStream& operator< < (const T& t)
  {
    serialize(*this, t);
    return *this;
  }
};

输入流类IStream用来反序列化一个字节流，它引用了deserialize函数，如下。

struct ROSCPP_SERIALIZATION_DECL IStream : public Stream
{
  static const StreamType stream_type = stream_types::Input;
  IStream(uint8_t* data, uint32_t count) : Stream(data, count) {}
  /* Deserialize an item from this input stream */
  template< typename T >
  ROS_FORCE_INLINE void next(T& t)
{
    deserialize(*this, t);
  }
  template< typename T >
  ROS_FORCE_INLINE IStream& operator >>(T& t)
  {
    deserialize(*this, t);
    return *this;
  }
};

自然，serialize函数和deserialize函数就是改变数据形式的地方，它们的定义在比较靠前的地方。它们都接收两个模板，都是内联函数，然后里面没什么东西，只是又调用了Serializer类的成员函数write和read。所以，serialize和deserialize函数就是个二道贩子。

// Serialize an object.  Stream here should normally be a ros::serialization::OStream
template< typename T, typename Stream >
inline void serialize(Stream& stream, const T& t)
{
  Serializer< T >::write(stream, t);
}
// Deserialize an object.  Stream here should normally be a ros::serialization::IStream
template< typename T, typename Stream >
inline void deserialize(Stream& stream, T& t)
{
  Serializer< T >::read(stream, t);
}

所以，我们来分析Serializer类，如下。我们发现，write和read函数又调用了类型里的serialize函数和deserialize函数。

头别晕，这里的serialize和deserialize函数跟上面的同名函数不是一回事。

注释中说：“Specializing the Serializer class is the only thing you need to do to get the ROS serialization system to work with a type”（要想让ROS的序列化功能适用于其它的某个类型，你唯一需要做的就是特化这个Serializer类）。

这就涉及到的另一个知识点——模板特化（template specialization）。

template< typename T > struct Serializer
{
  // Write an object to the stream.  Normally the stream passed in here will be a ros::serialization::OStream
  template< typename Stream >
  inline static void write(Stream& stream, typename boost::call_traits< T >::param_type t)
{
    t.serialize(stream.getData(), 0);
  }
   // Read an object from the stream.  Normally the stream passed in here will be a ros::serialization::IStream
  template< typename Stream >
  inline static void read(Stream& stream, typename boost::call_traits< T >::reference t)
{
    t.deserialize(stream.getData());
  }
  // Determine the serialized length of an object.
  inline static uint32_t serializedLength(typename boost::call_traits< T >::param_type t)
{
    return t.serializationLength();
  }
};

接着又定义了一个带参数的宏函数ROS_CREATE_SIMPLE_SERIALIZER(Type)，然后把这个宏作用到了ROS中的10种基本数据类型，分别是：uint8_t, int8_t, uint16_t, int16_t, uint32_t, int32_t, uint64_t, int64_t, float, double。

说明这10种数据类型的处理方式都是类似的。看到这里大家应该明白了，write和read函数都使用了memcpy函数进行数据的移动。

注意宏定义中的template<>语句，这正是模板特化的标志，关键词template后面跟一对尖括号。

关于模板特化可以看这里。

#define ROS_CREATE_SIMPLE_SERIALIZER(Type) 
  template<  > struct Serializer<  Type > 
  { 
    template< typename Stream > inline static void write(Stream& stream, const Type v) 
{ 
      memcpy(stream.advance(sizeof(v)), &v, sizeof(v) ); 
    } 
    template< typename Stream > inline static void read(Stream& stream, Type& v) 
{ 
      memcpy(&v, stream.advance(sizeof(v)), sizeof(v) ); 
    } 
    inline static uint32_t serializedLength(const Type&) 
{ 
      return sizeof(Type); 
    } 
};
ROS_CREATE_SIMPLE_SERIALIZER(uint8_t)
ROS_CREATE_SIMPLE_SERIALIZER(int8_t)
ROS_CREATE_SIMPLE_SERIALIZER(uint16_t)
ROS_CREATE_SIMPLE_SERIALIZER(int16_t)
ROS_CREATE_SIMPLE_SERIALIZER(uint32_t)
ROS_CREATE_SIMPLE_SERIALIZER(int32_t)
ROS_CREATE_SIMPLE_SERIALIZER(uint64_t)
ROS_CREATE_SIMPLE_SERIALIZER(int64_t)
ROS_CREATE_SIMPLE_SERIALIZER(float)
ROS_CREATE_SIMPLE_SERIALIZER(double)

对于其它类型的数据，例如bool、std::string、std::vector、ros::Time、ros::Duration、boost::array等等，它们各自的处理方式有细微的不同，所以不再用上面的宏函数，而是用模板特化的方式每种单独定义，这也是为什么serialization.h这个文件这么冗长。

对于int、double这种单个元素的数据，直接用上面特化的Serializer类中的memcpy函数实现序列化。

对于vector、array这种多个元素的数据类型怎么办呢？方法是分成几种情况，对于固定长度简单类型的（fixed-size simple types），还是用各自特化的Serializer类中的memcpy函数实现，没啥太大区别。

对于固定但是类型不简单的（fixed-size non-simple types）或者既不固定也不简单的（non-fixed-size, non-simple types）或者固定但是不简单的（fixed-size, non-simple types），用for循环遍历，一个元素一个元素的单独处理。

那怎么判断一个数据是不是固定是不是简单呢？这是在roscpp_traits文件夹中的message_traits.h完成的。

其中采用了萃取Type Traits，这是相对高级一点的编程技巧了，笔者也不太懂。

对序列化的介绍暂时就到这里了，有一些细节还没讲，等笔者看懂了再补。