devbox
盐析白兔
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Muduo + ProtoBuffer的使用_muduo protobuf使用

作者：盐析白兔 | 2024-05-30 15:17:54

赞

踩

muduo protobuf使用

Google Protocol Buffer（简称protobuf）是Google内部混合语言数据标准，protobuf是一种紧凑的可扩展的二进制消息格式，适合做网络数据传输、数据存储的消息格式（摘自网络）。这里不介绍protobuf的基本用法，详情可以参考： 1. ProtoBuf语法指南  http://colobu.com/2015/01/07/Protobuf-language-guide/  2. 玩转ProtoBuf http://www.searchtb.com/2012/09/protocol-buffers.html   3. Protobuf 反射和相关API  http://name5566.com/2473.html
  这里主要介绍Muduo中使用ProtoBuffer实现消息类型自解析自动调用相应的处理函数，对应的代码为： https://github.com/chenshuo/muduo/tree/master/examples/protobuf/codec
  首先我们要描述一下需要解决的问题：1. 处理端有各种消息的完整类型，但是不知道当前接收的消息类型（消息识别）  2. 处理端可以正确处理不同种类的消息（消息解析 + 消息处理）。2. 处理端不要用大量的if else或者switch来判断消息的类型然后做相应的处理 （代码不要太丑陋太难维护）

一. 消息识别

 为了达到消息识别的目的，首先需要定义消息格式：

 
 len : 消息长度
 namelen : 消息类型名长度
 messageName : 消息类型名
 protoBufData : 消息体数据
 checkSum : 校验和 (这里使用 adler32 算法)
 注意：使用ProtoBuffer进行传输是不需要消息的版本号的，因为ProtoBuffer支持optional字段，可以通过增加或减少optional字段来实现不同版本的消息，而且不会引起解析错误。
  
 

二. 消息解析

当处理端收到消息后，应该怎样处理了，注意处理端收到的字节流很可能不能构成一个完整的消息，也可能构成多个消息
       对数据的解码在ProtobufCodec类中，onMessage函数就是处理原始字节流的，parse函数从原始字节流中提取出消息

  1. void ProtobufCodec::onMessage(const TcpConnectionPtr& conn,
  2.                               Buffer* buf,
  3.                               Timestamp receiveTime)
  4. {
  5.   while (buf->readableBytes() >= kMinMessageLen + kHeaderLen)
  6.   {
  7.     // 1. 获取消息长度 (头四个字节)
  8.     const int32_t len = buf->peekInt32();
  9.     if (len > kMaxMessageLen || len < kMinMessageLen)
  10.     {
  11.       errorCallback_(conn, buf, receiveTime, kInvalidLength);
  12.       break;
  13.     }
  14.     else if (buf->readableBytes() >= implicit_cast<size_t>(len + kHeaderLen))
  15.     {
  16.       ErrorCode errorCode = kNoError;
  17.       // 2. parse生成具体消息
  18.       MessagePtr message = parse(buf->peek()+kHeaderLen, len, &errorCode);
  19.       if (errorCode == kNoError && message)
  20.       {
  21.         // 3. 调用消息处理函数
  22.         messageCallback_(conn, message, receiveTime);
  23.         // 4. 从缓冲区中删除已处理的消息字节数据
  24.         buf->retrieve(kHeaderLen+len);
  25.       }
  26.       else
  27.       {
  28.         errorCallback_(conn, buf, receiveTime, errorCode);
  29.         break;
  30.       }
  31.     }
  32.     else
  33.     {
  34.       break;
  35.     }
  36.   }
  37. }
  1. MessagePtr ProtobufCodec::parse(const char* buf, int len, ErrorCode* error)
  2. {
  3.   MessagePtr message;
  4.  
  5.   // check sum
  6.   // 1. 检查校验和
  7.   int32_t expectedCheckSum = asInt32(buf + len - kHeaderLen);
  8.   int32_t checkSum = static_cast<int32_t>(
  9.       ::adler32(1,
  10.                 reinterpret_cast<const Bytef*>(buf),
  11.                 static_cast<int>(len - kHeaderLen)));
  12.   if (checkSum == expectedCheckSum)
  13.   {
  14.     // get message type name
  15.     // 2. 获取消息类型名长度
  16.     int32_t nameLen = asInt32(buf);
  17.     if (nameLen >= 2 && nameLen <= len - 2*kHeaderLen)
  18.     {
  19.       // 3. 获取消息类型名
  20.       std::string typeName(buf + kHeaderLen, buf + kHeaderLen + nameLen - 1);
  21.       // create message object
  22.       // 4. 根据消息类型名创建一个消息
  23.       message.reset(createMessage(typeName));
  24.       if (message)
  25.       {
  26.         // parse from buffer
  27.         const char* data = buf + kHeaderLen + nameLen;
  28.         int32_t dataLen = len - nameLen - 2*kHeaderLen;
  29.         // 5. 从原始字节流中反序列化出消息数据
  30.         if (message->ParseFromArray(data, dataLen))
  31.         {
  32.           *error = kNoError;
  33.         }
  34.         else
  35.         {
  36.           *error = kParseError;
  37.         }
  38.       }
  39.       else
  40.       {
  41.         *error = kUnknownMessageType;
  42.       }
  43.     }
  44.     else
  45.     {
  46.       *error = kInvalidNameLen;
  47.     }
  48.   }
  49.   else
  50.   {
  51.     *error = kCheckSumError;
  52.   }
  53.  
  54.   return message;
  55. }

这里很重要的一点就是如何根据消息类型名创建一个消息，createMessage完成这个功能

  1. google::protobuf::Message* ProtobufCodec::createMessage(const std::string& typeName)
  2. {
  3.   google::protobuf::Message* message = NULL;
  4.   const google::protobuf::Descriptor* descriptor =
  5.     google::protobuf::DescriptorPool::generated_pool()->FindMessageTypeByName(typeName);
  6.   if (descriptor)
  7.   {
  8.     const google::protobuf::Message* prototype =
  9.       google::protobuf::MessageFactory::generated_factory()->GetPrototype(descriptor);
  10.     if (prototype)
  11.     {
  12.       message = prototype->New();
  13.     }
  14.   }
  15.   return message;
  16. }

 

该函数用到的几个重要的步骤：

  1. 获取 MessageFactory 对象
    MessageFactory 类提供了一个 generated_factory() 的静态函数，此静态函数可以获取一个 MessageFactory 对象，此 MessageFactory 对象能够用来创建被编译入程序的所有的 message 对象。注意，此 Factory 是一个 Singleton，因此重复多次调用 generated_factory 函数不会创建多个 MessageFactory 对象，另外调用者也不能通过调用 delete 删除此对象。
  2. 获取 DescriptorPool 对象
    通过 DescriptorPool 类的 generated_pool() 静态函数能够获取 DescriptorPool 的指针。此 DescriptorPool 中包含了被编译入程序的 message 的 descriptor。generated_pool 类似于 generated_factory 函数，可以被重复调用多次而不会创建多个 DescriptorPool 对象。
  3. 获取 message descriptor
    有了 DescriptorPool 对象就可以获取到 message 的 descriptor 了。常见的一个函数是 const Descriptor * FindMessageTypeByName(const string & name) const，此函数可以通过 message 名字获取到顶层 message 的 descriptor。当然除此之外还有一些 API 可以用来获取 message discriptor，可以参考相关文档，这里就不一一详述了。
  4. 获取 message prototype 并构建 message 对象
    前面已经讲述了获取 MessageFactory 对象的方法，有了 MessageFactory 对象就可以通过函数 MessageFactory::GetPrototype(const Descriptor * type) 获取 message prototype（实质上就是一个 message 对象）。通过调用 message prototype 的 New 函数则可以构造此类型的 message。
    对同一个 Descriptor 多次调用 MessageFactory::GetPrototype 函数将返回同一个对象。通过调用 prototype 的 New 函数构造的 message 对象必须在 MessageFactory 销毁前销毁。

三. 消息处理

  消息的处理是在函数messageCallBack_中进行，这个函数是我们在创建QueryServer时绑定的：

  1. QueryServer(EventLoop* loop,
  2.               const InetAddress& listenAddr)
  3.   : server_(loop, listenAddr, "QueryServer"),
  4.     dispatcher_(boost::bind(&QueryServer::onUnknownMessage, this, _1, _2, _3)),
  5.  
  6.     // 将ProtobufDispatcher::onProtobufMessage 绑定到messageCallBack_上
  7.     codec_(boost::bind(&ProtobufDispatcher::onProtobufMessage, &dispatcher_, _1, _2, _3))
  8.   {
  9.     // 注册的消息处理函数
  10.     dispatcher_.registerMessageCallback<muduo::Query>(
  11.         boost::bind(&QueryServer::onQuery, this, _1, _2, _3));
  12.     dispatcher_.registerMessageCallback<muduo::Answer>(
  13.         boost::bind(&QueryServer::onAnswer, this, _1, _2, _3));
  14.     server_.setConnectionCallback(
  15.         boost::bind(&QueryServer::onConnection, this, _1));
  16.     server_.setMessageCallback(
  17.         boost::bind(&ProtobufCodec::onMessage, &codec_, _1, _2, _3));
  18.   }

 onProtobufMessage根据每个类型的descriptor来从一张表中查找处理函数，因为每种类型都有一个全局的Descriptor对象，它的地址不变，因此使用它来查找注册的函数

  1. void onProtobufMessage(const muduo::net::TcpConnectionPtr& conn,
  2.                          const MessagePtr& message,
  3.                          muduo::Timestamp receiveTime) const
  4.   {
  5.     CallbackMap::const_iterator it = callbacks_.find(message->GetDescriptor());
  6.     if (it != callbacks_.end())
  7.     {
  8.       it->second->onMessage(conn, message, receiveTime);
  9.     }
  10.     else
  11.     {
  12.       defaultCallback_(conn, message, receiveTime);
  13.     }
  14.   }

至此，我们就实现了消息的自解析和自处理。

向原创致敬，转自：http://blog.chinaunix.net/uid-28587158-id-5193485.html

 

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/盐析白兔/article/detail/647500
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号