🌈 个人主页：Zfox_
🔥 系列专栏：C++框架/库

🔥 介绍

brpc 是用 c++ 语言编写的工业级 RPC 框架，常用于搜索、存储、机器学习、广告、推荐等高性能系统。

你可以使用它

• 搭建能在一个端口支持多协议的服务, 或访问各种服务
  • restful http/https, h2/gRPC。使用 brpc 的 http 实现比 libcurl 方便多了。从其他语言通过 HTTP/h2+json 访问基于 protobuf 的协议.
  • redis 和 memcached, 线程安全，比官方 client 更方便。
  • rtmp/flv/hls, 可用于搭建流媒体服务.
  • 支持 thrift , 线程安全，比官方 client 更方便
  • 各种百度内使用的协议: baidu_std, streaming_rpc, hulu_pbrpc, sofa_pbrpc, nova_pbrpc, public_pbrpc, ubrpc 和使用 nshead 的各种协议.
  • 基于工业级的 RAFT 算法实现搭建高可用分布式系统，已在 braft 开源。
• Server 能同步或异步处理请求。
• Client 支持同步、 异步、 半同步，或使用组合 channels 简化复杂的分库或并发访问。
• 通过 http 界面调试服务, 使用 cpu, heap, contention profilers.
• 获得更好的延时和吞吐.
• 把你组织中使用的协议快速地加入 brpc，或定制各类组件, 包括命名服务 (dns, zk, etcd), 负载均衡 (rr, random, consistent hashing)

🔥 安装

先安装依赖

$ 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/
build && cd build
cmake - DCMAKE_INSTALL_PREFIX=/usr .. && cmake --build . -j6
make && sudo make install

🔥 类与接口介绍

🦋 日志输出类与接口

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

日志输出这里，本质上我们其实用不着 brpc 的日志输出，因此在这里主要介绍如何关闭日志输出。

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

🦋 创建 proto 文件

main.proto

syntax="proto3";
package example;

option cc_generic_services = true;

// 定义 Echo 方法请求参数结构
message EchoRequest {
  string message = 1;
};

// 定义 Echo 方法响应参数结构
message EchoResponse {
  string message = 1;
};

// 定义 RPC 远端方法
service EchoService {
  rpc Echo(EchoRequest) returns (EchoResponse);
};

🔥 使用

🦋 同步调用

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

下面我们以 Echo（输出 hello world）方法为例, 来讲解基础的同步 RPC 请求是如何实现的

创建服务端源码
brpc_server.cpp

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

// 1. 继承于 EchoService 创建一个子类，并实现 rpc 调用的业务功能
class EchoServiceImpl : public example::EchoService {
public:
    EchoServiceImpl() {}
    ~EchoServiceImpl() {}

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

        // rpc_guard 自动调用 run 接口  done->Run();
    }
}; 

int main()  
{ 
    // 关闭 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 = 4;            // io线程数
    ret = server.Start(8080, &options);
    if (ret == -1) {
        std::cout << "启动服务器失败! \n";
        return -1;
    }

    server.RunUntilAskedToQuit();

    return 0;
}

创建客户端源码

client.cpp

#include <brpc/channel.h>
#include <iostream>
#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()) {
        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;    // 连接超时等待时间
    options.timeout_ms = -1;            // rpc 请求等待超时时间，-1 表示一直等待
    options.max_retry = 3;              // 请求重试次数
    options.protocol = "baidu_std";     // 序列化协议，默认使用 baidu_std
    brpc::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("你好, server");

    // 上下文
    brpc::Controller *cntl = new brpc::Controller();
    example::EchoResponse *resp = new example::EchoResponse();

    // 异步
    auto clusure = google::protobuf::NewCallback(callback, cntl, resp);
    stub.Echo(cntl, &req, resp, clusure);
    std::cout << "异步调用结束! \n";
    std::this_thread::sleep_for(std::chrono::seconds(3));

    // stub.Echo(cntl, &req, resp, nullptr);
    // if (cntl->Failed()) {
    //     std::cout << "Rpc调用失败: " << cntl->ErrorText() << std::endl;
    //     return -1;
    // }

    // std::cout << "收到响应: " << resp->message() << std::endl;

    // delete cntl;
    // delete resp;

    return 0;
}

🔥 封装思想

rpc 调用这里的封装，因为不同的服务调用使用的是不同的 Stub，这个封装起来的意义不大，因此我们只需要封装通信所需的 Channel 管理即可，这样当需要进行什么样的服务调用的时候，只需要通过服务名称获取对应的 channel，然后实例化 Stub 进行调用即可。

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

channel.hpp

#pragma once

#include <brpc/channel.h>
#include <string>
#include <vector>
#include <mutex>
#include <unordered_map>
#include "logger.hpp"

// 1. 封装单个服务的信道管理类  一个服务节点一个信道
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;            // rpc 请求等待超时时间，-1 表示一直等待
        options.max_retry = 3;              // 请求重试次数
        options.protocol = "baidu_std";     // 序列化协议，默认使用 baidu_std
        int ret = channel->Init(host.c_str(), &options);
        if (ret == -1) {
            LOG_ERROR("初始化{}-{}信道失败!", _service_name);
            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);
                _hosts.erase(it);
                break;
            }
        }
    }

    // 通过 rr 轮转策略，获取一个 Channel 用于发起对应服务的 Rpc 调用
    ChannelPtr choose() {
        std::unique_lock<std::mutex> lock(_mutex);
        if (_channels.size() == 0) {
            LOG_ERROR("当前没有能够提供 {} 服务的节点!", _service_name);
            return ChannelPtr();
        }
        int 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>;
    using ChannelPtr = std::shared_ptr<brpc::Channel>;

    ServiceManager() {}

    // 获取指定服务的节点信道（发起 Rpc 请求）
    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 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;
};

🔥 共勉

😋 以上就是我对 【C++组件】Spdlog 日志组件的安装及使用 的理解, 觉得这篇博客对你有帮助的，可以点赞收藏关注支持一波~ 😉