当前位置：首页 > news >正文

brpc异步请求封装

news 2026/7/3 10:12:00

使用brpc发送proto格式的消息时，针对不同类型的请求，每个模块可能各自使用一套接口。但是对于该客户端上的各类请求，其链接管理，故障重发，这些是一致的。为了统一这些请求的处理，因此做了如何封装。https://gitee.com/lang2020/brpc_async

一、核心设计定位和价值

基于 BRPC 框架封装的异步 RPC 客户端通用框架，核心目标是屏蔽底层 BRPC 异步调用细节，为上层业务提供统一、简洁、可复用的异步 / 同步 RPC 调用能力，同时内置完善的可靠性保障机制，实现业务逻辑与 RPC 通信逻辑解耦。

简化使用：业务方只需关注业务请求和回调逻辑，无需编写复杂的 BRPC 异步代码。
统一规范：全系统 RPC 调用遵循统一范式，降低协作和维护成本。
高性能高可用：异步架构保证高并发，内置重试、超时保证服务可靠性。
易扩展：模板化 + 多态设计，快速支持新业务 RPC 接口和新特性扩展。

二、使用示例

针对一个具体的proto的rpc协议，只需要加两行声明具体的模板类。

/* package example; service EchoService { rpc Echo(EchoRequest) returns (EchoResponse); }; */ struct EchoTypes { using Client = BRPCAsyncClientTemplate<example::EchoRequest, example::EchoResponse, example::EchoService_Stub, &example::EchoService_Stub::Echo>; using ORequest = ObjectRequest<example::EchoRequest, example::EchoResponse, Client>; using Operation = ObjectOperation<example::EchoRequest, example::EchoResponse>; }; using EchoObjectRequest = EchoTypes::ORequest; using EchoObjectOperation = EchoTypes::Operation;

然后就可以直接定义自己的请求，实现具体的同步或者异步发送。每一个请求new一个ObjectRequest，然后发送出去，请求完成后会自动释放资源。

class TestOperation { public: explicit TestOperation() = default; void test_sync() { EchoObjectOperation op; op.request.set_message("test_value"); op.request.set_id(1); auto* sync_req = new EchoObjectRequest(op); sync_req->request(); ldout(nullptr, 0) << "test1: sync_req->request() end" << dendl; } void test_async_callback() { EchoObjectOperation op; op.request.set_message("test_value"); op.request.set_id(2); //支持ceph的Context回调，可以改成通用lambda Context* ctx = create_context_callback<TestOperation, &TestOperation::handle_echo_response>(this); auto* async_req = new EchoObjectRequest(op, ctx); async_req->async_request(); } void handle_echo_response(int result) { ldout(nullptr, 0) << "handle_echo_response: result=" << result << dendl; } void test_async_lambda() { EchoObjectOperation op; op.request.set_message("test_value"); op.request.set_id(3); Context* ctx = make_lambda_context([this, response = op.response](int result) { ldout(nullptr, 0) << " test3: response.use_count()=" << response.use_count() << dendl; handle_echo_response(result, response.get()); }); auto* async_req = new EchoObjectRequest(op, ctx); async_req->async_request(); } void handle_echo_response(int result, example::EchoResponse* response) { ldout(nullptr, 0) << "handle_echo_response: " << result << ", message: " << response->message() << dendl; } ~TestOperation() {} private: };