brpc 安装及使用

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();}}
}

回调机制

• Closure 是异步 RPC 的完成通知载体，NewCallback 为其提供快速创建的辅助函数。

• RpcController 通过 Failed() 和 ErrorText() 提供回调中可能需要的错误信息。

1. Closure 类

作用：定义了一个通用的、可扩展的回调操作，用于异步操作完成后的通知。子类必须实现 Run() 方法，定义具体的回调逻辑。

核心方法

设计意图

• 解耦异步操作与回调逻辑：允许用户自定义回调行为（如释放资源、触发事件等）。

• 与 Protobuf RPC 集成：广泛用于 RPC 完成后的异步通知。

示例用法

class MyCallback : public google::protobuf::Closure {
public:void Run() override {std::cout << "Callback executed!" << std::endl;delete this; // 自管理内存}
};
// 使用
google::protobuf::Closure* callback = new MyCallback();
callback->Run(); // 输出 "Callback executed!"

2. NewCallback 函数

作用：将普通函数（无参数、无返回值）包装为 Closure 对象。避免手动子类化，简化一次性回调的创建。

示例

void OnRpcDone() {std::cout << "RPC finished!" << std::endl;
}// 创建回调
google::protobuf::Closure* callback = google::protobuf::NewCallback(&OnRpcDone);
callback->Run(); // 输出 "RPC finished!"
delete callback; // 需手动释放

3. RpcController 类

作用：提供 RPC 执行过程中的错误处理和状态查询。

核心方法

典型用途

brpc::Controller cntl;  // brpc 的实现类
MyService_Stub stub(&channel);
MyRequest request;
MyResponse response;// 发起异步 RPC
stub.MyMethod(&cntl, &request, &response, NewCallback(&OnRpcDone));
if (cntl.Failed()) {std::cerr << "RPC failed: " << cntl.ErrorText() << std::endl;
} 

服务端类与接口

protoc --cpp_out=./ main.proto

创建服务端源码 - server.cpp

#include <iostream>
#include <brpc/server.h>
#include <butil/logging.h>
#include "main.pb.h"
// 继承EchoService创建一个子类，并实现rpc调用的业务功能
class EchoServiceImpl : public example::EchoService
{
public:void Echo(google::protobuf::RpcController *controller,const example::EchoRequest *request, example::EchoResponse *response, google::protobuf::Closure *done){brpc::ClosureGuard guard(done);std::cout << "收到消息：" << request->message() << std::endl;std::string str = "这是响应--" + request->message();response->set_message(str);// done->Run(); 由于使用了ClosureGuard，异步完成时就不需要手动调用Run函数}
};
int main()
{// 关闭brpc的默认日志输出logging::LoggingSettings settings;settings.logging_dest = logging::LoggingDestination::LOG_TO_NONE;logging::InitLogging(settings);// 构造服务器对象brpc::Server server;// 向服务器对象添加EchoService服务EchoServiceImpl echo_service;int ret = server.AddService(&echo_service, brpc::ServiceOwnership::SERVER_OWNS_SERVICE);if (ret == -1){std::cout << "添加Rpc服务失败！\n";return -1;}// 启动服务器brpc::ServerOptions options;options.idle_timeout_sec = -1; // 连接空闲超时时间，超时后连接被关闭，-1为一直连接options.num_threads = 1;       // io线程数量ret = server.Start(8080, &options);if (ret == -1){std::cout << "启动服务器失败！\n";return -1;}server.RunUntilAskedToQuit(); // 修改等待运行结束return 0;
}

创建客户端源码 - client.cpp

#include <iostream>
#include <thread>
#include <brpc/channel.h>
#include "main.pb.h"
#include <memory>
void callback(brpc::Controller *cntl, example::EchoResponse *response)
{std::unique_ptr<brpc::Controller> cntl_guard(cntl);std::unique_ptr<example::EchoResponse> rsp_guard(response);if (cntl->Failed() == true){std::cout << "Rpc调用失败：" << cntl->ErrorText() << std::endl;return;}std::cout << "收到响应: " << response->message() << std::endl;
}
int main()
{// 构造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;}// 构造EchoService_Stub对象，用于进行rpc调用example::EchoService_Stub stub(&channel);// 进行rpc调用example::EchoRequest req;req.set_message("hello world!");brpc::Controller *cntl = new brpc::Controller();example::EchoResponse *rsp = new example::EchoResponse();// 1.同步调用stub.Echo(cntl, &req, rsp, nullptr);if (cntl->Failed() == true){std::cout << "Rpc调用失败：" << cntl->ErrorText() << std::endl;return -1;}std::cout << "收到响应: " << rsp->message() << std::endl;delete cntl;delete rsp;// 2.异步调用//  auto closure = google::protobuf::NewCallback(callback,cntl,rsp);//  stub.Echo(cntl, &req, rsp, closure);//  std::cout << "异步调用结束！\n";//  std::this_thread::sleep_for(std::chrono::seconds(3));return 0;
}

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

二次封装

• rpc 调用这里的封装，因为不同的服务调用使用的是不同的 Stub，这个封装起来的意义不大，因此我们只需要封装通信所需的 Channel 管理即可，这样当需要进行什么样的服务调用的时候，只需要通过服务名称获取对应的 channel，然后实例化 Stub 进行调用即可。
• 封装 Channel 的管理，每个不同的服务可能都会有多个主机提供服务，因此一个服务可能会对应多个 Channel，需要将其管理起来，并提供获取指定服务 channel 的接口。进行 rpc 调用时，获取 channel，目前以 RR 轮转的策略选择 channel。
• 提供进行服务声明的接口，因为在整个系统中，提供的服务有很多，但是当前可能并不一定会用到所有的服务，因此通过声明来告诉模块哪些服务是自己关心的，需要建立连接管理起来，没有添加声明的服务即使上线也不需要进行连接的建立。
• 提供服务上线时的处理接口，也就是新增一个指定服务的 channel。
• 提供服务下线时的处理接口，也就是删除指定服务下的指定 channel。

#pragma once
#include <iostream>
#include <string>
#include <vector>
#include <unordered_map>
#include <mutex>
#include "logger.hpp"
#include <brpc/channel.h>// 封装单个服务的信道管理类
class ServiceChannel
{
public:using ptr = std::shared_ptr<ServiceChannel>;using ChannelPtr = std::shared_ptr<brpc::Channel>;ServiceChannel(const std::string &name): _service_name(name), _index(0){}// 服务上线了一个节点，则调用append新增信道void append(const std::string &host){// 构造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_stdauto channel = std::make_shared<brpc::Channel>();int ret = channel->Init(host.c_str(), &options);if (ret == -1){LOG_ERROR("初始化{}-{}信道失败！", _service_name, host);return;}std::unique_lock<std::mutex> lock(_mutex);_channels.push_back(channel);_hosts[host] = channel;}// 服务下线了一个节点，则调用remove释放信道void remove(const std::string &host){std::unique_lock<std::mutex> lock(_mutex);auto it = _hosts.find(host);if (it == _hosts.end()){LOG_WARN("删除{}-{}节点信道时，没有找到信道信息！", _service_name, host);return;}for (auto vit = _channels.begin(); vit != _channels.end(); vit++){if (*vit == it->second){_channels.erase(vit);break;}}_hosts.erase(it);}// 通过RR轮转策略，获取一个Channel用于发起对应服务的Rpc调用ChannelPtr choose(){std::unique_lock<std::mutex> lock(_mutex);if (_channels.size() == 0){LOG_ERROR("当前没有能够提供{}服务的节点！", _service_name);return ChannelPtr();}int32_t index = _index++ % _channels.size();return _channels[index];}private:std::mutex _mutex;int32_t _index;                                     // 当前轮转下标的计数器std::string _service_name;                          // 服务名称std::vector<ChannelPtr> _channels;                  // 当前服务对应的信道集合std::unordered_map<std::string, ChannelPtr> _hosts; // 主机地址与信道的映射
};// 总体的服务信道管理类
class ServiceManager
{
public:using ptr = std::shared_ptr<ServiceManager>;// 获取指定服务的节点信道ServiceChannel::ChannelPtr choose(const std::string &service_name){std::unique_lock<std::mutex> lock(_mutex);auto it = _services.find(service_name);if (it == _services.end()){LOG_ERROR("当前没有能够提供{}服务的节点！", service_name);return ServiceChannel::ChannelPtr();}return it->second->choose();}// 先声明我关心哪些服务的上下线，不关心的则不需要管理void declared(const std::string &service_name){std::unique_lock<std::mutex> lock(_mutex);_follow_services.insert(service_name);}// 服务上线时调用的回调接口，将服务节点管理起来void onServiceOnline(const std::string &service_instance, const std::string &host){std::string service_name = getServiceName(service_instance);ServiceChannel::ptr service;{std::unique_lock<std::mutex> lock(_mutex);auto fit = _follow_services.find(service_name);if (fit == _follow_services.end()){LOG_DEBUG("{}-{}服务上线了，但是当前并不关心！", service_name, host);return;}// 先获取管理对象，没有则创建，有则添加节点auto sit = _services.find(service_name);if (sit == _services.end()){service = std::make_shared<ServiceChannel>(service_name);_services[service_name] = service;}elseservice = sit->second;}if (!service){LOG_ERROR("新增{}服务管理节点失败！", service_name);return;}service->append(host);LOG_DEBUG("{}-{} 服务上线新节点，进行添加管理！", service_name, host);}// 服务下线时调用的回调接口，从相应的服务节点删除信道void onServiceOffline(const std::string &service_instance, const std::string &host){std::string service_name = getServiceName(service_instance);ServiceChannel::ptr service;{std::unique_lock<std::mutex> lock(_mutex);auto fit = _follow_services.find(service_name);if (fit == _follow_services.end()){LOG_DEBUG("{}-{}服务下线了，但是当前并不关心！", service_name, host);return;}// 先获取管理对象，有则删除节点信道auto sit = _services.find(service_name);if (sit == _services.end()){LOG_WARN("删除{}服务节点时，没有找到管理对象", service_name);return;}elseservice = sit->second;}service->remove(host);LOG_DEBUG("{}-{} 服务下线节点，进行删除管理！", service_name, host);}private:std::string getServiceName(const std::string &service_instance){auto pos = service_instance.find_last_of('/');if (pos == std::string::npos)return service_instance;elsereturn service_instance.substr(0, pos);}private:std::mutex _mutex;std::unordered_set<std::string> _follow_services;std::unordered_map<std::string, ServiceChannel::ptr> _services;
};