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

【仿RabbitMQ消息队列项目day2】使用muduo库中基于protobuf的应用层协议进行通信

作者：你好赵伟 | 2024-05-22 10:01:57

赞

踩

【仿RabbitMQ消息队列项目day2】使用muduo库中基于protobuf的应用层协议进行通信

一.什么是muduo?

muduo库是⼀个基于非阻塞IO和事件驱动的C++高并发TCP网络编程库。

简单来理解，它就是对原生的TCP套接字的封装，是一个比socket编程接口更好用的编程库。 

二.使用muduo库完成一个英译汉翻译服务

TranslateServer.hpp:

  1. #pragma once
  2. #include <iostream>
  3. #include <functional>
  4. #include <unordered_map>
  5. #include <string>
  6. #include "muduo/net/TcpConnection.h"
  7. #include "muduo/net/TcpServer.h"
  8. #include "muduo/net/EventLoop.h"
  9.  
  10. using std::cout;
  11. using std::endl;
  12. class TranslateServer
  13. {
  14. private:
  15.     muduo::net::EventLoop _baseloop;
  16.     muduo::net::TcpServer _server;
  17. public:
  18.     TranslateServer(int port)
  19.         :_server(&_baseloop, muduo::net::InetAddress("0.0.0.0", port), \
  20.             "TranslateServer", muduo::net::TcpServer::kReusePort)
  21.     {
  22.         //bind是一个函数适配器
  23.         _server.setConnectionCallback(std::bind(&TranslateServer::_onConnection, this, std::placeholders::_1));
  24.         _server.setMessageCallback(std::bind(&TranslateServer::_onMessage, this, std::placeholders::_1, \
  25.             std::placeholders::_2, std::placeholders::_3));
  26.         
  27.     }
  28.     void start()
  29.     {
  30.         _server.start(); //开始事件监听
  31.         _baseloop.loop(); //开始事件监控，这是一个死循环阻塞接口
  32.     }
  33.  
  34. // typedef std::function<void (const TcpConnectionPtr&)> ConnectionCallback;
  35. // typedef std::function<void (const TcpConnectionPtr&,
  36. //                             Buffer*,
  37. //                             Timestamp)> MessageCallback;
  38.  
  39.     //连接建立成功或者关闭时侯的回调函数
  40.     void _onConnection(const muduo::net::TcpConnectionPtr& conn)
  41.     {
  42.         if (conn->connected())
  43.         {
  44.             cout << "新连接建立成功\n";
  45.         }
  46.         else
  47.         {
  48.             cout << "连接关闭\n";
  49.         }
  50.     }
  51.     //通信连接收到请求时的回调函数
  52.     void _onMessage(const muduo::net::TcpConnectionPtr& conn, muduo::net::Buffer* buffer, muduo::Timestamp time)
  53.     {
  54.         std::string str = buffer->retrieveAllAsString();
  55.         std::string resp = translate(str);
  56.         conn->send(resp);
  57.     }
  58.  
  59.     std::string translate(const std::string& str)
  60.     {
  61.         static std::unordered_map<std::string, std::string> _dict = {
  62.             {"hello", "你好"},
  63.             {"white", "白色"}
  64.         };
  65.  
  66.         if (_dict.count(str))
  67.         {
  68.             return _dict[str];
  69.         }
  70.         return "没找到";
  71.     }
  72.     
  73. };

TranslateClient.hpp:

  1. #pragma once
  2. #include <functional>
  3. #include <iostream>
  4. #include "muduo/net/TcpClient.h"
  5. #include "muduo/net/TcpConnection.h"
  6. #include "muduo/net/EventLoopThread.h"
  7.  
  8. class TranslateClient
  9. {
  10. private:
  11.     muduo::net::EventLoopThread _loopThread; //EventLoop是阻塞式死循环，必须另起一个线程，否则用户无法在主线程输入。
  12.     //_loopThread一建立就立马启动
  13.  
  14.     muduo::net::TcpClient _client;
  15.     muduo::net::TcpConnectionPtr _conn;
  16.     //TcpClient的connect是非阻塞接口，调用立马返回，这有可能导致用户send时尚未建立连接，而解引用空指针
  17.     muduo::CountDownLatch _latch; //保证建立连接和send之间的同步关系
  18. public:
  19.     TranslateClient(const std::string& serverIp, int serverPort)
  20.         :_client(_loopThread.startLoop(), muduo::net::InetAddress(serverIp, serverPort), "TranslateClient")
  21.         ,_latch(1)
  22.     {
  23.         _client.setConnectionCallback(std::bind(&TranslateClient::_onConnection, this, std::placeholders::_1));
  24.         _client.setMessageCallback(std::bind(&TranslateClient::_onMessage, this, std::placeholders::_1, \
  25.             std::placeholders::_2,std::placeholders::_3));
  26.     }
  27.  
  28.     void connect()
  29.     {
  30.         _client.connect();
  31.         _latch.wait();
  32.     }
  33.  
  34.     bool send(std::string& msg)
  35.     {
  36.         if (_conn->connected())
  37.         {
  38.             _conn->send(msg);
  39.             return true;
  40.         }
  41.         else
  42.         {
  43.             return false;
  44.         }
  45.     }
  46.  
  47. private:
  48.     /**************
  49.      * 连接建立或者断开时的回调函数
  50.      * **************/
  51.     void _onConnection(const muduo::net::TcpConnectionPtr &conn)
  52.     {
  53.         if (conn->connected())
  54.         {
  55.             _latch.countDown();
  56.             _conn = conn;
  57.         }
  58.         else
  59.         {
  60.             _conn.reset();
  61.         }
  62.     }
  63.  
  64.     void _onMessage(const muduo::net::TcpConnectionPtr& conn, muduo::net::Buffer* buffer, muduo::Timestamp time)
  65.     {
  66.         std::cout << "翻译结果：" << buffer->retrieveAllAsString() << std::endl;
  67.     }
  68. };

muduo的精髓在于大量的回调函数，建立或断开连接，收到消息时，都会调用我们传入的回调函数，回调函数就是我们处理业务的地方。

三.muduo中基于protobuf的自定义协议

像上述的英译汉服务，双方肯定是能正常通信，但这绝不是一个成熟的方案。TCP通信时面向字节流的，存在数据粘包问题，要想解决必须使用用户层协议。

用户层协议主要就是解决数据粘包问题，另外序列化和反序列化也是其中的重要环节。muduo库是由陈硕大佬编写的，在安装好的muduo库中，他提供了一些编程样例，其中有一个就是基于protobuf，定制了一个用户层协议，用于网络通信。所以严格来说，该自定义协议并不是muduo库中的一部分。

  1. class ProtobufCodec : muduo::noncopyable
  2. {
  3.  public:
  4.  
  5.   enum ErrorCode
  6.   {
  7.     kNoError = 0,
  8.     kInvalidLength,
  9.     kCheckSumError,
  10.     kInvalidNameLen,
  11.     kUnknownMessageType,
  12.     kParseError,
  13.   };
  14.  
  15.   typedef std::function<void (const muduo::net::TcpConnectionPtr&,
  16.                                 const MessagePtr&,
  17.                                 muduo::Timestamp)> ProtobufMessageCallback;
  18.  
  19.   typedef std::function<void (const muduo::net::TcpConnectionPtr&,
  20.                                 muduo::net::Buffer*,
  21.                                 muduo::Timestamp,
  22.                                 ErrorCode)> ErrorCallback;
  23.  
  24.   explicit ProtobufCodec(const ProtobufMessageCallback& messageCb)
  25.     : messageCallback_(messageCb),
  26.       errorCallback_(defaultErrorCallback)
  27.   {
  28.   }
  29.  
  30.   ProtobufCodec(const ProtobufMessageCallback& messageCb, const ErrorCallback& errorCb)
  31.     : messageCallback_(messageCb),
  32.       errorCallback_(errorCb)
  33.   {
  34.   }
  35.  
  36.   void onMessage(const muduo::net::TcpConnectionPtr& conn,
  37.                  muduo::net::Buffer* buf,
  38.                  muduo::Timestamp receiveTime);
  39.  
  40.   void send(const muduo::net::TcpConnectionPtr& conn,
  41.             const google::protobuf::Message& message)
  42.   {
  43.     // FIXME: serialize to TcpConnection::outputBuffer()
  44.     muduo::net::Buffer buf;
  45.     fillEmptyBuffer(&buf, message);
  46.     conn->send(&buf);
  47.   }
  48.  
  49.   static const muduo::string& errorCodeToString(ErrorCode errorCode);
  50.   static void fillEmptyBuffer(muduo::net::Buffer* buf, const google::protobuf::Message& message);
  51.   static google::protobuf::Message* createMessage(const std::string& type_name);
  52.   static MessagePtr parse(const char* buf, int len, ErrorCode* errorCode);
  53.  
  54.  private:
  55.   static void defaultErrorCallback(const muduo::net::TcpConnectionPtr&,
  56.                                    muduo::net::Buffer*,
  57.                                    muduo::Timestamp,
  58.                                    ErrorCode);
  59.  
  60.   ProtobufMessageCallback messageCallback_;
  61.   ErrorCallback errorCallback_;
  62.  
  63.   const static int kHeaderLen = sizeof(int32_t);
  64.   const static int kMinMessageLen = 2*kHeaderLen + 2; // nameLen + typeName + checkSum
  65.   const static int kMaxMessageLen = 64*1024*1024; // same as codec_stream.h kDefaultTotalBytesLimit
  66. };

ProtobufCodec类就是基于protobuf定义的结构化数据的应用层协议，协议格式如下：

onMessage的实现如下： 

  1. void ProtobufCodec::onMessage(const TcpConnectionPtr& conn,
  2.                               Buffer* buf,
  3.                               Timestamp receiveTime)
  4. {
  5.   while (buf->readableBytes() >= kMinMessageLen + kHeaderLen)
  6.   {
  7.     const int32_t len = buf->peekInt32();
  8.     if (len > kMaxMessageLen || len < kMinMessageLen)
  9.     {
  10.       errorCallback_(conn, buf, receiveTime, kInvalidLength);
  11.       break;
  12.     }
  13.     else if (buf->readableBytes() >= implicit_cast<size_t>(len + kHeaderLen))
  14.     {
  15.       ErrorCode errorCode = kNoError;
  16.       MessagePtr message = parse(buf->peek()+kHeaderLen, len, &errorCode);
  17.       if (errorCode == kNoError && message)
  18.       {
  19.         messageCallback_(conn, message, receiveTime);
  20.         buf->retrieve(kHeaderLen+len);
  21.       }
  22.       else
  23.       {
  24.         errorCallback_(conn, buf, receiveTime, errorCode);
  25.         break;
  26.       }
  27.     }
  28.     else
  29.     {
  30.       break;
  31.     }
  32.   }
  33. }

onMessage函数解决了TCP粘包的问题，从缓冲区中解析出一个完整的protobuf结构化数据(一个message)后，再调用messageCallback_处理。messageCallback_是构造ProtobufCodec时传入的回调函数。

如果我们的业务场景很单一，例如上面的英译汉服务器，直接把我们写的业务逻辑作为回调传给messageCallback_就OK了。但如果我们有多种业务，例如翻译和计算业务，则还可以在此基础上引入任务分发器ProtobufDispatcher，回调它的ProtobufDispatcher函数。

  1. class ProtobufDispatcher
  2. {
  3.  public:
  4.   typedef std::function<void (const muduo::net::TcpConnectionPtr&,
  5.                                 const MessagePtr& message,
  6.                                 muduo::Timestamp)> ProtobufMessageCallback;
  7.  
  8.   explicit ProtobufDispatcher(const ProtobufMessageCallback& defaultCb)
  9.     : defaultCallback_(defaultCb)
  10.   {
  11.   }
  12.  
  13.   void onProtobufMessage(const muduo::net::TcpConnectionPtr& conn,
  14.                          const MessagePtr& message,
  15.                          muduo::Timestamp receiveTime) const
  16.   {
  17.     CallbackMap::const_iterator it = callbacks_.find(message->GetDescriptor());
  18.     if (it != callbacks_.end())
  19.     {
  20.       it->second->onMessage(conn, message, receiveTime);
  21.     }
  22.     else
  23.     {
  24.       defaultCallback_(conn, message, receiveTime);
  25.     }
  26.   }
  27.  
  28.   template<typename T>
  29.   void registerMessageCallback(const typename CallbackT<T>::ProtobufMessageTCallback& callback)
  30.   {
  31.     std::shared_ptr<CallbackT<T> > pd(new CallbackT<T>(callback));
  32.     callbacks_[T::descriptor()] = pd;
  33.   }
  34.  
  35.  private:
  36.   typedef std::map<const google::protobuf::Descriptor*, std::shared_ptr<Callback> > CallbackMap;
  37.  
  38.   CallbackMap callbacks_;
  39.   ProtobufMessageCallback defaultCallback_;
  40. };

onProtobufMessage会根据你传入的结构化数据类型(message)，调用不同的回调函数，这些回调函数就是我们注册的业务处理方法。

四.编写一个翻译+加法服务

  1. 编写.并翻译proto文件，构建翻译的请求和响应，加法的请求和响应的类
  2. 编写服务端
  3. 编写客户端

Server.cc:

  1. #include <memory>
  2. #include "muduo/protobuf/codec.h"
  3. #include "muduo/protobuf/dispatcher.h"
  4. #include "muduo/base/Logging.h"
  5.  
  6. #include "muduo/net/TcpServer.h"
  7. #include "muduo/net/TcpConnection.h"
  8. #include "muduo/net/EventLoop.h"
  9.  
  10. #include "business.pb.h"
  11.  
  12. using std::placeholders::_1;
  13. using std::placeholders::_2;
  14. using std::placeholders::_3;
  15. class Server
  16. {
  17. public:
  18.     typedef std::shared_ptr<google::protobuf::Message> MessagePtr;
  19.     typedef std::shared_ptr<business::TranslateRequest> TranslateRequestPtr;
  20.     typedef std::shared_ptr<business::AddRequest> AddRequestPtr;
  21.  
  22. private:
  23.     muduo::net::EventLoop _baseLoop;
  24.     muduo::net::TcpServer _server;
  25.     ProtobufDispatcher _dispatcher; // 请求分发器
  26.     ProtobufCodec _codec;           // protobuf处理器--解析出结构化数据，发送结构化数据(序列化和发序列化内部会做)
  27. public:
  28.     Server(int port)
  29.         : _server(&_baseLoop, muduo::net::InetAddress("0.0.0.0", port), "Server",
  30.                   muduo::net::TcpServer::kReusePort)
  31.         ,_dispatcher(std::bind(&Server::_onUnknownMessage, this,\
  32.             std::placeholders::_1, std::placeholders::_2, std::placeholders::_3))
  33.         ,_codec(std::bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher,
  34.             std::placeholders::_1, std::placeholders::_2, std::placeholders::_3))
  35.     {
  36.         // 注册业务处理函数
  37.        _dispatcher.registerMessageCallback<business::AddRequest>(bind(&Server::_onAdd, this, _1, _2, _3));
  38.        _dispatcher.registerMessageCallback<business::TranslateRequest>(bind(&Server::_onTranslate, this, _1, _2, _3));
  39.  
  40.         //注册_server的回调函数
  41.         _server.setMessageCallback(std::bind(&ProtobufCodec::onMessage, &_codec,
  42.             std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
  43.         _server.setConnectionCallback(std::bind(&Server::_onConnection, this, std::placeholders::_1));
  44.     }
  45.  
  46.     void start()
  47.     {
  48.         _server.start();
  49.         _baseLoop.loop();
  50.     }
  51.  
  52. private:
  53.     void _onUnknownMessage(const muduo::net::TcpConnectionPtr &conn, const MessagePtr &message, muduo::Timestamp)
  54.     {
  55.         LOG_INFO << "onUnknownMessage: " << message->GetTypeName();
  56.         conn->shutdown();
  57.     }
  58.  
  59.     void _onAdd(const muduo::net::TcpConnectionPtr& conn, const AddRequestPtr &messagePtr, muduo::Timestamp time)
  60.     {
  61.         int x = messagePtr->num1();
  62.         int y = messagePtr->num2();
  63.         
  64.         business::AddResponse resp;
  65.         resp.set_result(x + y);
  66.         _codec.send(conn, resp); //让protobuf处理器帮我们序列化并用conn发送
  67.     }
  68.  
  69.     void _onTranslate(const muduo::net::TcpConnectionPtr& conn, const TranslateRequestPtr &messagePtr, muduo::Timestamp time)
  70.     {
  71.         const std::string& ret = translate(messagePtr->msg());
  72.         business::TranslateResponse resp;
  73.         resp.set_msg(ret);
  74.         _codec.send(conn, resp);
  75.     }
  76.  
  77.     void _onConnection(const muduo::net::TcpConnectionPtr &conn)
  78.     {
  79.         if (conn->connected())
  80.         {
  81.             LOG_INFO << "新连接建立成功！";
  82.         }
  83.         else
  84.         {
  85.             LOG_INFO << "连接即将关闭！";
  86.         }
  87.     }
  88.  
  89.     std::string translate(const std::string& str)
  90.     {
  91.         static std::unordered_map<std::string, std::string> dict_map = {
  92.                 {"hello", "你好"},
  93.                 {"Hello", "你好"},
  94.                 {"你好", "Hello"},
  95.                 {"吃了吗", "油泼面"}
  96.         };
  97.         auto it = dict_map.find(str);
  98.         if (it == dict_map.end()) {
  99.             return "没听懂！！";
  100.         }
  101.         return it->second;
  102.     }
  103. };
  104.  
  105. int main()
  106. {
  107.     Server server(8085);
  108.     server.start();
  109.     return 0;
  110. }

Client.cc: 

  1.  
  2. #include "muduo/protobuf/codec.h"
  3. #include "muduo/protobuf/dispatcher.h"
  4. #include "muduo/base/Logging.h"
  5. #include "muduo/base/Mutex.h"
  6. #include "muduo/net/EventLoop.h"
  7. #include "muduo/net/TcpClient.h"
  8. #include "muduo/net/EventLoopThread.h"
  9. #include "muduo/base/CountDownLatch.h"
  10.  
  11. #include "business.pb.h"
  12. #include <iostream>
  13. #include <functional>
  14.  
  15. class Client {
  16.     public:
  17.         typedef std::shared_ptr<google::protobuf::Message> MessagePtr; //这是Protobuf库的头文件
  18.         typedef std::shared_ptr<business::AddResponse> AddResponsePtr;
  19.         typedef std::shared_ptr<business::TranslateResponse> TranslateResponsePtr;
  20.         Client(const std::string &sip, int sport):
  21.             _latch(1), _client(_loopthread.startLoop(), muduo::net::InetAddress(sip, sport), "Client"),
  22.             _dispatcher(std::bind(&Client::onUnknownMessage, this, std::placeholders::_1, 
  23.                 std::placeholders::_2, std::placeholders::_3)),
  24.             _codec(std::bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher, 
  25.                 std::placeholders::_1, std::placeholders::_2, std::placeholders::_3)){
  26.  
  27.             _dispatcher.registerMessageCallback<business::TranslateResponse>(std::bind(&Client::onTranslate, this, 
  28.                 std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
  29.                 
  30.             _dispatcher.registerMessageCallback<business::AddResponse>(std::bind(&Client::onAdd, this, 
  31.                 std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
  32.  
  33.             _client.setMessageCallback(std::bind(&ProtobufCodec::onMessage, &_codec,
  34.                 std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
  35.             _client.setConnectionCallback(std::bind(&Client::onConnection, this, std::placeholders::_1));      
  36.         }
  37.         void connect() {
  38.             _client.connect();
  39.             _latch.wait();//阻塞等待，直到连接建立成功
  40.         }
  41.         void Translate(const std::string &msg){
  42.             business::TranslateRequest req;
  43.             req.set_msg(msg);
  44.             send(&req);
  45.         }
  46.         void Add(int num1, int num2) {
  47.             business::AddRequest req;
  48.             req.set_num1(num1);
  49.             req.set_num2(num2);
  50.             send(&req);
  51.         }
  52.     private:
  53.         bool send(const google::protobuf::Message *message) {
  54.             if (_conn->connected()) {//连接状态正常，再发送，否则就返回false
  55.                 _codec.send(_conn, *message);
  56.                 return true;
  57.             }
  58.             return false;
  59.         }  
  60.         void onTranslate(const muduo::net::TcpConnectionPtr& conn, const TranslateResponsePtr& message, muduo::Timestamp) {
  61.             std::cout << "翻译结果：" << message->msg() << std::endl;
  62.         }
  63.         void onAdd(const muduo::net::TcpConnectionPtr& conn, const AddResponsePtr& message, muduo::Timestamp) {
  64.             std::cout << "加法结果：" << message->result() << std::endl;
  65.         }
  66.         void onUnknownMessage(const muduo::net::TcpConnectionPtr& conn, const MessagePtr& message, muduo::Timestamp) {
  67.             LOG_INFO << "onUnknownMessage: " << message->GetTypeName();
  68.             conn->shutdown();
  69.         }
  70.         void onConnection(const muduo::net::TcpConnectionPtr&conn){
  71.             if (conn->connected()) {
  72.                 _latch.countDown();//唤醒主线程中的阻塞
  73.                 _conn = conn;
  74.             }else {
  75.                 //连接关闭时的操作
  76.                 _conn.reset();
  77.             }
  78.         }
  79.     private:
  80.         muduo::CountDownLatch _latch;//实现同步的
  81.         muduo::net::EventLoopThread _loopthread;//异步循环处理线程
  82.         muduo::net::TcpConnectionPtr _conn;//客户端对应的连接
  83.         muduo::net::TcpClient _client;//客户端
  84.         ProtobufDispatcher _dispatcher;//请求分发器
  85.         ProtobufCodec _codec; //Protobuf处理器
  86. };
  87.  
  88. int main() 
  89. {
  90.     Client client("127.0.0.1", 8085);
  91.     client.connect();
  92.  
  93.     client.Translate("hello");
  94.     client.Add(11, 22);
  95.  
  96.     sleep(1);
  97.     return 0;
  98. }

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

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

  

闽ICP备14008679号