RPC 原理与 BRPC 实战：基于 C++ 的分布式通信实现

简单说：C++ 项目追求性能和省心，选 BRPC；需要多语言支持，用 gRPC。

三、基于 BRPC 实现简单的服务调用

上面说了对应的 brpc 的优点，因此这里可以看出更适合用 brpc 而不是 grpc 来实现。

BRPC 安装教程

安装步骤：

1. 安装依赖：

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

2. 安装 brpc：

git clone https://github.com/apache/brpc.git
cd brpc/
mkdir build && cd build
cmake -DCMAKE_INSTALL_PREFIX=/usr ..
cmake --build . -j6
sudo make install

简单实现客户端向 brpc 服务端口请求服务完成应答过程（以 echo 回显为例）

首先先说下对应思路：

• rpc 服务端：进行对应 echo 服务注册（通过 Protobuf 协议为服务端与客户端生成对应类，然后服务端填充对应的 echo 功能完成添加进入 rpc-server 中），最后启动对应 server，也就是把 rpc 服务部署在对应 ip+port 处。
• rpc 客户端：通过约定好的 Protobuf 协议生成的对应调用相关 rpc 服务的 rpc 请求接口进行调用（拿对应 rpc 服务信道进行初始化）；调用 rpc 服务请求接口发送对应请求等。
• rpc 服务端：收到对应请求就去拿着请求调用用户注册进来的服务函数进行操作，最后填充好答复以及其他相关信息发送回去。
• rpc 客户端：收到对应答复后进行解析拿到结果以及其他信息，然后调用回调函数进行操作或者通过答复结果进行其他处理操作。

简单来说分为 rpc 服务端与 rpc 客户端（具体操作如下）：

• 1·rpc 服务端：关闭默认日志 + 构建服务对象向 rpc-server 中添加 + 启动 rpc 服务器。
• 2·rpc 客户端：初始化信道 + 构建并发送 rpc 请求对象 + 调用回调恢复（可选）。

测试效果

• rpc-server 注册完对应的 echo 服务后启动在这等待。
• 服务端接收到客户端的请求进行相关服务处理（echo 服务处理函数）。
• rpc-client 通过特定接口进行 rpc 请求，之后拿到对应响应答复后异步调用回调进行操作。

代码汇总

对应代码及详细注释：

1. Protobuf 用于后续设定对应 rpc 服务：

syntax="proto3";
package example;
option cc_generic_services = true;

message EchoRequest {
    string message = 1;
}

message EchoResponse {
    string message = 1;
}

// 搭建对应的服务，然后为这个服务实例化出模版 echo 基类提供用户填充
service EchoService {
    rpc Echo(EchoRequest) returns (EchoResponse);
}

2. Makefile（注意这里链接的所有库及顺序）：

all : server client

server: server.cc echo.pb.cc
	g++ -g -std=c++17 $^ -o $@ -L/usr/local/lib -lspdlog -lfmt -letcd-cpp-api -lcpprest -lbrpc -lgflags -lssl -lcrypto -lprotobuf -lleveldb

client: client.cc echo.pb.cc
	g++ -g -std=c++17 $^ -o $@ -L/usr/local/lib -lspdlog -lfmt -letcd-cpp-api -lcpprest -lbrpc -lgflags -lssl -lcrypto -lprotobuf -lleveldb

.PHONY: clean
clean:
	rm -r server client
# brpc 库依赖 glags 库，故必须把 gflags 链接在 brpc 库后面，否则报错
# 这里链接的 Protobuf 库的版本必须和 protoc 的时候是一样的否则报错
# -L/usr/local/lib 当前没有对应库再去系统默认的地方链接找库

3. rpc-client:

#include <brpc/channel.h>
#include <thread>
#include "echo.pb.h"

void clientcallback(brpc::Controller *cntl, ::example::EchoResponse *response) {
    // 把指针直接给智能指针对象此时直接被智能指针保存地址即可，智能指针就不用在内部 new 了，出了函数作用自动帮忙把客户端构建对象析构释放
    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;
}

int main() {
    // 客户端调用 rpc 接口通过信道与 rpc 服务端进行通信
    brpc::ChannelOptions options;
    options.connect_timeout_ms = -1; // 连接等待超时时间，-1 表示一直等待
    options.timeout_ms = -1;         // rpc 请求等待超时时间，-1 表示一直等待
    options.max_retry = 3;           // 请求重试次数
    options.protocol = "baidu_std";  // 序列化协议，默认使用 baidu_std

    brpc::Channel channel;
    auto 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);
    brpc::Controller *cntl = new brpc::Controller();
    example::EchoResponse *rsp = new example::EchoResponse();
    example::EchoRequest req;
    req.set_message("你好 rpc！");

    auto closure = google::protobuf::NewCallback(clientcallback, cntl, rsp);
    // 异步线程出动走 closure 调用也就是回调函数的处理，主线程收到答复填充完接着往下走
    // 主线程这里收到对应答复进行填充好对应的内容如 rsp cntl 等然后如果 closure 不为空就获取对应回调通知然后派异步线程去执行对应 clientcallback 回调函数处理即可
    stub.Echo(cntl, &req, rsp, closure); // 向 rpcserver 发送对应 req，然后接收到 response 填充 rsp 及 cntl，之后再去调用 closure 这个可调用对象即调用 clientcallback
    std::cout << "异步调用！\n";
    std::this_thread::sleep_for(std::chrono::seconds(3));
    return 0;
}

4. rpc-server:

#include <brpc/server.h>
#include <butil/logging.h>
#include "echo.pb.h"

class EchoService : public example::EchoService {
    void Echo(google::protobuf::RpcController *controller,
              const ::example::EchoRequest *request,
              ::example::EchoResponse *response,
              ::google::protobuf::Closure *done) {
        // 创建一个 brpc::ClosureGuard 对象，用于管理 done 的生命周期。当 rpc_guard 对象析构时，会自动调用 done，通知框架 RPC 处理完成
        brpc::ClosureGuard rpc_guard(done);
        // 这里析构的时候调用对应的 done->run（）构建对应答复信息来通知请求方可以进行自己的回调处理了
        std::cout << "收到消息：" << request->message() << std::endl;
        std::string str = request->message() + "--这是响应！！";
        response->set_message(str);
        // 调用完对应自定义函数 echo 后由 rpc 发送对应答复集合
    }
};

int main() {
    // 关闭 brpc 默认输出日志
    logging::LoggingSettings logset;
    logset.logging_dest = logging::LoggingDestination::LOG_TO_NONE;
    logging::InitLogging(logset);

    // 创建 rpc 服务器注册对应服务
    brpc::Server server;
    EchoService echo;
    server.AddService(&echo, brpc::ServiceOwnership::SERVER_DOESNT_OWN_SERVICE);

    // 添加选项并启动 rpc 服务
    brpc::ServerOptions options;
    options.idle_timeout_sec = -1; // 连接空闲超时时间 - 超时后连接被关闭
    options.num_threads = 1;       // io 线程数量

    auto ret = server.Start(8080, &options); // rpc 服务端对应的 echo 服务启动了：也就是收到对应 echo 的 resquest 就去调用对应 Echo 函数完成 response 构建然后自动发送
    if (ret == -1) {
        std::cout << "启动服务器失败！\n";
        return -1;
    }
    server.RunUntilAskedToQuit(); // 修改等待运行结束
    return 0;
}

四、封装每个服务的 channels 及所有服务管理者

rpc 调用这里的封装，因为不同的服务调用使用的是不同的 Stub，这个封装起来的意义不大，因此这里封装的是每个服务占用的信道集合和管理起来所有服务。

封装思想（需要封装的两大类分别是 service-channels 与 servicesmanager）：

1. ServiceChannels：

• 首先要知道注册中心注册（etcd）注册对应服务是以服务 + 单例名称 +ip-port 来注册的，也就是说明一个指定服务可以有多个单例来运行（即多个 rpc 服务，可以是在不同 ip+port 处，因此就有了多个信道概念），得出结论是一个服务对应多个服务单例。
• ServiceChannels 就是管理一个服务间所有关于它的实例信道（也就是站在某个服务的所有单例对象的信道管理者角度）。
• 因此这个类就需要对应信道的增删查等功能，完成对应主机名与服务信道的映射，以及采取 RR 轮转方式使用对应服务的信道。

2. servicesmanager：

• 这个其实就是把上面的 service-channels 管理起来，暴露给外面使用的，并集合对应 etcd 的 watch 监控来用，当用户线设定好要关系你的服务后，把上线和下线逻辑函数交给 watch 进行回调，也就是当有服务上下线后（某个服务的单例）判断是否为当前用户关心来完成对应增加删除对应服务信道而已。
• 当用户进行选择服务的时候，采取 RR 轮转选取对应服务的某个信道。

实现代码即对应注释（channel.hpp）：

#pragma once
#include <brpc/channel.h>
#include <string>
#include <vector>
#include <unordered_map>
#include <mutex>
#include "log.hpp"

// 使用指针管理对象都：：
// etcd 通知某个服务上线或者下线是通过 host：ip+port 通知的，然后这边收到就要进行对应的管理，故进行 host 与对应信道映射。
// 管理一个服务间所有关于它的实例信道（也就是站在某个服务的所有单例对象的信道管理者角度）
class ServiceChannels {
public:
    using Ptr = std::shared_ptr<ServiceChannels>;
    using channelptr = std::shared_ptr<brpc::Channel>;

    ServiceChannels(const std::string &name) : _service_name(name), _idx(0) {}

    void Append(const std::string &host) {
        std::unique_lock<std::mutex> lock(_mtx);
        auto it = _hosts.find(host);
        if (it == _hosts.end()) {
            // 构建 host 对应的信道进行初始化
            std::shared_ptr<brpc::Channel> pchannel = 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";
            auto ok = pchannel->Init(host.c_str(), &options);
            if (ok == -1) {
                LOG_ERROR("初始化{}-{}信道失败!", _service_name, host);
                return;
            }
            // 完成信道这边的管理
            _hosts[host] = pchannel;
            _channels.push_back(pchannel);
        }
    }

    void Remove(const std::string &host) {
        std::unique_lock<std::mutex> lock(_mtx);
        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(host);
    }

    channelptr Choose() {
        std::unique_lock<std::mutex> lock(_mtx);
        if (!_channels.size()) return channelptr();
        // 存在进行轮询输出
        int32_t index = _idx++ % _channels.size();
        return _channels[index];
    }

private:
    std::mutex _mtx;
    std::string _service_name;
    std::unordered_map<std::string, channelptr> _hosts; // 主机名（ip+port）与 brpc 信道之间映射关系
    std::vector<channelptr> _channels;                  // 维护所有的 channel
    int32_t _idx;                                       // 用于 RR 轮转策略
};

// 所有服务的管理者，负责接收到 etcd 服务通知进行对应注册删除服务（或者为服务注册单例对象也就是信道等）等
class ServiceManager {
public:
    using Ptr = std::shared_ptr<ServiceManager>;

    ServiceManager() {}

    // 先声明，我关注哪些服务的上下线，不关心的就不需要管理了
    void Cared(const std::string &service_name) {
        std::unique_lock<std::mutex> lock(_mtx);
        _care_services.insert(service_name);
    }

    // etcd 监视的回调函数
    void OnlineService(const std::string &service_instance_name, const std::string &host) {
        const std::string &service_name = GetService(service_instance_name);
        auto service = ServiceChannels::Ptr(); // 空的服务信道集
        {
            std::unique_lock<std::mutex> lock(_mtx);
            auto fit = _care_services.find(service_name);
            if (fit == _care_services.end()) {
                LOG_DEBUG("{}-{} 服务上线了，但是当前并不关心！", service_name, host);
                return;
            }
            // 说明关心对应服务，如果服务存在就添加对应信道即可，不存在就创建服务再添加对应信道
            auto sit = _services.find(service_name);
            if (sit == _services.end()) {
                service = std::make_shared<ServiceChannels>(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 UnonlineService(const std::string &service_instance_name, const std::string &host) {
        const std::string &service_name = GetService(service_instance_name);
        auto service = ServiceChannels::Ptr(); // 空的服务信道集
        {
            std::unique_lock<std::mutex> lock(_mtx);
            auto fit = _care_services.find(service_name);
            if (fit == _care_services.end()) {
                LOG_DEBUG("{}-{} 服务下线了，但是当前并不关心！", service_name, host);
                return;
            }
            // 从服务集合中查找进行删除
            auto sit = _services.find(service_name);
            if (sit == _services.end()) {
                LOG_WARN("删除{}服务节点时，没有找到管理对象", service_name);
                return;
            } else {
                service = sit->second;
            }
        }
        service->Remove(host);
        LOG_DEBUG("{}-{} 服务下线节点，进行删除管理！", service_name, host);
    }

    ServiceChannels::channelptr ChooseService(const std::string &servicename) {
        std::unique_lock<std::mutex> lock(_mtx);
        auto sit = _services.find(servicename);
        if (sit == _services.end()) {
            LOG_ERROR("当前没有能够提供 {} 服务的节点！", servicename);
            return ServiceChannels::channelptr();
        }
        return sit->second->Choose();
    }

private:
    const std::string GetService(const std::string &service_instance_name) {
        auto pos = service_instance_name.find_last_of('/');
        if (pos == std::string::npos) return service_instance_name;
        // 一般来说不会找不到，注册进去的都是包含实例
        return service_instance_name.substr(0, pos);
    }

    std::mutex _mtx;
    std::unordered_map<std::string, ServiceChannels::Ptr> _services; // 添加进来的服务都是与业务有关的，也就是关心的
    std::unordered_set<std::string> _care_services;
};

五、基于 Etcd 实现服务上下线监控来完成 BRPC 服务调用

上面封装好了对应的 channel.hpp，下面就使用它结合之前封装的 etcd.hpp 以及复用下 brpc 的简单 echo 服务组合起来使用测试下（说白了其实就是 brpc 的 echo 服务的 server 与 client 端套用下然后结合封装 etcd 代码以及改变下获取 channel 方式而已）。

说下大致思路：

• 实现 register 与 discovery 两个程序 也就是对应的添加 rpc 服务启动 rpc-server+etcd 注册 echo 服务与借助 servicesmanager 获取对应服务信道 + 进行 rpc 服务请求发送。

逻辑过程图：

测试效果

• 一开始启动对应的 rpc 客户端，发现 etcd 中没有关心的服务即对应实例，只能看到之前在 etcd 服务器中注册的其他服务，但是是不相关的故不进行调用。
• 此时启动对应 rpc-server 也就是给 rpc 服务器添加对应服务并部署，给 etcd 注册进去对应服务后开始运行等待。
• 此时服务端收到对应的 rpc 客户端发来的请求，然后进行调用 echo 服务进行构建答复发送回去。
• 客户端收到服务端发送来的答复进行解析处理。

六、本篇小结

RPC 通过隐藏网络细节实现跨进程调用，BRPC 凭借高性能与 C++ 友好性成为优选；本文从理论到实践，结合 ETCD 服务治理，为分布式系统通信提供了可落地的参考路径。