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

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 客户端：通过约定好的 Protobuf 协议生成的调用相关 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;
}
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

3. rpc-client：

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

void clientcallback(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;
}

int main() {
    brpc::ChannelOptions options;
    options.connect_timeout_ms = -1;
    options.timeout_ms = -1;
    options.max_retry = 3;
    options.protocol = "baidu_std";
    
    brpc::Channel channel;
    auto ret = channel.Init("127.0.0.1:8080", &options);
    if (ret == -1) {
        std::cout << "初始化信道失败！\n";
        return -1;
    }
    
    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);
    stub.Echo(cntl, &req, rsp, closure);
    
    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 rpc_guard(done);
        std::cout << "收到消息：" << request->message() << std::endl;
        std::string str = request->message() + "--这是响应！！";
        response->set_message(str);
    }
};

int main() {
    logging::LoggingSettings logset;
    logset.logging_dest = logging::LoggingDestination::LOG_TO_NONE;
    logging::InitLogging(logset);
    
    brpc::Server server;
    EchoService echo;
    server.AddService(&echo, brpc::ServiceOwnership::SERVER_DOESNT_OWN_SERVICE);
    
    brpc::ServerOptions options;
    options.idle_timeout_sec = -1;
    options.num_threads = 1;
    
    auto ret = server.Start(8080, &options);
    if (ret == -1) {
        std::cout << "启动服务器失败！\n";
        return -1;
    }
    
    server.RunUntilAskedToQuit();
    return 0;
}

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

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

封装思想：

1. ServiceChannels：管理一个服务间所有关于它的实例信道。需要对应信道的增删查等功能，完成对应主机名与服务信道的映射，以及采取 RR 轮转方式使用对应服务的信道。
2. ServicesManager：把 ServiceChannels 管理起来，暴露给外面使用的，并集合对应 etcd 的 watch 监控来用。当有服务上下线后判断是否为当前用户关心来完成对应增加删除对应服务信道。

实现代码（channel.hpp）：

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

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()) {
            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;
    std::vector<channelptr> _channels;
    int32_t _idx;
};

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

    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 服务组合起来使用测试下。

大致思路：

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

测试效果

• 一开始启动对应的 rpc 客户端，发现 etcd 中没有关心的服务即对应实例，只能看到之前在 etcd 服务器中注册的其他服务，但是是不相关的故不进行调用。

• 此时启动对应 rpc-server 也就是给 rpc 服务器添加对应服务并部署，给 etcd 注册进去对应服务后开始运行等待。

• 此时服务端收到对应的 rpc 客户端发来的请求，然后进行调用 echo 服务进行构建答复发送回去。

• 客户端收到服务端发送来的答复进行解析处理。

代码汇总

访问代码仓库