brpc 与 Etcd 二次封装
brpc二次封装
封装思想
二次封装原因分析
brpc的核心功能是RPC调用，但是在分布式系统中，单纯的RPC调用无法满足需求，需要考虑清楚三个问题
- 提供服务的主体：服务发现，例如通过注册中心（Etcd）动态获取服务列表（后续会将两者借结合）
- 高效管理服务通信的方式：负载均衡、连接复用等
- 扩展和维护服务端实现：动态增减服务节点，避免大量代码被修改
综上所述，brpc的二次封装主要目的是信道管理，不是对RPC方法的封装
核心思想
指定服务的信道管理类
- 一个服务可能会有多个节点提供服务，每个节点都对应一个独立的Channel
- 封装的目标就是将这些节点的信道管理起来，建立服务于信道之间的映射关系（一般情况下应该是多对一的映射关系，也就是一个服务名对应多个节点的Channel）
- 负载均衡实现：通过轮询策略，在多个节点中选择一个Channel发起请求
全局信道管管理类
- 对多个服务的信道进行统一管理
- 提供一个全局的入口，便于通过服务名称动态获取对应的信道
- 动态加载或者移除服务信道，支持线上服务的动态拓展和下线
封装实现逻辑分析
- 服务与信道的管理
- 将每个服务节点的信道管理起来，每个节点有独立的Channel
- 通过服务ID找到对应的信道组
- 发起RPC调用时的信道选择
- 根据服务名称获取信道组；在信道组中根据负载均衡策略选择一个信道
- 使用选择的信道发起RPC调用
- 动态拓展
- 当新的服务节点上线时，动态创建并添加信道
- 当服务节点下线时，动态移除信道，保证资源回收
具体实现
#pragma once
#include <brpc/channel.h>
#include <string>
#include <vector>
#include <unordered_map>
#include <mutex>
#include "logger.hpp"
namespace mag {
// 封装单个服务的信道管理类
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) {}
// 服务上线了一个节点时调用，新增信道
void append(const std::string &host) {
auto channel = std::make_shared<brpc::Channel>();
brpc::ChannelOptions options;
options.connect_timeout_ms = -1; // 设置连接超时为无限等待
options.timeout_ms = -1; // 设置请求超时为无限等待
options.max_retry = 3; // 设置最大重试次数
options.protocol = "baidu_std"; // 使用百度标准协议
// 初始化信道
int ret = channel->Init(host.c_str(), &options);
if (ret == -1) {
LOG_ERROR("初始化{}-{}信道失败!", _service_name, host);
return;
}
// 加锁保护，更新信道集合和主机映射
std::unique_lock<std::mutex> lock(_mutex);
_hosts.insert(std::make_pair(host, channel));
_channels.push_back(channel);
}
// 服务下线了一个节点时调用，移除信道
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 idx = _index++ % _channels.size();
return _channels[idx];
}
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>;
// 构造函数
ServiceManager() {}
// 获取指定服务的节点信道
ServiceChannel::ChannelPtr choose(const std::string &service_name) {
std::unique_lock<std::mutex> lock(_mutex);
auto sit = _services.find(service_name);
if (sit == _services.end()) {
LOG_ERROR("当前没有能够提供 {} 服务的节点！", service_name);
return ServiceChannel::ChannelPtr();
}
return sit->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.insert(std::make_pair(service_name, service));
} else {
service = 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;
}
service = 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;
return service_instance.substr(0, pos);
}
private:
std::mutex _mutex; // 互斥锁，保护多线程环境下的成员变量
std::unordered_set<std::string> _follow_services; // 需要关注的服务名称集合
std::unordered_map<std::string, ServiceChannel::ptr> _services; // 服务名称到信道管理对象的映射关系
};
} Etcd 和 brpc两者结合
核心思想理解
背景理解成在一个城市中找咖啡店购买咖啡
Channel
- 功能：负责通信，将顾客（请求）送达目标分店（服务节点）
- Channel就像是一条连接顾客和咖啡店的 道路
- 每条道路对应一个服务节点（例如不同的咖啡店分店）；如果有多个分店（节点），会有多条道路，每条道路都是独立的
服务信道管理类
- 这是一个管理所有道路的“导航系统”
- 管理所有通往咖啡店的道路（信道），记录哪些道路有效，哪些需要维护
- 每次有顾客需要服务时，根据具体规则（例如最近距离或排队时间最短）选择最优的一条道路
顾客想要一杯咖啡，但不知道具体该走哪条路，这时服务信道管理类会告诉你：走 “道路1” 或 “道路2”
Server（咖啡店）
- 咖啡店作为服务端，负责提供咖啡服务
- 主要工作
- 注册所有的服务（菜单上列出的咖啡种类）
- 启动服务端（开店营业）
- 监听来自顾客的请求（顾客通过不同的道路来到店里）
当一个服务节点（分店）注册成功后，顾客就可以通过对应的道路来访问这个节点
服务注册中心（类似于咖啡店地图）
- 服务注册中心（如 etcd）就像是一个共享的咖啡店地图，它记录了所有的分店地址
- 存储服务节点的信息（例如每个分店的位置）
- 动态更新服务节点状态（某个分店是否营业、新开分店等）
当某个分店暂停营业时，注册中心会通知所有导航系统（服务信道管理类），将对应的道路标记为不可用
Controller（请求的调度和跟踪）
- Controller 就像一个订单管理系统，用于追踪顾客的请求是否成功
- 记录每个请求的状态（比如：订单是否成功、是否被取消）
- 跟踪请求的详细信息（比如：是否在途中发生了错误，或者咖啡店是否超时未响应）
顾客点了一杯咖啡，但发现等待超时了，Controller 会记录错误信息并反馈给顾客
负载均衡策略（防止一家咖啡店爆单）
- 当一个城市有多个咖啡店分店，导航系统需要选择一个合适的分店为顾客服务
- 实现负载均衡，例如通过轮询（Round Robin，RR）选择最少排队的分店
顾客想买咖啡，导航系统会选择离顾客最近的分店，或者选择当前空闲状态的分店
回调机制（异步处理通知）
- 顾客下单后，不需要一直等待，可以去做其他事情。当订单完成时，咖啡店会通知顾客“咖啡做好了”
- 在异步调用中，处理请求完成后的逻辑；例如顾客下单后，回调函数可以执行通知、记录订单状态等任务
顾客下单后离开座位，咖啡做好时，店员通过铃声（回调函数）通知顾客
ClosureGuard（保证回调机制一定会执行机制）
- ClosureGuard 就像是一个“催促器”，确保咖啡做好后一定会通知顾客
- 确保异步请求完成时回调函数一定会被执行；即使在处理过程中发生错误，回调函数也能被调用，避免资源泄漏
店员忘记通知顾客，但系统（ClosureGuard）会强制通知，保证订单状态最终得到更新
etcd与RPC结合逻辑
Etcd：咖啡店地图和动态信息管理员
Etcd 在这个系统中负责 服务的注册和发现，就像是一个 咖啡店地图和动态信息管理员
- 记录了所有分店的位置（也就是记录了服务节点的地址），Etcd通过路径管理对应的服务
- 服务上线的时候新增一条道路，服务下线的时候则通知系统移除这条道路；Discovery类则相当于管理员，其负责Etcd通信，监听咖啡店上下线动态
ServiceManager：导航系统
主要功能就是路由选择和动态调整
- 当顾客（RPC 调用）要去购买咖啡时，它会根据地图（Etcd 提供的服务信息），选择一个最近或最优的咖啡店分店
- 它会根据负载均衡策略（比如轮询，Round Robin），动态地在分店中选择一个路径
服务节点注册逻辑
- 它会把自己的位置（IP 和端口）上传到地图（Etcd）
- 地图管理员（Discovery）会通知导航系统（ServiceManager）新增这个分店的信息
- 导航系统为这家分店创建一条通路（信道）
客户端发起RPC调用
- 顾客（客户端）使用导航系统（ServiceManager）查找最近的咖啡店
- 导航系统根据地图信息，找到一个分店地址，并返回对应的通路（信道）
负载均衡
- 导航系统会根据负载均衡策略（如轮询），在多个分店中选择一个最优的
- 如果分店关门（信道不可用），导航系统会选择其他分店
RPC调用-开始点单
顾客通过信道（Channel）发送点单请求（RPC 调用），并等待咖啡做好（服务响应）
收到响应
- 如果点单成功，顾客会收到咖啡（RPC 响应消息）
- 如果失败，顾客会稍等片刻并重试，尝试其他分店
具体实现
服务发现
核心逻辑
- Etcd动态服务发现
- Etcd提供服务节点的注册和发现能力，监听服务节点的上线和下线事件
- 通过Discovery类将服务节点的变化回调到ServiceManager，然后动态更新信道管理
- 动态信道管理
- ServiceManager管理所有的服务信道，同时通过负载均衡动态选择信道
- 结合 Etcd 服务发现，实现动态扩展和容灾
- 基于信道的RPC调用
- 通过动态选择的信道，发起RPC调用
具体实现代码-服务发现
//
///服务端逻辑
/
#include "../common/etcd.hpp" // Etcd 服务发现和注册相关头文件
#include <gflags/gflags.h> // Google Flags，用于解析命令行参数
#include <thread> // C++ 标准库的线程管理头文件
#include <brpc/server.h> // brpc 服务端相关头文件
#include <butil/logging.h> // brpc 默认日志设置
#include "main.pb.h" // Protobuf 生成的 RPC 服务头文件
// 定义命令行参数
DEFINE_bool(run_mode, false, "程序的运行模式，false-调试； true-发布；");
DEFINE_string(log_file, "", "发布模式下，用于指定日志的输出文件");
DEFINE_int32(log_level, 0, "发布模式下，用于指定日志输出等级");
DEFINE_string(etcd_host, "http://127.0.0.1:2379", "服务注册中心地址");
DEFINE_string(base_service, "/service", "服务监控根目录");
DEFINE_string(instance_name, "/echo/instance", "当前实例名称");
DEFINE_string(access_host, "127.0.0.1:7070", "当前实例的外部访问地址");
DEFINE_int32(listen_port, 7070, "Rpc服务器监听端口");
// 实现 Echo 服务
class EchoServiceImpl : public example::EchoService {
public:
EchoServiceImpl() {}
~EchoServiceImpl() {}
// 重写 Echo 方法，用于处理客户端请求
void Echo(google::protobuf::RpcController* controller,
const ::example::EchoRequest* request,
::example::EchoResponse* response,
::google::protobuf::Closure* done) override {
// 用于确保在处理完成后正确调用回调函数
brpc::ClosureGuard rpc_guard(done);
// 打印接收到的请求消息
std::cout << "收到消息: " << request->message() << std::endl;
// 构造响应消息
std::string str = request->message() + "--这是响应！！";
response->set_message(str);
}
};
int main(int argc, char *argv[]) {
// 1. 解析命令行参数并初始化日志
google::ParseCommandLineFlags(&argc, &argv, true);
init_logger(FLAGS_run_mode, FLAGS_log_file, FLAGS_log_level);
// 2. 关闭 brpc 的默认日志输出
logging::LoggingSettings settings;
settings.logging_dest = logging::LoggingDestination::LOG_TO_NONE; // 禁用 brpc 默认日志
logging::InitLogging(settings);
// 3. 创建 RPC 服务端对象
brpc::Server server;
// 4. 添加服务到服务端对象
EchoServiceImpl echo_service;
int ret = server.AddService(&echo_service, brpc::ServiceOwnership::SERVER_DOESNT_OWN_SERVICE);
if (ret == -1) {
std::cerr << "添加 Rpc 服务失败！" << std::endl;
return -1;
}
// 5. 启动服务器
brpc::ServerOptions options;
options.idle_timeout_sec = -1; // 设置连接空闲超时时间，-1 表示不超时
options.num_threads = 1; // 设置 IO 线程数量
ret = server.Start(FLAGS_listen_port, &options);
if (ret == -1) {
std::cerr << "启动服务器失败！" << std::endl;
return -1;
}
// 6. 注册服务到 Etcd
Registry::ptr rclient = std::make_shared<Registry>(FLAGS_etcd_host);
rclient->registry(FLAGS_base_service + FLAGS_instance_name, FLAGS_access_host); // 服务注册
// 7. 启动服务器并等待终止信号
server.RunUntilAskedToQuit();
return 0;
}///
///客户端逻辑
//
#include "../common/etcd.hpp" // Etcd服务发现相关头文件
#include "../common/channel.hpp" // 服务信道管理相关头文件
#include <gflags/gflags.h> // Google Flags，用于解析命令行参数
#include <thread> // C++标准库的线程管理头文件
#include "main.pb.h" // Protobuf生成的RPC服务头文件
// 定义命令行参数
DEFINE_bool(run_mode, false, "程序的运行模式，false-调试； true-发布；");
DEFINE_string(log_file, "", "发布模式下，用于指定日志的输出文件");
DEFINE_int32(log_level, 0, "发布模式下，用于指定日志输出等级");
DEFINE_string(etcd_host, "http://127.0.0.1:2379", "服务注册中心地址");
DEFINE_string(base_service, "/service", "服务监控根目录");
DEFINE_string(call_service, "/service/echo", "需要调用的服务路径");
int main(int argc, char *argv[]) {
// 解析命令行参数并初始化日志
google::ParseCommandLineFlags(&argc, &argv, true);
init_logger(FLAGS_run_mode, FLAGS_log_file, FLAGS_log_level);
// 1. 创建服务信道管理对象（ServiceManager）
// - 用于动态管理服务信道（服务节点的通信通道）。
auto sm = std::make_shared<ServiceManager>();
sm->declared(FLAGS_call_service); // 声明需要管理的服务（/service/echo）
// 绑定服务发现的回调函数
auto put_cb = std::bind(&ServiceManager::onServiceOnline, sm.get(), std::placeholders::_1, std::placeholders::_2);
auto del_cb = std::bind(&ServiceManager::onServiceOffline, sm.get(), std::placeholders::_1, std::placeholders::_2);
// 2. 创建服务发现客户端（Discovery）
// - 连接到Etcd，监听服务节点的上下线事件。
Discovery::ptr dclient = std::make_shared<Discovery>(FLAGS_etcd_host, FLAGS_base_service, put_cb, del_cb);
while (true) {
// 3. 通过服务信道管理对象选择一个信道（Channel）
// - 信道用于与服务节点通信。
auto channel = sm->choose(FLAGS_call_service);
if (!channel) {
std::this_thread::sleep_for(std::chrono::seconds(1)); // 若没有可用信道，等待1秒后重试
return -1;
}
// 4. 使用信道创建服务存根（Stub），发起RPC调用
example::EchoService_Stub stub(channel.get());
example::EchoRequest req; // 创建请求对象
req.set_message("你好~etcd&brpc~！"); // 设置请求内容
brpc::Controller *cntl = new brpc::Controller(); // 创建RPC调用的控制器
example::EchoResponse *rsp = new example::EchoResponse(); // 创建响应对象
// 发起同步RPC调用
stub.Echo(cntl, &req, rsp, nullptr);
// 5. 处理RPC调用结果
if (cntl->Failed()) {
// 如果调用失败，打印错误信息并释放资源
std::cout << "Rpc调用失败：" << cntl->ErrorText() << std::endl;
delete cntl;
delete rsp;
std::this_thread::sleep_for(std::chrono::seconds(1)); // 等待1秒后重试
continue;
}
// 如果调用成功，输出服务端返回的响应消息
std::cout << "收到响应: " << rsp->message() << std::endl;
// 释放资源并等待1秒后进行下一次调用
delete cntl;
delete rsp;
std::this_thread::sleep_for(std::chrono::seconds(1));
}
return 0;
}总结
客户端逻辑
实现了 RPC 客户端，用于通过 Etcd 动态发现服务并调用远程服务
- 动态发现服务：监听 Etcd，获取服务端的地址信息
- 通过负载均衡选择一个可用的服务节点（信道
Channel） - 使用动态选择的信道发起 RPC 请求，获取服务端返回的响应
服务端逻辑
- 提供服务：启动一个
EchoService服务，用于处理客户端的请求 - 服务注册：将自身服务地址注册到 Etcd，以供客户端发现