直接如题,从brpc开源的文档中看到brpc既支持同步调用,同时也支持异步调用。这里直接给出同步、异步的例子,同时对其进行分析。
1、brpc同步调用
brcp的同步调用是之前的echo的简单例子,所谓同步就是client对远端的server进行调用,同时自己原地等待,等待rpc返回之后,在进行之后的操作。
代码如下:
#include <iostream>
#include <gflags/gflags.h>
#include <butil/logging.h>
#include <butil/time.h>
#include <brpc/channel.h>
#include "echo.pb.h"
using namespace std;
DEFINE_string(attachment, "", "Carry this along with requests");
DEFINE_string(protocol, "baidu_std", "Protocol type. Defined in src/brpc/options.proto");
DEFINE_string(connection_type, "", "Connection type. Available values: single, pooled, short");
DEFINE_string(server, "0.0.0.0:8000", "IP Address of server");
DEFINE_string(load_balancer, "", "The algorithm for load balancing");
DEFINE_int32(timeout_ms, 100, "RPC timeout in milliseconds");
DEFINE_int32(max_retry, 3, "Max retries(not including the first RPC)");
DEFINE_int32(interval_ms, 1000, "Milliseconds between consecutive requests");
int main(int argc, char* argv[]) {
// Parse gflags. We recommend you to use gflags as well.
GFLAGS_NS::ParseCommandLineFlags(&argc, &argv, true);
// A Channel represents a communication line to a Server. Notice that
// Channel is thread-safe and can be shared by all threads in your program.
brpc::Channel channel;
// Initialize the channel, NULL means using default options.
brpc::ChannelOptions options;
options.protocol = FLAGS_protocol;
options.connection_type = FLAGS_connection_type;
options.timeout_ms = FLAGS_timeout_ms/*milliseconds*/;
options.max_retry = FLAGS_max_retry;
if (channel.Init(FLAGS_server.c_str(), FLAGS_load_balancer.c_str(), &options) != 0) {
LOG(ERROR) << "Fail to initialize channel";
return -1;
}
// Normally, you should not call a Channel directly, but instead construct
// a stub Service wrapping it. stub can be shared by all threads as well.
example::EchoService_Stub stub(&channel);
// Send a request and wait for the response every 1 second.
int log_id = 0;
while (!brpc::IsAskedToQuit()) {
// We will receive response synchronously, safe to put variables
// on stack.
example::EchoRequest request;
example::EchoResponse response;
brpc::Controller cntl;
request.set_message("hello world");
cntl.set_log_id(log_id ++); // set by user
// Set attachment which is wired to network directly instead of
// being serialized into protobuf messages.
cntl.request_attachment().append(FLAGS_attachment);
// Because `done'(last parameter) is NULL, this function waits until
// the response comes back or error occurs(including timedout).
//注意这里是同步调用,因为回调函数done被设置为NULL
//同步表示stub调用rpc,返回之后才会进行后面的语句
stub.Echo(&cntl, &request, &response, NULL);
cout << "after stub_method 0 " << endl;
if (!cntl.Failed()) {
LOG(INFO) << "Received response from " << cntl.remote_side()
<< " to " << cntl.local_side()
<< ": " << response.message() << " (attached="
<< cntl.response_attachment() << ")"
<< " latency=" << cntl.latency_us() << "us";
} else {
LOG(WARNING) << cntl.ErrorText();
}
cout << "after stub_method 1 " << endl;
usleep(FLAGS_interval_ms * 1000L);
}
LOG(INFO) << "EchoClient is going to quit";
return 0;
}
注意
1、stub.Echo调用可以看到回调函数被设置为NULL,那么此时调用被设置为同步操作,原地等待。
2、同步访问中的response/controller不会在CallMethod后被框架使用,它们都可以分配在栈上。注意,如果request/response字段特别多字节数特别大的话,还是更适合分配在堆上。
运行结果
同步进行,按照操作流程一步步打印
2、brpc异步调用
brpc的异步调用指的是和同步调用相比,在进行rpc调用之后,此时callmethod就结束了,进行继续执行后续的动作,等到rpc返回之后,会调用事先注册的回调函数,回调函数进行后面rpc返回之后的操作。
代码如下:
#include <iostream>
#include <gflags/gflags.h>
#include <butil/logging.h>
#include <butil/time.h>
#include <brpc/channel.h>
#include "echo.pb.h"
using namespace std;
DEFINE_bool(send_attachment, true, "Carry attachment along with requests");
DEFINE_string(protocol, "baidu_std", "Protocol type. Defined in src/brpc/options.proto");
DEFINE_string(connection_type, "", "Connection type. Available values: single, pooled, short");
DEFINE_string(server, "0.0.0.0:8003", "IP Address of server");
DEFINE_string(load_balancer, "", "The algorithm for load balancing");
DEFINE_int32(timeout_ms, 100, "RPC timeout in milliseconds");
DEFINE_int32(max_retry, 3, "Max retries(not including the first RPC)");
//定义回调函数
void HandleEchoResponse(
brpc::Controller* cntl,
example::EchoResponse* response) {
// std::unique_ptr makes sure cntl/response will be deleted before returning.
std::unique_ptr<brpc::Controller> cntl_guard(cntl);
std::unique_ptr<example::EchoResponse> response_guard(response);
if (cntl->Failed()) {
LOG(WARNING) << "Fail to send EchoRequest, " << cntl->ErrorText();
return;
}
LOG(INFO) << "Received response from " << cntl->remote_side()
<< ": " << response->message() << " (attached="
<< cntl->response_attachment() << ")"
<< " latency=" << cntl->latency_us() << "us";
}
int main(int argc, char* argv[]) {
// Parse gflags. We recommend you to use gflags as well.
GFLAGS_NS::ParseCommandLineFlags(&argc, &argv, true);
// A Channel represents a communication line to a Server. Notice that
// Channel is thread-safe and can be shared by all threads in your program.
brpc::Channel channel;
// Initialize the channel, NULL means using default options.
brpc::ChannelOptions options;
options.protocol = FLAGS_protocol;
options.connection_type = FLAGS_connection_type;
options.timeout_ms = FLAGS_timeout_ms/*milliseconds*/;
options.max_retry = FLAGS_max_retry;
if (channel.Init(FLAGS_server.c_str(), FLAGS_load_balancer.c_str(), &options) != 0) {
LOG(ERROR) << "Fail to initialize channel";
return -1;
}
// Normally, you should not call a Channel directly, but instead construct
// a stub Service wrapping it. stub can be shared by all threads as well.
example::EchoService_Stub stub(&channel);
// Send a request and wait for the response every 1 second.
int log_id = 0;
while (!brpc::IsAskedToQuit()) {
// Since we are sending asynchronous RPC (`done' is not NULL),
// these objects MUST remain valid until `done' is called.
// As a result, we allocate these objects on heap
//注意一般在堆上创建,为了防止done中调用时候被弹出栈,但是访问无效块
example::EchoResponse* response = new example::EchoResponse();
brpc::Controller* cntl = new brpc::Controller();
// Notice that you don't have to new request, which can be modified
// or destroyed just after stub.Echo is called.
example::EchoRequest request;
request.set_message("hello world");
cntl->set_log_id(log_id ++); // set by user
if (FLAGS_send_attachment) {
// Set attachment which is wired to network directly instead of
// being serialized into protobuf messages.
cntl->request_attachment().append("foo");
}
// We use protobuf utility `NewCallback' to create a closure object
// that will call our callback `HandleEchoResponse'. This closure
// will automatically delete itself after being called once
//注册回调函数,rpc返回之后的操作都在done的回调函数中进行
google::protobuf::Closure* done = brpc::NewCallback(
&HandleEchoResponse, cntl, response);
stub.Echo(cntl, &request, response, done);
cout << "after stub_method " << endl;
// This is an asynchronous RPC, so we can only fetch the result
// inside the callback
sleep(1);
}
LOG(INFO) << "EchoClient is going to quit";
return 0;
}
注意
1、由于CallMethod结束不意味着RPC结束,response/controller仍可能被框架及done->Run()使用,它们一般得创建在堆上,并在done->Run()中删除。如果提前删除了它们,那当done->Run()被调用时,将访问到无效内存。
2、brpc提供两种异步访问的定义,可以独立地创建这些对象,并使用NewCallback生成done,也可以把Response和Controller作为done的成员变量,一起new出来,一般使用前一种方法。
独立创建:
MyResponse* response = new MyResponse;
brpc::Controller* cntl = new brpc::Controller;
MyService_Stub stub(&channel);
MyRequest request; // 你不用new request,即使在异步访问中.
request.set_foo(...);
cntl->set_timeout_ms(...);
stub.some_method(cntl, &request, response, google::protobuf::NewCallback(OnRPCDone, response, cntl));
一起调用:
class OnRPCDone: public google::protobuf::Closure {
public:
void Run() {
// unique_ptr会帮助我们在return时自动delete this,防止忘记。gcc 3.4下的unique_ptr是模拟版本。
std::unique_ptr<OnRPCDone> self_guard(this);
if (cntl->Failed()) {
// RPC失败了. response里的值是未定义的,勿用。
} else {
// RPC成功了,response里有我们想要的数据。开始RPC的后续处理。
}
}
MyResponse response;
brpc::Controller cntl;
}
OnRPCDone* done = new OnRPCDone;
MyService_Stub stub(&channel);
MyRequest request; // 你不用new request,即使在异步访问中.
request.set_foo(...);
done->cntl.set_timeout_ms(...);
stub.some_method(&done->cntl, &request, &done->response, done);
运行结果
先去打印结果,等rpc调用返回时候,去异步调用done函数
3、半同步
前面已经提到了异步操作,那么当然异步操作可以支持多并发的操作,那么在异步操作的背景下,可以完成半同步操作吗?使用的是等待rpc完成操作。
所谓的等待rpc完成操作实际是提供一种等待机制,在任意时刻可以出发等待操作,即使是在异步调用的过程中,可以等待异步的操作完成,然后进行后续的操作,也有人称之为半同步操作。
代码如下:
#include <iostream>
#include <gflags/gflags.h>
#include <butil/logging.h>
#include <butil/time.h>
#include <brpc/channel.h>
#include "echo.pb.h"
using namespace std;
DEFINE_bool(send_attachment, true, "Carry attachment along with requests");
DEFINE_string(protocol, "baidu_std", "Protocol type. Defined in src/brpc/options.proto");
DEFINE_string(connection_type, "", "Connection type. Available values: single, pooled, short");
DEFINE_string(server, "0.0.0.0:8003", "IP Address of server");
DEFINE_string(load_balancer, "", "The algorithm for load balancing");
DEFINE_int32(timeout_ms, 100, "RPC timeout in milliseconds");
DEFINE_int32(max_retry, 3, "Max retries(not including the first RPC)");
void HandleEchoResponse(
brpc::Controller* cntl,
example::EchoResponse* response) {
// std::unique_ptr makes sure cntl/response will be deleted before returning.
std::unique_ptr<brpc::Controller> cntl_guard(cntl);
std::unique_ptr<example::EchoResponse> response_guard(response);
if (cntl->Failed()) {
LOG(WARNING) << "Fail to send EchoRequest, " << cntl->ErrorText();
return;
}
LOG(INFO) << "Received response from " << cntl->remote_side()
<< ": " << response->message() << " (attached="
<< cntl->response_attachment() << ")"
<< " latency=" << cntl->latency_us() << "us";
}
int main(int argc, char* argv[]) {
// Parse gflags. We recommend you to use gflags as well.
GFLAGS_NS::ParseCommandLineFlags(&argc, &argv, true);
// A Channel represents a communication line to a Server. Notice that
// Channel is thread-safe and can be shared by all threads in your program.
brpc::Channel channel;
// Initialize the channel, NULL means using default options.
brpc::ChannelOptions options;
options.protocol = FLAGS_protocol;
options.connection_type = FLAGS_connection_type;
options.timeout_ms = FLAGS_timeout_ms/*milliseconds*/;
options.max_retry = FLAGS_max_retry;
if (channel.Init(FLAGS_server.c_str(), FLAGS_load_balancer.c_str(), &options) != 0) {
LOG(ERROR) << "Fail to initialize channel";
return -1;
}
// Normally, you should not call a Channel directly, but instead construct
// a stub Service wrapping it. stub can be shared by all threads as well.
example::EchoService_Stub stub(&channel);
// Send a request and wait for the response every 1 second.
int log_id = 0;
while (!brpc::IsAskedToQuit()) {
// Since we are sending asynchronous RPC (`done' is not NULL),
// these objects MUST remain valid until `done' is called.
// As a result, we allocate these objects on heap
example::EchoResponse* response_1 = new example::EchoResponse();
brpc::Controller* cntl_1 = new brpc::Controller();
example::EchoRequest request1;
request1.set_message("hello world from request1");
example::EchoResponse* response_2 = new example::EchoResponse();
brpc::Controller* cntl_2 = new brpc::Controller();
example::EchoRequest request2;
request2.set_message("hello world from request2");
cntl_1->set_log_id(log_id ++); // set by user
cntl_2->set_log_id(log_id ++);
//获取rpc调用id号
const brpc::CallId id1 = cntl_1->call_id();
const brpc::CallId id2 = cntl_2->call_id();
if (FLAGS_send_attachment) {
// Set attachment which is wired to network directly instead of
// being serialized into protobuf messages.
cntl_1->request_attachment().append("foo_1");
cntl_2->request_attachment().append("foo_2");
}
// We use protobuf utility `NewCallback' to create a closure object
// that will call our callback `HandleEchoResponse'. This closure
// will automatically delete itself after being called once
google::protobuf::Closure* done_1 = brpc::NewCallback(
&HandleEchoResponse, cntl_1, response_1);
google::protobuf::Closure* done_2 = brpc::NewCallback(
&HandleEchoResponse, cntl_2, response_2);
stub.Echo(cntl_1, &request1, response_1, done_1);
stub.Echo(cntl_2, &request2, response_2, done_2);
//等待两个rpc调用的完成
brpc::Join(id1);
brpc::Join(id2);
//等待完成之后,才能进行后续的操作
cout << "stub method " << endl;
// This is an asynchronous RPC, so we can only fetch the result
// inside the callback
sleep(1);
}
LOG(INFO) << "EchoClient is going to quit";
return 0;
}
注意:
1、join函数操作实际是在发起RPC前调用Controller.call_id()获得一个id,发起RPC调用后Join那个id。Join()的行为是等到RPC结束且done->Run()运行后,一些Join的性质如下:
如果对应的RPC已经结束,Join将立刻返回。
多个线程可以Join同一个id,它们都会醒来。
同步RPC也可以在另一个线程中被Join,但一般不会这么做
2、在RPC调用后Join(controller->call_id())是错误的行为,一定要先把call_id保存下来。因为RPC调用后controller可能被随时开始运行的done删除。
3、半同步操作的常见框架:
brpc::Controller cntl1;
brpc::Controller cntl2;
MyResponse response1;
MyResponse response2;
...
stub1.method1(&cntl1, &request1, &response1, brpc::DoNothing());
stub2.method2(&cntl2, &request2, &response2, brpc::DoNothing());
...
brpc::Join(cntl1.call_id());
brpc::Join(cntl2.call_id());
运行结果
等待rpc异步调用完成之后在打印信息:
4、取消rpc调用
取消操作理解较为简单,即在另外一个线层里面取消其他线程的rpc调用,直接看代码,这一段代码完成线程1取消线程2的rpc调用,线程2取消线程1的调用。
具体逻辑操作如下:
// Time RPC1 RPC2
// 1 response1 comes back.
// 2 running done1.
// 3 cancel RPC2
// 4 running done2 (NOTE: done also runs)
// 5 cancel RPC1 (no effect)
代码如下:
#include <gflags/gflags.h>
#include <butil/logging.h>
#include <butil/time.h>
#include <brpc/channel.h>
#include "echo.pb.h"
DEFINE_string(protocol, "baidu_std", "Protocol type. Defined in src/brpc/options.proto");
DEFINE_string(connection_type, "", "Connection type. Available values: single, pooled, short");
DEFINE_string(server, "0.0.0.0:8000", "IP Address of server");
DEFINE_string(load_balancer, "", "The algorithm for load balancing");
DEFINE_int32(timeout_ms, 100, "RPC timeout in milliseconds");
DEFINE_int32(max_retry, 3, "Max retries(not including the first RPC)");
// A special done for canceling another RPC.
class CancelRPC : public google::protobuf::Closure {
public:
explicit CancelRPC(brpc::CallId rpc_id) : _rpc_id(rpc_id) {}
//取消rpc操作
void Run() {
brpc::StartCancel(_rpc_id);
}
private:
brpc::CallId _rpc_id;
};
int main(int argc, char* argv[]) {
// Parse gflags. We recommend you to use gflags as well.
GFLAGS_NS::ParseCommandLineFlags(&argc, &argv, true);
// A Channel represents a communication line to a Server. Notice that
// Channel is thread-safe and can be shared by all threads in your program.
brpc::Channel channel;
// Initialize the channel, NULL means using default options.
brpc::ChannelOptions options;
options.protocol = FLAGS_protocol;
options.connection_type = FLAGS_connection_type;
options.timeout_ms = FLAGS_timeout_ms/*milliseconds*/;
options.max_retry = FLAGS_max_retry;
if (channel.Init(FLAGS_server.c_str(), FLAGS_load_balancer.c_str(), &options) != 0) {
LOG(ERROR) << "Fail to initialize channel";
return -1;
}
// Normally, you should not call a Channel directly, but instead construct
// a stub Service wrapping it. stub can be shared by all threads as well.
example::EchoService_Stub stub(&channel);
// Send a request and wait for the response every 1 second.
int log_id = 0;
while (!brpc::IsAskedToQuit()) {
example::EchoRequest request1;
example::EchoResponse response1;
brpc::Controller cntl1;
example::EchoRequest request2;
example::EchoResponse response2;
brpc::Controller cntl2;
request1.set_message("hello1");
request2.set_message("hello2");
cntl1.set_log_id(log_id ++); // set by user
cntl2.set_log_id(log_id ++);
//获取rpc调用的id号
const brpc::CallId id1 = cntl1.call_id();
const brpc::CallId id2 = cntl2.call_id();
//定义回调函数
CancelRPC done1(id2);
CancelRPC done2(id1);
butil::Timer tm;
tm.start();
//注册回调函数,注意这里回调函数会完成取消操作,
stub.Echo(&cntl1, &request1, &response1, &done1);
stub.Echo(&cntl2, &request2, &response2, &done2);
//等待回调函数完成
brpc::Join(id1);
brpc::Join(id2);
tm.stop();
if (cntl1.Failed() && cntl2.Failed()) {
LOG(WARNING) << "Both failed. rpc1:" << cntl1.ErrorText()
<< ", rpc2: " << cntl2.ErrorText();
} else if (!cntl1.Failed()) {
LOG(INFO) << "Received `" << response1.message() << "' from rpc1="
<< id1.value << '@' << cntl1.remote_side()
<< " latency=" << tm.u_elapsed() << "us"
<< " rpc1_latency=" << cntl1.latency_us() << "us"
<< " rpc2_latency=" << cntl2.latency_us() << "us";
} else {
LOG(INFO) << "Received `" << response2.message() << "' from rpc2="
<< id2.value << '@' << cntl2.remote_side()
<< " latency=" << tm.u_elapsed() << "us"
<< " rpc1_latency=" << cntl1.latency_us() << "us"
<< " rpc2_latency=" << cntl2.latency_us() << "us";
}
sleep(1);
}
LOG(INFO) << "EchoClient is going to quit";
return 0;
}
注意:
1、StartCancel调用完成后RPC并未立刻结束,你不应该碰触Controller的任何字段或删除任何资源,它们自然会在RPC结束时被done中对应逻辑处理。如果你一定要在原地等到RPC结束(一般不需要),则可通过Join(call_id)。
2、关于StartCancel的一些事实:
call_id在发起RPC前就可以被取消,RPC会直接结束(done仍会被调用)。
call_id可以在另一个线程中被取消。
取消一个已经取消的call_id不会有任何效果。推论:同一个call_id可以被多个线程同时取消,但最多一次有效果。
这里的取消是纯client端的功能,server端未必会取消对应的操作,server cancelation是另一个功能。
运行结果
线程1取消线程2结果:
线程2取消线程1结果:
参考:
https://github.com/brpc/brpc/blob/master/docs/cn/client.md#异步访问