阅读量:2
既有适合小白学习的零基础资料，也有适合3年以上经验的小伙伴深入学习提升的进阶课程，涵盖了95%以上C C++开发知识点，真正体系化！
由于文件比较多，这里只是将部分目录截图出来，全套包含大厂面经、学习笔记、源码讲义、实战项目、大纲路线、讲解视频，并且后续会持续更新
server-thread-local与一次service回调绑定，从进service回调开始，到出service回调结束。所有的server-thread-local data会被尽量重用，在server停止前不会被删除。在实现上server-thread-local是一个特殊的bthread-local。
设置ServerOptions.thread_local_data_factory后访问Controller.thread_local_data()即可获得thread-local数据。若没有设置，Controller.thread_local_data()总是返回NULL。
若ServerOptions.reserved_thread_local_data大于0，Server会在启动前就创建这么多个数据。
与session-local的区别
session-local data得从server端的Controller获得， server-thread-local可以在任意函数中获得，只要这个函数直接或间接地运行在server线程中。
当service是同步时，session-local和server-thread-local基本没有差别，除了前者需要Controller创建。当service是异步时，且你需要在done->Run()中访问到数据，这时只能用session-local，因为server-thread-local在service回调外已经失效。
示例用法
struct MyThreadLocalData {
MyThreadLocalData() : y(0) {}
int y;
};
class EchoServiceImpl : public example::EchoService {
public:
…
void Echo(google::protobuf::RpcController* cntl_base,
const example::EchoRequest* request,
example::EchoResponse* response,
google::protobuf::Closure* done) {
…
brpc::Controller* cntl = static_castbrpc::Controller*(cntl_base);
// Get the thread-local data which is created by ServerOptions.thread_local_data_factory // and reused between different threads. // "tls" is short for "thread local storage". MyThreadLocalData* tls = static_cast<MyThreadLocalData*>(brpc::thread_local_data()); if (tls == NULL) { cntl->SetFailed("Require ServerOptions.thread_local_data_factory " "to be set with a correctly implemented instance"); return; } ... struct ServerOptions {
…
// The factory to create/destroy data attached to each searching thread
// in server.
// If this field is NULL, brpc::thread_local_data() is always NULL.
// NOT owned by Server and must be valid when Server is running.
// Default: NULL
const DataFactory* thread_local_data_factory;
// Prepare so many thread-local data before server starts, so that calls // to brpc::thread_local_data() get data directly rather than calling // thread_local_data_factory->Create() at first time. Useful when Create() // is slow, otherwise the RPC session may be blocked by the creation // of data and not served within timeout. // Default: 0 size_t reserved_thread_local_data; };
thread_local_data_factory的类型为DataFactory，你需要实现其中的CreateData和DestroyData。
注意：CreateData和DestroyData会被多个线程同时调用，必须线程安全。
class MyThreadLocalDataFactory : public brpc::DataFactory {
public:
void* CreateData() const {
return new MyThreadLocalData;
}
void DestroyData(void* d) const {
delete static_cast<MyThreadLocalData*>(d);
}
};
int main(int argc, char* argv[]) {
…
MyThreadLocalDataFactory thread_local_data_factory;
brpc::Server server; brpc::ServerOptions options; ... options.thread_local_data_factory = &thread_local_data_factory; ... bthread-local
Session-local和server-thread-local对大部分server已经够用。不过在一些情况下，我们需要更通用的thread-local方案。在这种情况下，你可以使用bthread_key_create, bthread_key_destroy, bthread_getspecific, bthread_setspecific等函数，它们的用法类似pthread中的函数。
这些函数同时支持bthread和pthread，当它们在bthread中被调用时，获得的是bthread私有变量; 当它们在pthread中被调用时，获得的是pthread私有变量。但注意，这里的“pthread私有变量”不是通过pthread_key_create创建的，使用pthread_key_create创建的pthread-local是无法被bthread_getspecific访问到的，这是两个独立的体系。由gcc的__thread，c++11的thread_local等声明的私有变量也无法被bthread_getspecific访问到。
由于brpc会为每个请求建立一个bthread，server中的bthread-local行为特殊：一个server创建的bthread在退出时并不删除bthread-local，而是还回server的一个pool中，以被其他bthread复用。这可以避免bthread-local随着bthread的创建和退出而不停地构造和析构。这对于用户是透明的。
主要接口
// Create a key value identifying a slot in a thread-specific data area.
// Each thread maintains a distinct thread-specific data area.
// destructor', if non-NULL, is called with the value associated to that key // when the key is destroyed. destructor’ is not called if the value
// associated is NULL when the key is destroyed.
// Returns 0 on success, error code otherwise.
extern int bthread_key_create(bthread_key_t* key, void (destructor)(void data)) __THROW;
// Delete a key previously returned by bthread_key_create().
// It is the responsibility of the application to free the data related to
// the deleted key in any running thread. No destructor is invoked by
// this function. Any destructor that may have been associated with key
// will no longer be called upon thread exit.
// Returns 0 on success, error code otherwise.
extern int bthread_key_delete(bthread_key_t key) __THROW;
// Store data' in the thread-specific slot identified by key’.
// bthread_setspecific() is callable from within destructor. If the application
// does so, destructors will be repeatedly called for at most
// PTHREAD_DESTRUCTOR_ITERATIONS times to clear the slots.
// NOTE: If the thread is not created by brpc server and lifetime is
// very short(doing a little thing and exit), avoid using bthread-local. The
// reason is that bthread-local always allocate keytable on first call to
// bthread_setspecific, the overhead is negligible in long-lived threads,
// but noticeable in shortly-lived threads. Threads in brpc server
// are special since they reuse keytables from a bthread_keytable_pool_t
// in the server.
// Returns 0 on success, error code otherwise.
// If the key is invalid or deleted, return EINVAL.
extern int bthread_setspecific(bthread_key_t key, void* data) __THROW;
// Return current value of the thread-specific slot identified by `key’.
// If bthread_setspecific() had not been called in the thread, return NULL.
// If the key is invalid or deleted, return NULL.
extern void* bthread_getspecific(bthread_key_t key) __THROW;
使用方法
用bthread_key_create创建一个bthread_key_t，它代表一种bthread私有变量。
用bthread_[get|set]specific查询和设置bthread私有变量。一个线程中第一次访问某个私有变量返回NULL。
在所有线程都不使用和某个bthread_key_t相关的私有变量后再删除它。如果删除了一个仍在被使用的bthread_key_t，相关的私有变量就泄露了。
static void my_data_destructor(void* data) {
…
}
bthread_key_t tls_key;
if (bthread_key_create(&tls_key, my_data_destructor) != 0) {
LOG(ERROR) << “Fail to create tls_key”;
return -1;
}
// in some thread …
MyThreadLocalData* tls = static_cast<MyThreadLocalData*>(bthread_getspecific(tls_key));
if (tls == NULL) { // First call to bthread_getspecific (and before any bthread_setspecific) returns NULL
tls = new MyThreadLocalData; // Create thread-local data on demand.
CHECK_EQ(0, bthread_setspecific(tls_key, tls)); // set the data so that next time bthread_getspecific in the thread returns the data.
}
示例代码
static void my_thread_local_data_deleter(void* d) {
delete static_cast<MyThreadLocalData*>(d);
}
class EchoServiceImpl : public example::EchoService {
public:
EchoServiceImpl() {
CHECK_EQ(0, bthread_key_create(&_tls2_key, my_thread_local_data_deleter));
}
~EchoServiceImpl() {
CHECK_EQ(0, bthread_key_delete(_tls2_key));
};
…
private:
bthread_key_t _tls2_key;
}
class EchoServiceImpl : public example::EchoService {
public:
…
void Echo(google::protobuf::RpcController* cntl_base,
const example::EchoRequest* request,
example::EchoResponse* response,
google::protobuf::Closure* done) {
…
// You can create bthread-local data for your own.
// The interfaces are similar with pthread equivalence:
// pthread_key_create -> bthread_key_create
// pthread_key_delete -> bthread_key_delete
// pthread_getspecific -> bthread_getspecific
// pthread_setspecific -> bthread_setspecific
MyThreadLocalData* tls2 = static_cast<MyThreadLocalData*>(bthread_getspecific(_tls2_key));
if (tls2 == NULL) {
tls2 = new MyThreadLocalData;
CHECK_EQ(0, bthread_setspecific(_tls2_key, tls2));
}
…
FAQ
Q: Fail to write into fd=1865 SocketId=8905@10.208.245.43:54742@8230: Got EOF是什么意思
A: 一般是client端使用了连接池或短连接模式，在RPC超时后会关闭连接，server写回response时发现连接已经关了就报这个错。Got EOF就是指之前已经收到了EOF（对端正常关闭了连接）。client端使用单连接模式server端一般不会报这个。
Q: Remote side of fd=9 SocketId=2@10.94.66.55:8000 was closed是什么意思
这不是错误，是常见的warning，表示对端关掉连接了（EOF)。这个日志有时对排查问题有帮助。
默认关闭，把参数-log_connection_close设置为true就打开了（支持动态修改）。
Q: 为什么server端线程数设了没用
brpc同一个进程中所有的server共用线程，如果创建了多个server，最终的工作线程数多半是最大的那个ServerOptions.num_threads。
Q: 为什么client端的延时远大于server端的延时
可能是server端的工作线程不够用了，出现了排队现象。排查方法请查看高效率排查服务卡顿。
Q: 程序切换到brpc之后出现了像堆栈写坏的coredump
brpc的Server是运行在bthread之上，默认栈大小为1MB，而pthread默认栈大小为10MB，所以在pthread上正常运行的程序，在bthread上可能遇到栈不足。
注意：不是说程序core了就意味着”栈不够大“，只是因为这个试起来最容易，所以优先排除掉可能性。事实上百度内如此多的应用也很少碰到栈不够大的情况。
解决方案：添加以下gflags以调整栈大小，比如–stack_size_normal=10000000 --tc_stack_normal=1。第一个flag把栈大小修改为10MB，第二个flag表示每个工作线程缓存的栈的个数(避免每次都从全局拿).
Q: Fail to open /proc/self/io
有些内核没这个文件，不影响服务正确性，但如下几个bvar会无法更新：
process_io_read_bytes_second
process_io_write_bytes_second
process_io_read_second
process_io_write_second
Q: json串"[1,2,3]"没法直接转为protobuf message
这不是标准的json。最外层必须是花括号{}包围的json object。
附:Server端基本流程
/********** server **********/
/********** client **********/
事实速查
Channel.Init()是线程不安全的。 Channel.CallMethod()是线程安全的，一个Channel可以被所有线程同时使用。 Channel可以分配在栈上。 Channel在发送异步请求后可以析构。 没有brpc::Client这个类。 Channel
Client指发起请求的一端，在brpc中没有对应的实体，取而代之的是brpc::Channel，它代表和一台或一组服务器的交互通道，Client和Channel在角色上的差别在实践中并不重要，你可以把Channel视作Client。
Channel可以被所有线程共用，你不需要为每个线程创建独立的Channel，也不需要用锁互斥。不过Channel的创建和Init并不是线程安全的，请确保在Init成功后再被多线程访问，在没有线程访问后再析构。
一些RPC实现中有ClientManager的概念，包含了Client端的配置信息和资源管理。brpc不需要这些，以往在ClientManager中配置的线程数、长短连接等等要么被加入了brpc::ChannelOptions，要么可以通过gflags全局配置，这么做的好处：
方便。你不需要在创建Channel时传入ClientManager，也不需要存储ClientManager。否则不少代码需要一层层地传递ClientManager，很麻烦。gflags使一些全局行为的配置更加简单。 共用资源。比如server和channel可以共用后台线程。(bthread的工作线程) 生命周期。析构ClientManager的过程很容易出错，现在由框架负责则不会有问题。 就像大部分类那样，Channel必须在Init之后才能使用，options为NULL时所有参数取默认值，如果你要使用非默认值，这么做就行了：
brpc::ChannelOptions options; // 包含了默认值
options.xxx = yyy;
…
channel.Init(…, &options);
注意Channel不会修改options，Init结束后不会再访问options。所以options一般就像上面代码中那样放栈上。Channel.options()可以获得channel在使用的所有选项。
Init函数分为连接一台服务器和连接服务集群。
连接一台服务器
// options为NULL时取默认值
int Init(EndPoint server_addr_and_port, const ChannelOptions* options);
int Init(const char* server_addr_and_port, const ChannelOptions* options);
int Init(const char* server_addr, int port, const ChannelOptions* options);
这类Init连接的服务器往往有固定的ip地址，不需要名字服务和负载均衡，创建起来相对轻量。但是请勿频繁创建使用域名的Channel。这需要查询dns，可能最多耗时10秒(查询DNS的默认超时)。重用它们。
合法的“server_addr_and_port”：
127.0.0.1:80 www.foo.com:8765 localhost:9000 不合法的"server_addr_and_port"：
127.0.0.1:90000 # 端口过大 10.39.2.300:8000 # 非法的ip 连接服务集群
int Init(const char* naming_service_url,
const char* load_balancer_name,
const ChannelOptions* options);
这类Channel需要定期从naming_service_url指定的名字服务中获得服务器列表，并通过load_balancer_name指定的负载均衡算法选择出一台机器发送请求。
你不应该在每次请求前动态地创建此类（连接服务集群的）Channel。因为创建和析构此类Channel牵涉到较多的资源，比如在创建时得访问一次名字服务，否则便不知道有哪些服务器可选。由于Channel可被多个线程共用，一般也没有必要动态创建。
当load_balancer_name为NULL或空时，此Init等同于连接单台server的Init，naming_service_url应该是"ip:port"或"域名:port"。你可以通过这个Init函数统一Channel的初始化方式。比如你可以把naming_service_url和load_balancer_name放在配置文件中，要连接单台server时把load_balancer_name置空，要连接服务集群时则设置一个有效的算法名称。
名字服务
名字服务把一个名字映射为可修改的机器列表，在client端的位置如下：
img
有了名字服务后client记录的是一个名字，而不是每一台下游机器。而当下游机器变化时，就只需要修改名字服务中的列表，而不需要逐台修改每个上游。这个过程也常被称为“解耦上下游”。当然在具体实现上，上游会记录每一台下游机器，并定期向名字服务请求或被推送最新的列表，以避免在RPC请求时才去访问名字服务。使用名字服务一般不会对访问性能造成影响，对名字服务的压力也很小。
naming_service_url的一般形式是"protocol://service_name"
bns://
BNS是百度内常用的名字服务，比如bns://rdev.matrix.all，其中"bns"是protocol，"rdev.matrix.all"是service-name。相关一个gflag是-ns_access_interval: img
如果BNS中显示不为空，但Channel却说找不到服务器，那么有可能BNS列表中的机器状态位（status）为非0，含义为机器不可用，所以不会被加入到server候选集中．状态位可通过命令行查看：
get_instance_by_service [bns_node_name] -s
file://
服务器列表放在path所在的文件里，比如"file://conf/local_machine_list"中的“conf/local_machine_list”对应一个文件，其中每行应是一台服务器的地址。当文件更新时, brpc会重新加载。
list://,…
服务器列表直接跟在list://之后，以逗号分隔，比如"list://db-bce-81-3-186.db01:7000,m1-bce-44-67-72.m1:7000,cp01-rd-cos-006.cp01:7000" 中有三个地址。
http://
连接一个域名下所有的机器, 例如http://www.baidu.com:80 ，注意连接单点的Init（两个参数）虽然也可传入域名，但只会连接域名下的一台机器。
名字服务过滤器
当名字服务获得机器列表后，可以自定义一个过滤器进行筛选，最后把结果传递给负载均衡：
img
过滤器的接口如下：
// naming_service_filter.h
class NamingServiceFilter {
public:
// Return true to take this `server’ as a candidate to issue RPC
// Return false to filter it out
virtual bool Accept(const ServerNode& server) const = 0;
};
// naming_service.h
struct ServerNode {
butil::EndPoint addr;
std::string tag;
};
常见的业务策略如根据server的tag进行过滤。
自定义的过滤器配置在ChannelOptions中，默认为NULL（不过滤）。
class MyNamingServiceFilter : public brpc::NamingServiceFilter {
public:
bool Accept(const brpc::ServerNode& server) const {
return server.tag == “main”;
}
};
int main() {
…
MyNamingServiceFilter my_filter;
…
brpc::ChannelOptions options;
options.ns_filter = &my_filter;
…
}
负载均衡
当下游机器超过一台时，我们需要分割流量，此过程一般称为负载均衡，在client端的位置如下图所示：
img
理想的算法是每个请求都得到及时的处理，且任意机器crash对全局影响较小。但由于client端无法及时获得server端的延迟或拥塞，而且负载均衡算法不能耗费太多的cpu，一般来说用户得根据具体的场景选择合适的算法，目前rpc提供的算法有（通过load_balancer_name指定）：
rr
即round robin，总是选择列表中的下一台服务器，结尾的下一台是开头，无需其他设置。比如有3台机器a,b,c，那么brpc会依次向a, b, c, a, b, c, …发送请求。注意这个算法的前提是服务器的配置，网络条件，负载都是类似的。
random
随机从列表中选择一台服务器，无需其他设置。和round robin类似，这个算法的前提也是服务器都是类似的。
la
locality-aware，优先选择延时低的下游，直到其延时高于其他机器，无需其他设置。实现原理请查看Locality-aware load balancing。
c_murmurhash or c_md5
一致性哈希，与简单hash的不同之处在于增加或删除机器时不会使分桶结果剧烈变化，特别适合cache类服务。
发起RPC前需要设置Controller.set_request_code()，否则RPC会失败。request_code一般是请求中主键部分的32位哈希值，不需要和负载均衡使用的哈希算法一致。比如用c_murmurhash算法也可以用md5计算哈希值。
src/brpc/policy/hasher.h中包含了常用的hash函数。如果用std::string key代表请求的主键，controller.set_request_code(brpc::policy::MurmurHash32(key.data(), key.size()))就正确地设置了request_code。
注意甄别请求中的“主键”部分和“属性”部分，不要为了偷懒或通用，就把请求的所有内容一股脑儿计算出哈希值，属性的变化会使请求的目的地发生剧烈的变化。另外也要注意padding问题，比如struct Foo { int32_t a; int64_t b; }在64位机器上a和b之间有4个字节的空隙，内容未定义，如果像hash(&foo, sizeof(foo))这样计算哈希值，结果就是未定义的，得把内容紧密排列或序列化后再算。
实现原理请查看Consistent Hashing。
健康检查
连接断开的server会被暂时隔离而不会被负载均衡算法选中，brpc会定期连接被隔离的server，以检查他们是否恢复正常，间隔由参数-health_check_interval控制:
Name Value Description Defined At
health_check_interval （R） 3 seconds between consecutive health-checkings src/brpc/socket_map.cpp
一旦server被连接上，它会恢复为可用状态。如果在隔离过程中，server从名字服务中删除了，brpc也会停止连接尝试。
发起访问
一般来说，我们不直接调用Channel.CallMethod，而是通过protobuf生成的桩XXX_Stub，过程更像是“调用函数”。stub内没什么成员变量，建议在栈上创建和使用，而不必new，当然你也可以把stub存下来复用。Channel::CallMethod和stub访问都是线程安全的，可以被所有线程同时访问。比如：
XXX_Stub stub(&channel);
stub.some_method(controller, request, response, done);
甚至
XXX_Stub(&channel).some_method(controller, request, response, done);
一个例外是http client。访问http服务和protobuf没什么关系，直接调用CallMethod即可，除了Controller和done均为NULL，详见访问HTTP服务。
同步访问
指的是：CallMethod会阻塞到收到server端返回response或发生错误（包括超时）。
同步访问中的response/controller不会在CallMethod后被框架使用，它们都可以分配在栈上。注意，如果request/response字段特别多字节数特别大的话，还是更适合分配在堆上。
MyRequest request;
MyResponse response;
brpc::Controller cntl;
XXX_Stub stub(&channel);
request.set_foo(…);
cntl.set_timeout_ms(…);
stub.some_method(&cntl, &request, &response, NULL);
if (cntl->Failed()) {
// RPC失败了. response里的值是未定义的，勿用。
} else {
// RPC成功了，response里有我们想要的回复数据。
}
异步访问
指的是：给CallMethod传递一个额外的回调对象done，CallMethod在发出request后就结束了，而不是在RPC结束后。当server端返回response或发生错误（包括超时）时，done->Run()会被调用。对RPC的后续处理应该写在done->Run()里，而不是CallMethod后。
由于CallMethod结束不意味着RPC结束，response/controller仍可能被框架及done->Run()使用，它们一般得创建在堆上，并在done->Run()中删除。如果提前删除了它们，那当done->Run()被调用时，将访问到无效内存。
你可以独立地创建这些对象，并使用NewCallback生成done，也可以把Response和Controller作为done的成员变量，一起new出来，一般使用前一种方法。
发起异步请求后Request和Channel也可以立刻析构。这两样和response/controller是不同的。注意:这是说Channel的析构可以立刻发生在CallMethod之后，并不是说析构可以和CallMethod同时发生，删除正被另一个线程使用的Channel是未定义行为（很可能crash）。
使用NewCallback
static void OnRPCDone(MyResponse* response, brpc::Controller* cntl) {
// unique_ptr会帮助我们在return时自动删掉response/cntl，防止忘记。gcc 3.4下的unique_ptr是模拟版本。
std::unique_ptr response_guard(response);
std::unique_ptrbrpc::Controller cntl_guard(cntl);
if (cntl->Failed()) {
// RPC失败了. response里的值是未定义的，勿用。
} else {
// RPC成功了，response里有我们想要的数据。开始RPC的后续处理。
}
// NewCallback产生的Closure会在Run结束后删除自己，不用我们做。
}
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));
由于protobuf 3把NewCallback设置为私有，r32035后brpc把NewCallback独立于src/brpc/callback.h（并增加了一些重载）。如果你的程序出现NewCallback相关的编译错误，把google::protobuf::NewCallback替换为brpc::NewCallback就行了。
继承google::protobuf::Closure
使用NewCallback的缺点是要分配三次内存：response, controller, done。如果profiler证明这儿的内存分配有瓶颈，可以考虑自己继承Closure，把response/controller作为成员变量，这样可以把三次new合并为一次。但缺点就是代码不够美观，如果内存分配不是瓶颈，别用这种方法。
class OnRPCDone: public google::protobuf::Closure {
public:
void Run() {
// unique_ptr会帮助我们在return时自动delete this，防止忘记。gcc 3.4下的unique_ptr是模拟版本。
std::unique_ptr 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);
如果异步访问中的回调函数特别复杂会有什么影响吗?
没有特别的影响，回调会运行在独立的bthread中，不会阻塞其他的逻辑。你可以在回调中做各种阻塞操作。
rpc发送处的代码和回调函数是在同一个线程里执行吗?
一定不在同一个线程里运行，即使该次rpc调用刚进去就失败了，回调也会在另一个bthread中运行。这可以在加锁进行rpc（不推荐）的代码中避免死锁。
等待RPC完成
注意：当你需要发起多个并发操作时，可能ParallelChannel更方便。
如下代码发起两个异步RPC后等待它们完成。
const brpc::CallId cid1 = controller1->call_id();
const brpc::CallId cid2 = controller2->call_id();
…
stub.method1(controller1, request1, response1, done1);
stub.method2(controller2, request2, response2, done2);
…
brpc::Join(cid1);
brpc::Join(cid2);
在发起RPC前调用Controller.call_id()获得一个id，发起RPC调用后Join那个id。
Join()的行为是等到RPC结束且done->Run()运行后，一些Join的性质如下：
如果对应的RPC已经结束，Join将立刻返回。 多个线程可以Join同一个id，它们都会醒来。 同步RPC也可以在另一个线程中被Join，但一般不会这么做。 Join()在之前的版本叫做JoinResponse()，如果你在编译时被提示deprecated之类的，修改为Join()。
在RPC调用后Join(controller->call_id())是错误的行为，一定要先把call_id保存下来。因为RPC调用后controller可能被随时开始运行的done删除。下面代码的Join方式是错误的。
static void on_rpc_done(Controller* controller, MyResponse* response) {
… Handle response …
delete controller;
delete response;
}
Controller* controller1 = new Controller;
Controller* controller2 = new Controller;
MyResponse* response1 = new MyResponse;
MyResponse* response2 = new MyResponse;
…
stub.method1(controller1, &request1, response1, google::protobuf::NewCallback(on_rpc_done, controller1, response1));
stub.method2(controller2, &request2, response2, google::protobuf::NewCallback(on_rpc_done, controller2, response2));
…
brpc::Join(controller1->call_id()); // 错误，controller1可能被on_rpc_done删除了
brpc::Join(controller2->call_id()); // 错误，controller2可能被on_rpc_done删除了
半同步
Join可用来实现“半同步”访问：即等待多个异步访问完成。由于调用处的代码会等到所有RPC都结束后再醒来，所以controller和response都可以放栈上。
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());
brpc::DoNothing()可获得一个什么都不干的done，专门用于半同步访问。它的生命周期由框架管理，用户不用关心。
注意在上面的代码中，我们在RPC结束后又访问了controller.call_id()，这是没有问题的，因为DoNothing中并不会像上节中的on_rpc_done中那样删除Controller。
取消RPC
brpc::StartCancel(call_id)可取消对应的RPC，call_id必须在发起RPC前通过Controller.call_id()获得，其他时刻都可能有race condition。
注意：是brpc::StartCancel(call_id)，不是controller->StartCancel()，后者被禁用，没有效果。后者是protobuf默认提供的接口，但是在controller对象的生命周期上有严重的竞争问题。
顾名思义，StartCancel调用完成后RPC并未立刻结束，你不应该碰触Controller的任何字段或删除任何资源，它们自然会在RPC结束时被done中对应逻辑处理。如果你一定要在原地等到RPC结束（一般不需要），则可通过Join(call_id)。
关于StartCancel的一些事实：
call_id在发起RPC前就可以被取消，RPC会直接结束（done仍会被调用）。 call_id可以在另一个线程中被取消。 取消一个已经取消的call_id不会有任何效果。推论：同一个call_id可以被多个线程同时取消，但最多一次有效果。 这里的取消是纯client端的功能，server端未必会取消对应的操作，server cancelation是另一个功能。 获取Server的地址和端口
remote_side()方法可知道request被送向了哪个server，返回值类型是butil::EndPoint，包含一个ip4地址和端口。在RPC结束前调用这个方法都是没有意义的。
打印方式：
LOG(INFO) << “remote_side=” << cntl->remote_side();
printf(“remote_side=%s\n”, butil::endpoint2str(cntl->remote_side()).c_str());
获取Client的地址和端口
r31384后通过local_side()方法可在RPC结束后获得发起RPC的地址和端口。
打印方式：
LOG(INFO) << “local_side=” << cntl->local_side();
printf(“local_side=%s\n”, butil::endpoint2str(cntl->local_side()).c_str());
应该重用brpc::Controller吗?
不用刻意地重用，但Controller是个大杂烩，可能会包含一些缓存，Reset()可以避免反复地创建这些缓存。
在大部分场景下，构造Controller(snippet1)和重置Controller(snippet2)的性能差异不大。
// snippet1
for (int i = 0; i < n; ++i) {
brpc::Controller controller;
…
stub.CallSomething(…, &controller);
}
// snippet2
brpc::Controller controller;
for (int i = 0; i < n; ++i) {
controller.Reset();
…
stub.CallSomething(…, &controller);
}
但如果snippet1中的Controller是new出来的，那么snippet1就会多出“内存分配”的开销，在一些情况下可能会慢一些。
设置
Client端的设置主要由三部分组成：
brpc::ChannelOptions: 定义在src/brpc/channel.h中，用于初始化Channel，一旦初始化成功无法修改。 brpc::Controller: 定义在src/brpc/controller.h中，用于在某次RPC中覆盖ChannelOptions中的选项，可根据上下文每次均不同。 全局gflags：常用于调节一些底层代码的行为，一般不用修改。请自行阅读服务/flags页面中的说明。 Controller包含了request中没有的数据和选项。server端和client端的Controller结构体是一样的，但使用的字段可能是不同的，你需要仔细阅读Controller中的注释，明确哪些字段可以在server端使用，哪些可以在client端使用。
一个Controller对应一次RPC。一个Controller可以在Reset()后被另一个RPC复用，但一个Controller不能被多个RPC同时使用（不论是否在同一个线程发起）。
Controller的特点：
一个Controller只能有一个使用者，没有特殊说明的话，Controller中的方法默认线程不安全。 因为不能被共享，所以一般不会用共享指针管理Controller，如果你用共享指针了，很可能意味着出错了。 Controller创建于开始RPC前，析构于RPC结束后，常见几种模式： 同步RPC前Controller放栈上，出作用域后自行析构。注意异步RPC的Controller绝对不能放栈上，否则其析构时异步调用很可能还在进行中，从而引发未定义行为。 异步RPC前new Controller，done中删除。 线程数
和大部分的RPC框架不同，brpc中并没有独立的Client线程池。所有Channel和Server通过bthread共享相同的线程池. 如果你的程序同样使用了brpc的server, 仅仅需要设置Server的线程数。 或者可以通过gflags设置-bthread_concurrency来设置全局的线程数.
超时
ChannelOptions.timeout_ms是对应Channel上所有RPC的总超时，Controller.set_timeout_ms()可修改某次RPC的值。单位毫秒，默认值1秒，最大值2^31（约24天），-1表示一直等到回复或错误。
ChannelOptions.connect_timeout_ms是对应Channel上所有RPC的连接超时，单位毫秒，默认值1秒。-1表示等到连接建立或出错，此值被限制为不能超过timeout_ms。注意此超时独立于TCP的连接超时，一般来说前者小于后者，反之则可能在connect_timeout_ms未达到前由于TCP连接超时而出错。
注意1：brpc中的超时是deadline，超过就意味着RPC结束，超时后没有重试。其他实现可能既有单次访问的超时，也有代表deadline的超时。迁移到brpc时请仔细区分。
注意2：RPC超时的错误码为ERPCTIMEDOUT (1008)，ETIMEDOUT的意思是连接超时，且可重试。
重试
ChannelOptions.max_retry是该Channel上所有RPC的默认最大重试次数，Controller.set_max_retry()可修改某次RPC的值，默认值3，0表示不重试。
r32111后Controller.retried_count()返回重试次数。
r34717后Controller.has_backup_request()获知是否发送过backup_request。
重试时框架会尽量避开之前尝试过的server。
重试的触发条件有(条件之间是AND关系）：
连接出错
如果server一直没有返回，但连接没有问题，这种情况下不会重试。如果你需要在一定时间后发送另一个请求，使用backup request。
工作机制如下：如果response没有在backup_request_ms内返回，则发送另外一个请求，哪个先回来就取哪个。新请求会被尽量送到不同的server。注意如果backup_request_ms大于超时，则backup request总不会被发送。backup request会消耗一次重试次数。backup request不意味着server端cancel。
ChannelOptions.backup_request_ms影响该Channel上所有RPC，单位毫秒，默认值-1（表示不开启），Controller.set_backup_request_ms()可修改某次RPC的值。
没到超时
超时后RPC会尽快结束。
有剩余重试次数
Controller.set_max_retry(0)或ChannelOptions.max_retry=0关闭重试。
错误值得重试
一些错误重试是没有意义的，就不会重试，比如请求有错时(EREQUEST)不会重试，因为server总不会接受,没有意义。
用户可以通过继承brpc::RetryPolicy自定义重试条件。比如brpc默认不重试HTTP相关的错误，而你的程序中希望在碰到HTTP_STATUS_FORBIDDEN (403)时重试，可以这么做：
#include <brpc/retry_policy.h>
class MyRetryPolicy : public brpc::RetryPolicy {
public:
bool DoRetry(const brpc::Controller* cntl) const {
if (cntl->ErrorCode() == brpc::EHTTP && // HTTP错误
cntl->http_response().status_code() == brpc::HTTP_STATUS_FORBIDDEN) {
return true;
}
// 把其他情况丢给框架。
return brpc::DefaultRetryPolicy()->DoRetry(cntl);
}
};
…
// 给ChannelOptions.retry_policy赋值就行了。
// 注意：retry_policy必须在Channel使用期间保持有效，Channel也不会删除retry_policy，所以大部分情况下RetryPolicy都应以单例模式创建。
brpc::ChannelOptions options;
static MyRetryPolicy g_my_retry_policy;
options.retry_policy = &g_my_retry_policy;
…
一些提示：
通过cntl->response()可获得对应RPC的response。 对ERPCTIMEDOUT代表的RPC超时总是不重试，即使你继承的RetryPolicy中允许。 重试应当保守
由于成本的限制，大部分线上server的冗余度是有限的，主要是满足多机房互备的需求。而激进的重试逻辑很容易导致众多client对server集群造成2-3倍的压力，最终使集群雪崩：由于server来不及处理导致队列越积越长，使所有的请求得经过很长的排队才被处理而最终超时，相当于服务停摆。默认的重试是比较安全的: 只要连接不断RPC就不会重试，一般不会产生大量的重试请求。用户可以通过RetryPolicy定制重试策略，但也可能使重试变成一场“风暴”。当你定制RetryPolicy时，你需要仔细考虑client和server的协作关系，并设计对应的异常测试，以确保行为符合预期。
协议
Channel的默认协议是baidu_std，可通过设置ChannelOptions.protocol换为其他协议，这个字段既接受enum也接受字符串。
目前支持的有：
PROTOCOL_BAIDU_STD 或 “baidu_std"，即百度标准协议，默认为单连接。 PROTOCOL_HULU_PBRPC 或 "hulu_pbrpc"，hulu的协议，默认为单连接。 PROTOCOL_NOVA_PBRPC 或 ”nova_pbrpc“，网盟的协议，默认为连接池。 PROTOCOL_HTTP 或 ”http", http 1.0或1.1协议，默认为连接池(Keep-Alive)。具体方法见访问HTTP服务。 PROTOCOL_SOFA_PBRPC 或 "sofa_pbrpc"，sofa-pbrpc的协议，默认为单连接。 PROTOCOL_PUBLIC_PBRPC 或 "public_pbrpc"，public_pbrpc的协议，默认为连接池。 PROTOCOL_UBRPC_COMPACK 或 "ubrpc_compack"，public/ubrpc的协议，使用compack打包，默认为连接池。具体方法见ubrpc (by protobuf)。相关的还有PROTOCOL_UBRPC_MCPACK2或ubrpc_mcpack2，使用mcpack2打包。 PROTOCOL_NSHEAD_CLIENT 或 "nshead_client"，这是发送baidu-rpc-ub中所有UBXXXRequest需要的协议，默认为连接池。具体方法见访问UB。 PROTOCOL_NSHEAD 或 "nshead"，这是发送NsheadMessage需要的协议，默认为连接池。具体方法见nshead+blob 。 PROTOCOL_MEMCACHE 或 "memcache"，memcached的二进制协议，默认为单连接。具体方法见访问memcached。 PROTOCOL_REDIS 或 "redis"，redis 1.2后的协议（也是hiredis支持的协议），默认为单连接。具体方法见访问Redis。 PROTOCOL_NSHEAD_MCPACK 或 "nshead_mcpack", 顾名思义，格式为nshead + mcpack，使用mcpack2pb适配，默认为连接池。 PROTOCOL_ESP 或 "esp"，访问使用esp协议的服务，默认为连接池。 连接方式
brpc支持以下连接方式：
短连接：每次RPC前建立连接，结束后关闭连接。由于每次调用得有建立连接的开销，这种方式一般用于偶尔发起的操作，而不是持续发起请求的场景。没有协议默认使用这种连接方式，http 1.0对连接的处理效果类似短链接。 连接池：每次RPC前取用空闲连接，结束后归还，一个连接上最多只有一个请求，一个client对一台server可能有多条连接。http 1.1和各类使用nshead的协议都是这个方式。 单连接：进程内所有client与一台server最多只有一个连接，一个连接上可能同时有多个请求，回复返回顺序和请求顺序不需要一致，这是baidu_std，hulu_pbrpc，sofa_pbrpc协议的默认选项。 短连接 连接池 单连接 长连接 否 是 是
server端连接数(单client) qpslatency (原理见little’s law) qpslatency 1
极限qps 差，且受限于单机端口数 中等 高
latency 1.5RTT(connect) + 1RTT + 处理时间 1RTT + 处理时间 1RTT + 处理时间
cpu占用 高, 每次都要tcp connect 中等, 每个请求都要一次sys write 低, 合并写出在大流量时减少cpu占用
框架会为协议选择默认的连接方式，用户一般不用修改。若需要，把ChannelOptions.connection_type设为：
CONNECTION_TYPE_SINGLE 或 "single" 为单连接 CONNECTION_TYPE_POOLED 或 "pooled" 为连接池, 与单个远端的最大连接数由-max_connection_pool_size控制: Name Value Description Defined At max_connection_pool_size (R) 100 maximum pooled connection count to a single endpoint src/brpc/socket.cpp CONNECTION_TYPE_SHORT 或 "short" 为短连接 设置为“”（空字符串）则让框架选择协议对应的默认连接方式。 brpc支持Streaming RPC，这是一种应用层的连接，用于传递流式数据。
关闭连接池中的闲置连接
当连接池中的某个连接在-idle_timeout_second时间内没有读写，则被视作“闲置”，会被自动关闭。默认值为10秒。此功能只对连接池(pooled)有效。打开-log_idle_connection_close在关闭前会打印一条日志。
Name Value Description Defined At
idle_timeout_second 10 Pooled connections without data transmission for so many seconds will be closed. No effect for non-positive values src/brpc/socket_map.cpp
log_idle_connection_close false Print log when an idle connection is closed src/brpc/socket.cpp
延迟关闭连接
多个channel可能通过引用计数引用同一个连接，当引用某个连接的最后一个channel析构时，该连接将被关闭。但在一些场景中，channel在使用前才被创建，用完立刻析构，这时其中一些连接就会被无谓地关闭再被打开，效果类似短连接。
一个解决办法是用户把所有或常用的channel缓存下来，这样自然能避免channel频繁产生和析构，但目前brpc没有提供这样一个utility，用户自己（正确）实现有一些工作量。
另一个解决办法是设置全局选项-defer_close_second
Name Value Description Defined At
defer_close_second 0 Defer close of connections for so many seconds even if the connection is not used by anyone. Close immediately for non-positive values src/brpc/socket_map.cpp
设置后引用计数清0时连接并不会立刻被关闭，而是会等待这么多秒再关闭，如果在这段时间内又有channel引用了这个连接，它会恢复正常被使用的状态。不管channel创建析构有多频率，这个选项使得关闭连接的频率有上限。这个选项的副作用是一些fd不会被及时关闭，如果延时被误设为一个大数值，程序占据的fd个数可能会很大。
连接的缓冲区大小
-socket_recv_buffer_size设置所有连接的接收缓冲区大小，默认-1（不修改）
-socket_send_buffer_size设置所有连接的发送缓冲区大小，默认-1（不修改）
Name Value Description Defined At
socket_recv_buffer_size -1 Set the recv buffer size of socket if this value is positive src/brpc/socket.cpp
socket_send_buffer_size -1 Set send buffer size of sockets if this value is positive src/brpc/socket.cpp
log_id
通过set_log_id()可设置64位整型log_id。这个id会和请求一起被送到服务器端，一般会被打在日志里，从而把一次检索经过的所有服务串联起来。字符串格式的需要转化为64位整形才能设入log_id。
附件
baidu_std和hulu_pbrpc协议支持附件，这段数据由用户自定义，不经过protobuf的序列化。站在client的角度，设置在Controller::request_attachment()的附件会被server端收到，response_attachment()则包含了server端送回的附件。附件不受压缩选项影响。
在http协议中，附件对应message body，比如要POST的数据就设置在request_attachment()中。
认证
client端的认证一般分为2种：
基于请求的认证：每次请求都会带上认证信息。这种方式比较灵活，认证信息中可以含有本次请求中的字段，但是缺点是每次请求都会需要认证，性能上有所损失 基于连接的认证：当TCP连接建立后，client发送认证包，认证成功后，后续该连接上的请求不再需要认证。相比前者，这种方式灵活度不高（一般ren认证包里只能携带本机一些静态信息），但性能较好，一般用于单连接/连接池场景 针对第一种认证场景，在实现上非常简单，将认证的格式定义加到请求结构体中，每次当做正常RPC发送出去即可；针对第二种场景，brpc提供了一种机制，只要用户继承实现：
class Authenticator {
public:
virtual ~Authenticator() {}
// Implement this method to generate credential information // into `auth_str' which will be sent to `VerifyCredential' // at server side. This method will be called on client side. // Returns 0 on success, error code otherwise virtual int GenerateCredential(std::string* auth_str) const = 0; };
那么当用户并发调用RPC接口用单连接往同一个server发请求时，框架会自动保证：建立TCP连接后，连接上的第一个请求中会带有上述GenerateCredential产生的认证包，其余剩下的并发请求不会带有认证信息，依次排在第一个请求之后。整个发送过程依旧是并发的，并不会等第一个请求先返回。若server端认证成功，那么所有请求都能成功返回；若认证失败，一般server端则会关闭连接，这些请求则会收到相应错误。
目前自带协议中支持客户端认证的有：baidu_std(默认协议), HTTP, hulu_pbrpc, ESP。对于自定义协议，一般可以在组装请求阶段，调用Authenticator接口生成认证串，来支持客户端认证。
重置
调用Reset方法可让Controller回到刚创建时的状态。
别在RPC结束前重置Controller，行为是未定义的。
压缩
set_request_compress_type()设置request的压缩方式，默认不压缩。
注意：附件不会被压缩。
HTTP body的压缩方法见client压缩request body。
支持的压缩方法有：
brpc::CompressTypeSnappy : snanpy压缩，压缩和解压显著快于其他压缩方法，但压缩率最低。 brpc::CompressTypeGzip : gzip压缩，显著慢于snappy，但压缩率高 brpc::CompressTypeZlib : zlib压缩，比gzip快10%~20%，压缩率略好于gzip，但速度仍明显慢于snappy。 下表是多种压缩算法应对重复率很高的数据时的性能，仅供参考。
Compress method Compress size(B) Compress time(us) Decompress time(us) Compress throughput(MB/s) Decompress throughput(MB/s) Compress ratio
Snappy 128 0.753114 0.890815 162.0875 137.0322 37.50%
Gzip 10.85185 1.849199 11.2488 66.01252 47.66%
Zlib 10.71955 1.66522 11.38763 73.30581 38.28%
Snappy 1024 1.404812 1.374915 695.1555 710.2713 8.79%
Gzip 16.97748 3.950946 57.52106 247.1718 6.64%
Zlib 15.98913 3.06195 61.07665 318.9348 5.47%
Snappy 16384 8.822967 9.865008 1770.946 1583.881 4.96%
Gzip 160.8642 43.85911 97.13162 356.2544 0.78%
Zlib 147.6828 29.06039 105.8011 537.6734 0.71%
Snappy 32768 16.16362 19.43596 1933.354 1607.844 4.82%
Gzip 229.7803 82.71903 135.9995 377.7849 0.54%
Zlib 240.7464 54.44099 129.8046 574.0161 0.50%
下表是多种压缩算法应对重复率很低的数据时的性能，仅供参考。
Compress method Compress size(B) Compress time(us) Decompress time(us) Compress throughput(MB/s) Decompress throughput(MB/s) Compress ratio
Snappy 128 0.866002 0.718052 140.9584 170.0021 105.47%
Gzip 15.89855 4.936242 7.678077 24.7294 116.41%
Zlib 15.88757 4.793953 7.683384 25.46339 107.03%
Snappy 1024 2.087972 1.06572 467.7087 916.3403 100.78%
Gzip 32.54279 12.27744 30.00857 79.5412 79.79%
Zlib 31.51397 11.2374 30.98824 86.90288 78.61%
Snappy 16384 12.598 6.306592 1240.276 2477.566 100.06%
Gzip 537.1803 129.7558 29.08707 120.4185 75.32%
Zlib 519.5705 115.1463 30.07291 135.697 75.24%
Snappy 32768 22.68531 12.39793 1377.543 2520.582 100.03%
Gzip 1403.974 258.9239 22.25825 120.6919 75.25%
Zlib 1370.201 230.3683 22.80687 135.6524 75.21%
FAQ
Q: brpc能用unix domain socket吗
不能。同机TCP socket并不走网络，相比unix domain socket性能只会略微下降。一些不能用TCP socket的特殊场景可能会需要，以后可能会扩展支持。
Q: Fail to connect to xx.xx.xx.xx:xxxx, Connection refused
一般是对端server没打开端口（很可能挂了）。
Q: 经常遇到至另一个机房的Connection timedout
img
这个就是连接超时了，调大连接和RPC超时：
struct ChannelOptions {
…
// Issue error when a connection is not established after so many
// milliseconds. -1 means wait indefinitely.
// Default: 200 (milliseconds)
// Maximum: 0x7fffffff (roughly 30 days)
int32_t connect_timeout_ms;
// Max duration of RPC over this Channel. -1 means wait indefinitely. // Overridable by Controller.set_timeout_ms(). // Default: 500 (milliseconds) // Maximum: 0x7fffffff (roughly 30 days) int32_t timeout_ms; ... };
注意: 连接超时不是RPC超时，RPC超时打印的日志是"Reached timeout=…"。
Q: 为什么同步方式是好的，异步就crash了
重点检查Controller，Response和done的生命周期。在异步访问中，RPC调用结束并不意味着RPC整个过程结束，而是在进入done->Run()时才会结束。所以这些对象不应在调用RPC后就释放，而是要在done->Run()里释放。你一般不能把这些对象分配在栈上，而应该分配在堆上。详见异步访问。
Q: 怎么确保请求只被处理一次
这不是RPC层面的事情。当response返回且成功时，我们确认这个过程一定成功了。当response返回且失败时，我们确认这个过程一定失败了。但当response没有返回时，它可能失败，也可能成功。如果我们选择重试，那一个成功的过程也可能会被再执行一次。一般来说带副作用的RPC服务都应当考虑幂等问题，否则重试可能会导致多次叠加副作用而产生意向不到的结果。只有读的检索服务大都没有副作用而天然幂等，无需特殊处理。而带写的存储服务则要在设计时就加入版本号或序列号之类的机制以拒绝已经发生的过程，保证幂等。
Q: Invalid address=`bns://group.user-persona.dumi.nj03’
FATAL 04-07 20:00:03 7778 src/brpc/channel.cpp:123] Invalid address=`bns://group.user-persona.dumi.nj03’. You should use Init(naming_service_name, load_balancer_name, options) to access multiple servers.
访问名字服务要使用三个参数的Init，其中第二个参数是load_balancer_name，而这里用的是两个参数的Init，框架认为是访问单点，就会报这个错。
Q: 两端都用protobuf，为什么不能互相访问
协议 !=protobuf。protobuf负责一个包的序列化，协议中的一个消息可能会包含多个protobuf包，以及额外的长度、校验码、magic number等等。打包格式相同不意味着协议可以互通。在brpc中写一份代码就能服务多协议的能力是通过把不同协议的数据转化为统一的编程接口完成的，而不是在protobuf层面。
Q: 为什么C++ client/server 能够互相通信， 和其他语言的client/server 通信会报序列化失败的错误
检查一下C++ 版本是否开启了压缩 (Controller::set_compress_type), 目前其他语言的rpc框架还没有实现压缩，互相返回会出现问题。
附:Client端基本流程
img
主要步骤：
创建一个bthread_id作为本次RPC的correlation_id。 根据Channel的创建方式，从进程级的SocketMap中或从LoadBalancer中选择一台下游server作为本次RPC发送的目的地。 根据连接方式（单连接、连接池、短连接），选择一个Socket。 如果开启验证且当前Socket没有被验证过时，第一个请求进入验证分支，其余请求会阻塞直到第一个包含认证信息的请求写入Socket。server端只对第一个请求进行验证。 根据Channel的协议，选择对应的序列化函数把request序列化至IOBuf。 如果配置了超时，设置定时器。从这个点开始要避免使用Controller对象，因为在设定定时器后随时可能触发超时->调用到用户的超时回调->用户在回调中析构Controller。 发送准备阶段结束，若上述任何步骤出错，会调用Channel::HandleSendFailed。 将之前序列化好的IOBuf写出到Socket上，同时传入回调Channel::HandleSocketFailed，当连接断开、写失败等错误发生时会调用此回调。 如果是同步发送，Join correlation_id；否则至此CallMethod结束。 网络上发消息+收消息。 收到response后，提取出其中的correlation_id，在O(1)时间内找到对应的Controller。这个过程中不需要查找全局哈希表，有良好的多核扩展性。 根据协议格式反序列化response。 调用Controller::OnRPCReturned，可能会根据错误码判断是否需要重试，或让RPC结束。如果是异步发送，调用用户回调。最后摧毁correlation_id唤醒Join着的线程。 /********** client **********/
搭建好Brpc的环境后，创建echo.proto并输入内容： option cc_generic_services = true;
message EchoRequest {
required string message = 1;
};
message EchoResponse {
required string message = 1;
};
service EchoService {
rpc Echo(EchoRequest) returns (EchoResponse);
};
proto文件的语法规则可参考：<https://www.cnblogs.com/yinheyi/p/6080244.html> [](https://www.cnblogs.com/yinheyi/p/6080244.html) 命令 protoc echo.proto --cpp\_out=. 在当前路径生成用于C++的头文件和源文件echo.pb.h和echo.pb.cc。 服务器代码： #include “…/…/output/include/butil/logging.h”
#include “…/…/output/include/brpc/server.h”
#include “echo.pb.h”
class MyEchoService : public EchoService
{
public:
MyEchoService() {};
virtual ~MyEchoService() {};
virtual void Echo(::google::protobuf::RpcController* controller,
const ::EchoRequest* request,
::EchoResponse* response,
::google::protobuf::Closure* done)
{
brpc::ClosureGuard done_guard(done);
brpc::Controller* cntl = static_castbrpc::Controller*(controller);
response->set_message(request->message());
LOG(INFO) << "remote_side=" << cntl->remote_side(); printf("remote_side=%s\n", butil::endpoint2str(cntl->remote_side()).c_str()); } };
int main(int argc, char* argv[])
{
brpc::Server server;
MyEchoService myechoservice;
if(server.AddService(&myechoservice, brpc::SERVER_OWNS_SERVICE) != 0) { LOG(FATAL) << "Fail to add myechoservice"; return -1; } if(server.Start("localhost:9000", NULL) != 0) { LOG(ERROR) << "Fail to start EchoServer"; return -1; } printf("Server starting...\n"); server.RunUntilAskedToQuit(); return 0; }
客户端代码： #include “…/…/output/include/butil/logging.h”
#include “…/…/output/include/brpc/channel.h”
#include “echo.pb.h”
int main(int argc, char* argv[])
{
brpc::Channel channel;
if(channel.Init("localhost:9000", NULL) != 0) { LOG(ERROR) << "Fail to initialize channel"; return -1; } EchoService_Stub stub(&channel); EchoRequest request; EchoResponse response; brpc::Controller cntl; request.set_message(argv[1]); stub.Echo(&cntl, &request, &response, NULL); if(!cntl.Failed()) { LOG(INFO) << "Received from " << cntl.remote_side() << " to " << cntl.local_side(); printf("Response : %s\n", response.message().c_str()); } else LOG(WARNING) << cntl.ErrorText(); return 0; }
g++ -L/usr/lib/x86\_64-linux-gnu -L../../output/lib -Xlinker "-(" echo.pb.o server.o -Wl,-Bstatic -lgflags -lprotobuf -lleveldb -lsnappy -lbrpc -Wl,-Bdynamic -Xlinker "-)" -lpthread -lrt -lssl -lcrypto -ldl -lz -o echo\_server 效果图：   **网上学习资料一大堆，但如果学到的知识不成体系，遇到问题时只是浅尝辄止，不再深入研究，那么很难做到真正的技术提升。** **[需要这份系统化的资料的朋友，可以添加戳这里获取](https://bbs.csdn.net/topics/618668825)** **一个人可以走的很快，但一群人才能走的更远！不论你是正从事IT行业的老鸟或是对IT行业感兴趣的新人，都欢迎加入我们的的圈子（技术交流、学习资源、职场吐槽、大厂内推、面试辅导），让我们一起学习成长！** } EchoService_Stub stub(&channel); EchoRequest request; EchoResponse response; brpc::Controller cntl; request.set_message(argv[1]); stub.Echo(&cntl, &request, &response, NULL); if(!cntl.Failed()) { LOG(INFO) << "Received from " << cntl.remote_side() << " to " << cntl.local_side(); printf("Response : %s\n", response.message().c_str()); } else LOG(WARNING) << cntl.ErrorText(); return 0; } g++ -L/usr/lib/x86_64-linux-gnu -L…/…/output/lib -Xlinker “-(” echo.pb.o server.o -Wl,-Bstatic -lgflags -lprotobuf -lleveldb -lsnappy -lbrpc -Wl,-Bdynamic -Xlinker “-)” -lpthread -lrt -lssl -lcrypto -ldl -lz -o echo_server
效果图：
[外链图片转存中…(img-7vCbZ0EK-1715679794050)]
[外链图片转存中…(img-B0eVvRIJ-1715679794051)]
网上学习资料一大堆，但如果学到的知识不成体系，遇到问题时只是浅尝辄止，不再深入研究，那么很难做到真正的技术提升。
一个人可以走的很快，但一群人才能走的更远！不论你是正从事IT行业的老鸟或是对IT行业感兴趣的新人，都欢迎加入我们的的圈子（技术交流、学习资源、职场吐槽、大厂内推、面试辅导），让我们一起学习成长！
