brpc

1. brpc 框架介绍

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）。

简单来说：

  1. brpc是一个rpc框架，远程调用 a. 例子：一个add函数，以前是将数据的处理在本地封装实现，进行调用完成 b. rpc框架将数据的处理过程交给服务器完成

2.类与接口介绍

1.包含头文件

#include <butil/logging.h>

2.1 日志输出类与接口

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

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

3.使用样例

先创建一个protobuff
main.proto

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

定义了一个Echo服务，有请求和响应消息，字段都是一个message，定义了rpc服务，表示客户端调用Echo，传入一个EchoRequest，返回一个EchoResponse

同步调用

同步调用是指客户端会阻塞收到server端的响应或发生错误。
下面我们以Echo（输出hello world）方法为例, 来讲解基础的同步RPC请求是如何实现的。
server.cc

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

// 1. 继承于EchoService创建一个子类，并实现rpc调用的业务功能
class EchoServiceImpl : public example::EchoService
{
public:
    EchoServiceImpl() {}
    ~EchoServiceImpl() {}
    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);
        // done->Run();
    }
};
int main(int argc, char *argv[])
{
    // 关闭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 = 1;       // io线程数量
    ret = server.Start(8080, &options);
    if (ret == -1)
    {
        std::cout << "启动服务器失败！\n";
        return -1;
    }
    server.RunUntilAskedToQuit(); // 修改等待运行结束
    return 0;
}

client.cc

#include <brpc/channel.h>
#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() == true) {
        std::cout << "Rpc调用失败：" << cntl->ErrorText() << std::endl;
        return;
    }
    std::cout << "收到响应: " << response->message() << std::endl;
}

int main(int argc, char *argv[])
{
    //1. 构造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_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("你好~康高锦~！");

    brpc::Controller *cntl = new brpc::Controller();
    example::EchoResponse *rsp = new example::EchoResponse();
    // 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;
    auto clusure = google::protobuf::NewCallback(callback, cntl, rsp);
    stub.Echo(cntl, &req, rsp, clusure); //异步调用
    std::cout << "异步调用结束！\n";
    std::this_thread::sleep_for(std::chrono::seconds(3));
    return 0;
}

Makefile:这里主要就是把brpc所需的库链接上即可

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

4.封装channel

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

brpc本质上是进行rpc调用的，但是向谁调用什么服务需要被管理起来，搭配etcd实现的注册中心管理
原因：通过注册中心，能够获知谁能提供什么服务，进而能够连接它发起的服务调用
封装思想

  1. 主要是管理起来网络通信的信道，将不同服务节点主机的通信信道管理起来
  2. 封装的服务节点信道的管理，而不是rpc调用的管理
    封装：
  3. 指定服务的信道管理类
    a. 一个服务可能会有多个节点提供服务，每个节点都有对应的channel，建立服务与信道的映射关系，并且关系是一对多的，采用RR轮转策略进行获取
  4. 总体的服务信道管理类
    a. 将多个服务的信道管理对象管理起来
#pragma once
#include <brpc/channel.h>
#include <string>
#include <vector>
#include <unordered_map>
#include <mutex>
#include "logger.hpp"

namespace yjt_im
{
    // 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) {}
        // 服务上线了一个节点，则调用append新增信道
        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);
        }
        // 服务下线了一个节点，则调用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 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;
    };
}
# c++ # 开发语言
Logo
AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

告别扫描文档处理烦恼：ScanTailor Advanced的智能文档优化解决方案

ScanTailor Advanced是一款专业的扫描文档后期处理工具，专为解决老旧文档数字化中的各种问题而设计。通过智能内容识别、批量处理和多线程优化，它能将倾斜、模糊、杂乱的扫描文档转化为整洁的电子文件，显著提升文档数字化效率。## 文档数字化的常见痛点与解决方案### 页面歪斜校正：智能算法自动修正传统扫描文档经常出现页面歪斜问题，手动调整每页耗时耗力。ScanTailor Adv

avatar AI Agent技术社区

SeedVR2终极指南：如何用单步扩散模型实现专业级视频修复

还在为老旧视频的模糊画质而烦恼吗？面对抖动严重的监控录像或褪色的家庭影像，你是否曾希望有一种简单高效的修复方案？SeedVR2正是为解决这些痛点而生的一站式视频修复工具。作为字节跳动推出的先进视频修复模型，它能够在单步推理中实现专业级的视频质量提升，彻底改变了传统视频修复的工作流程。## 核心价值：为什么选择SeedVR2进行视频修复？**传统视频修复的三大瓶颈**：1. **计算成本高

avatar AI Agent技术社区

终极指南：如何用Python自动化工具告别大麦抢票焦虑，轻松获取心仪演出门票

还记得那些让你心跳加速的抢票时刻吗？热门演唱会开票瞬间，你手忙脚乱地刷新页面，却眼睁睁看着"立即购买"按钮变灰，票在几秒内售罄。这种"拼手速"的抢票时代已经过去，现在有了基于Python开发的大麦自动抢票工具，让你告别手速焦虑，轻松获取周杰伦、五月天、周深等热门演出的门票。## 🔥 你的抢票痛点，我们都有解决方案### 场景一：热门演唱会一票难求"周杰伦演唱会开票3秒售罄，我连页面都没

avatar AI Agent技术社区