【微服务即时通讯系统】——brpc远程过程调用、百度开源的RPC框架、brpc的介绍、brpc的安装、brpc使用和功能测试

brpc

  

1. brpc的介绍

1.1 rpc的介绍

  RPC（Remote Procedure Call）远程过程调用，是一种计算机通信协议，它允许程序在不同的计算机之间进行通信和交互，就像本地调用一样。

  RPC可以屏蔽了底层的网络通信细节，使得程序间的远程通信如同本地调用一样简单。RPC机制使得开发者能够构建分布式计算系统，其中不同的组件可以分布在不同的计算机上，但它们之间可以像在同一台机器上一样相互调用。

  

1.2 rpc的原理

  服务提供者实现特定业务逻辑并注册到服务中心。

  服务中心接收注册信息，为服务消费者提供服务发现通知和负载均衡功能。

  服务消费者向服务中心查询服务订阅后，通过 RPC 机制发起对服务提供者的远程调用，接收并处理结果。

  

  

   rpc一次调用的过程
  
  在 RPC（远程过程调用）机制中，客户端以接口方式调用服务。客户端存根在接收到调用请求后，将方法、入参等信息组装并序列化成可网络传输的消息体（二进制流），然后找到远程服务地址，通过网络（sockets）将消息发送给服务端。

  服务端存根收到消息后进行反序列化操作（将二进制流反序列化为消息对象），接着调用本地服务进行处理。服务端将处理结果返回给服务端存根，服务端存根序列化处理结果（将结果消息对象序列化为二进制流）。

  再通过网络（sockets）发送至客户端。客户端存根接收到消息后进行反序列化解码（将结果二进制流反序列化为消息对象），最终客户端得到结果。

  

  

1.3 grpc和brpc

  gRPC

  由 Google 开发并开源。

  支持 C++、Java、Python、Go 。

  使用 Protocol Buffers 作为默认的接口定义语言和数据序列化格式。

  基于 HTTP/2 协议，支持双向流和多路复用，能够有效地提高网络传输效率。

  适用于构建分布式系统、微服务架构等场景。

  

  bRPC

  由 Baidu 开发并开源。

  只支持 C++ 。

  使用 Protocol Buffers 作为默认的接口定义语言和数据序列化格式。

  可以在一个端口上支持多协议服务，如 HTTP/HTTPS、Redis、Thrift 等，具有很高的灵活性。

  提供了一些独特的功能，如通过 HTTP 界面调试服务、使用各种性能分析工具等。

  适用于搜索、存储、机器学习等高性能系统的开发。

  

2. brpc的安装

  在 Linux 系统（ubuntu） 上安装 brpc 的基本步骤：
  
  先安装依赖：

sudo apt-get install -y git g++ make libssl-dev libprotobuf-dev libprotoc-dev protobuf-compiler libleveldb-dev

  安装 brpc：

git clone https://github.com/apache/brpc.git

cd brpc/

mkdir build && cd build

cmake -DCMAKE_INSTALL_PREFIX=/usr .. 

cmake --build . -j6

make && sudo make install

  

3. brpc使用

3.1 brpc接口介绍

  日志输出类与接口：

  包含头文件： #include <butil/logging.h>

namespace logging {
enum LoggingDestination {LOG_TO_NONE = 0
};
struct BUTIL_EXPORT LoggingSettings {LoggingSettings();LoggingDestination logging_dest;
};
bool InitLogging(const LoggingSettings& settings);
}

  
  protobuf 类与接口：

namespace google {
namespace protobuf {class PROTOBUF_EXPORT Closure {public:Closure() {}virtual ~Closure();virtual void Run() = 0;
};
inline Closure* NewCallback(void (*function)());
class PROTOBUF_EXPORT RpcController {bool Failed();std::string ErrorText() ;
} 
} 
}

  
  服务端类与接口：

namespace brpc {
struct ServerOptions {//无数据传输，则指定时间后关闭连接int idle_timeout_sec; // Default: -1 (disabled)int num_threads; // Default: #cpu-cores//....
}
enum ServiceOwnership {//添加服务失败时，服务器将负责删除服务对象SERVER_OWNS_SERVICE,//添加服务失败时，服务器也不会删除服务对象SERVER_DOESNT_OWN_SERVICE
};
class Server {int AddService(google::protobuf::Service* service,ServiceOwnership ownership);int Start(int port, const ServerOptions* opt);int Stop(int closewait_ms/*not used anymore*/);int Join();//休眠直到 ctrl+c 按下，或者 stop 和 join 服务器void RunUntilAskedToQuit();
} 
class ClosureGuard {explicit ClosureGuard(google::protobuf::Closure* done);~ClosureGuard() { if (_done) _done->Run(); }
} 
class HttpHeader {void set_content_type(const std::string& type)const std::string* GetHeader(const std::string& key)void SetHeader(const std::string& key, const std::string& value);const URI& uri() const { return _uri; }HttpMethod method() const { return _method; }void set_method(const HttpMethod method)int status_code()void set_status_code(int status_code);} 
class Controller : public google::protobuf::RpcController {void set_timeout_ms(int64_t timeout_ms);void set_max_retry(int max_retry);google::protobuf::Message* response();HttpHeader& http_response();HttpHeader& http_request();bool Failed();std::string ErrorText();using AfterRpcRespFnType = std::function<void(Controller* cntl,const google::protobuf::Message* req,const google::protobuf::Message* res)>;void set_after_rpc_resp_fn(AfterRpcRespFnType&& fn)
}

  
  客户端类与接口：

namespace brpc {
struct ChannelOptions {//请求连接超时时间int32_t connect_timeout_ms;// Default: 200 (milliseconds)//rpc 请求超时时间int32_t timeout_ms;// Default: 500 (milliseconds)//最大重试次数int max_retry;// Default: 3//序列化协议类型 options.protocol = "baidu_std";AdaptiveProtocolType protocol;//....
} 
class Channel : public ChannelBase {//初始化接口，成功返回 0；int Init(const char* server_addr_and_port, const ChannelOptions* options);

  

4. brpc使用测试

4.1 brpc同步和异步调用

  server.hpp

#include <brpc/server.h>
#include <butil/logging.h>
#include "main.pb.h"// 1. 继承于EchoService创建一个子类，并实现rpc调用的业务功能
class EchoServiceImpl : public example::EchoService 
{public:EchoServiceImpl(){}~EchoServiceImpl(){}void Echo(google::protobuf::RpcController* controller,const ::example::EchoRequest* request,::example::EchoResponse* response,::google::protobuf::Closure* done) {brpc::ClosureGuard rpc_guard(done); // 对象析构时自动调用done->Run()std::cout << "收到消息:" << request->message() << std::endl;std::string str = request->message() + "--这是响应！！";response->set_message(str);//done->Run();}
};int main(int argc, char *argv[])
{// 关闭brpc的默认日志输出logging::LoggingSettings settings;settings.logging_dest = logging::LoggingDestination::LOG_TO_NONE;logging::InitLogging(settings);// 2. 构造服务器对象brpc::Server server;// 3. 向服务器对象中，新增EchoService服务EchoServiceImpl echo_service; // 添加服务失败时，服务器不会删除服务对象int ret = server.AddService(&echo_service, brpc::ServiceOwnership::SERVER_DOESNT_OWN_SERVICE);if (ret == -1) {std::cout << "添加Rpc服务失败！\n";return -1;}// 4. 启动服务器brpc::ServerOptions options;options.idle_timeout_sec = -1; // 连接空闲超时时间-超时后连接被关闭options.num_threads = 1; // io线程数量ret = server.Start(8080, &options);if (ret == -1) {std::cout << "启动服务器失败！\n";return -1;}server.RunUntilAskedToQuit(); // 休眠等待运行结束，避免对SIGINT信号进行处理return 0;
}

  makefile:

all : server client
server : server.cc main.pb.ccg++ -std=c++17 $^ -o $@ -lbrpc -lgflags -lssl -lcrypto -lprotobuf -lleveldb
client : client.cc main.pb.ccg++ -std=c++17 $^ -o $@ -lbrpc -lgflags -lssl -lcrypto -lprotobuf -lleveldb

  

  同步调用：同步调用是指客户端会阻塞收到 server 端的响应或发生错误。

#include <brpc/channel.h>
#include <thread>
#include "main.pb.h"int main(int argc, char *argv[])
{// 1. 构造Channel信道，连接服务器brpc::ChannelOptions options;options.connect_timeout_ms = -1; // 连接等待超时时间，-1表示一直等待options.timeout_ms = -1; // rpc请求等待超时时间，-1表示一直等待options.max_retry = 3; // 请求重试次数options.protocol = "baidu_std"; // 序列化协议，默认使用baidu_stdbrpc::Channel channel;int ret = channel.Init("127.0.0.1:8080", &options);if (ret == -1) {std::cout << "初始化信道失败！\n";return -1;}// 2. 构造EchoService_Stub对象，用于进行rpc调用example::EchoService_Stub stub(&channel);// 3. 进行Rpc调用example::EchoRequest req;req.set_message("你好~brpc~！");brpc::Controller cntl;example::EchoResponse rsp;stub.Echo(&cntl, &req, &rsp, nullptr); // 同步调用if (cntl.Failed() == true) {std::cout << "Rpc调用失败：" << cntl.ErrorText() << std::endl;return -1;}std::cout << "同步调用结束！\n";std::cout << "收到响应: " << rsp.message() << std::endl;return 0;
}

  

  

  

  异步调用：异步调用是指客户端注册一个响应处理回调函数， 当调用一个 RPC 接口时立即返回，不会阻塞等待响应， 当 server 端返回响应时会调用传入的回调函数处理响应。

#include <brpc/channel.h>
#include <thread>
#include "main.pb.h"// 这个回调函数用于异步调用对象使用
void callback(brpc::Controller* cntl, ::example::EchoResponse* response) 
{std::unique_ptr<brpc::Controller> cntl_guard(cntl);std::unique_ptr<example::EchoResponse> resp_guard(response);if (cntl->Failed() == true) {std::cout << "Rpc调用失败：" << cntl->ErrorText() << std::endl;return;}std::cout << "收到响应: " << response->message() << std::endl;// delete cntl;// delete response;
}int main(int argc, char *argv[])
{// 1. 构造Channel信道，连接服务器brpc::ChannelOptions options;options.connect_timeout_ms = -1; // 连接等待超时时间，-1表示一直等待options.timeout_ms = -1; // rpc请求等待超时时间，-1表示一直等待options.max_retry = 3; // 请求重试次数options.protocol = "baidu_std"; // 序列化协议，默认使用baidu_stdbrpc::Channel channel;int ret = channel.Init("127.0.0.1:8080", &options);if (ret == -1) {std::cout << "初始化信道失败！\n";return -1;}// 2. 构造EchoService_Stub对象，用于进行rpc调用example::EchoService_Stub stub(&channel);// 3. 进行Rpc调用example::EchoRequest req;req.set_message("你好~brpc~！");// 防止异步调用时 cnt rsp 出作用域被销毁，要在堆上创建对象brpc::Controller *cntl = new brpc::Controller();example::EchoResponse *rsp = new example::EchoResponse();auto clusure = google::protobuf::NewCallback(callback, cntl, rsp); // 异步调用对象stub.Echo(cntl, &req, rsp, clusure); // 异步调用std::cout << "异步调用结束！\n";std::this_thread::sleep_for(std::chrono::seconds(3));return 0;
}