Brpc学习:简单回显服务器/客户端

sudo apt-get install git g++ make libssl-dev
sudo apt-get install realpath libgflags-dev libprotobuf-dev libprotoc-dev protobuf-compiler libleveldb-dev
sudo apt-get install libsnappy-dev

sudo apt-get install gperf
sudo apt-get install libgoogle-perftools-dev

git clone https://github.com/brpc/brpc.git
cd ./brpc/
sh config_brpc.sh --headers=/usr/include --libs=/usr/lib --nodebugsymbols
make

cd example/echo_c++
make
./echo_server & ./echo_client

sudo apt-get install libgtest-dev
cd /usr/src/gtest
sudo cmake .
sudo make
sudo mv libgtest* /usr/lib/
cd ./brpc/
sh config_brpc.sh --headers=/usr/include --libs=/usr/lib
cd test/
make
sh run_tests.sh

/********** Makefile **********/

//config.mk
# Generated by config_brpc.sh, don't modify manually
HDRS=/usr/include/
LIBS=/usr/lib/x86_64-linux-gnu
PROTOC=/usr/bin/protoc
PROTOBUF_HDR=/usr/include/
CC=gcc
CXX=g++
GCC_VERSION=50400
STATIC_LINKINGS= -lgflags -lprotobuf -lleveldb -lsnappy
DYNAMIC_LINKINGS=-lpthread -lrt -lssl -lcrypto -ldl -lz
ifeq ($(NEED_LIBPROTOC), 1)
    STATIC_LINKINGS+=-lprotoc
endif
ifeq ($(NEED_GPERFTOOLS), 1)
    LIBS+=/usr/lib
    DYNAMIC_LINKINGS+=-ltcmalloc_and_profiler
endif
CPPFLAGS+=-DBRPC_WITH_GLOG=0 -DGFLAGS_NS=google 
ifeq ($(NEED_GTEST), 1)
    LIBS+=/usr/lib
    STATIC_LINKINGS+=-lgtest
    STATIC_LINKINGS+=-lgtest_main
endif

//Makefile
BRPC_PATH = ../../
include $(BRPC_PATH)/config.mk
# Notes on the flags:
# 1. Added -fno-omit-frame-pointer: perf/tcmalloc-profiler use frame pointers by default
# 2. Added -D__const__= : Avoid over-optimizations of TLS variables by GCC>=4.8
CXXFLAGS+=$(CPPFLAGS) -std=c++0x -DNDEBUG -O2 -D__const__= -pipe -W -Wall -Wno-unused-parameter -fPIC -fno-omit-frame-pointer
HDRS+=$(BRPC_PATH)/output/include
LIBS+=$(BRPC_PATH)/output/lib
HDRPATHS = $(addprefix -I, $(HDRS))
LIBPATHS = $(addprefix -L, $(LIBS))
COMMA=,
SOPATHS=$(addprefix -Wl$(COMMA)-rpath=, $(LIBS))

STATIC_LINKINGS += -lbrpc

CLIENT_SOURCES = client.cpp
SERVER_SOURCES = server.cpp
PROTOS = $(wildcard *.proto)

PROTO_OBJS = $(PROTOS:.proto=.pb.o)
PROTO_GENS = $(PROTOS:.proto=.pb.h) $(PROTOS:.proto=.pb.cc)
CLIENT_OBJS = $(addsuffix .o, $(basename $(CLIENT_SOURCES))) 
SERVER_OBJS = $(addsuffix .o, $(basename $(SERVER_SOURCES))) 

.PHONY:all
all: echo_client echo_server

.PHONY:clean
clean:
	@echo "Cleaning"
	@rm -rf echo_client echo_server $(PROTO_GENS) $(PROTO_OBJS) $(CLIENT_OBJS) $(SERVER_OBJS)

echo_client:$(PROTO_OBJS) $(CLIENT_OBJS)
	@echo "Linking $@"
ifneq ("$(LINK_SO)", "")
	@$(CXX) $(LIBPATHS) $(SOPATHS) -Xlinker "-(" $^ -Xlinker "-)" $(STATIC_LINKINGS) $(DYNAMIC_LINKINGS) -o $@
	#@$(CXX) $(LIBPATHS) $(SOPATHS) -Xlinker "-(" $^ -Xlinker "-)" $(STATIC_LINKINGS) $(DYNAMIC_LINKINGS) -o $@
else
	@$(CXX) $(LIBPATHS) -Xlinker "-(" $^ -Wl,-Bstatic $(STATIC_LINKINGS) -Wl,-Bdynamic -Xlinker "-)" $(DYNAMIC_LINKINGS) -o $@
	#@$(CXX) $(LIBPATHS) -Xlinker "-(" $^ -Wl,-Bstatic $(STATIC_LINKINGS) -Wl,-Bdynamic -Xlinker "-)" $(DYNAMIC_LINKINGS) -o $@
endif

echo_server:$(PROTO_OBJS) $(SERVER_OBJS)
	@echo "Linking $@"
ifneq ("$(LINK_SO)", "")
	@$(CXX) $(LIBPATHS) $(SOPATHS) -Xlinker "-(" $^ -Xlinker "-)" $(STATIC_LINKINGS) $(DYNAMIC_LINKINGS) -o $@
	#@$(CXX) $(LIBPATHS) $(SOPATHS) -Xlinker "-(" $^ -Xlinker "-)" $(STATIC_LINKINGS) $(DYNAMIC_LINKINGS) -o $@
else
	@$(CXX) $(LIBPATHS) -Xlinker "-(" $^ -Wl,-Bstatic $(STATIC_LINKINGS) -Wl,-Bdynamic -Xlinker "-)" $(DYNAMIC_LINKINGS) -o $@
	#@$(CXX) $(LIBPATHS) -Xlinker "-(" $^ -Wl,-Bstatic $(STATIC_LINKINGS) -Wl,-Bdynamic -Xlinker "-)" $(DYNAMIC_LINKINGS) -o $@
endif

%.pb.cc %.pb.h:%.proto
	@echo "Generating $@"
	@$(PROTOC) --cpp_out=. --proto_path=. $(PROTOC_EXTRA_ARGS) $<

%.o:%.cpp
	@echo "Compiling $@"
	@$(CXX) -c $(HDRPATHS) $(CXXFLAGS) $< -o $@

%.o:%.cc
	@echo "Compiling $@"
	@$(CXX) -c $(HDRPATHS) $(CXXFLAGS) $< -o $@

/********** server **********/

填写proto文件 https://www.cnblogs.com/yinheyi/p/6080244.html

请求、回复、服务的接口均定义在proto文件中。

# 告诉protoc要生成C++ Service基类，如果是java或python，则应分别修改为java_generic_services和py_generic_services
option cc_generic_services = true;
 
message EchoRequest {
      required string message = 1;
};
message EchoResponse {
      required string message = 1;
};
 
service EchoService {
      rpc Echo(EchoRequest) returns (EchoResponse);
};br

protobuf的更多用法请阅读protobuf官方文档。
实现生成的Service接口

protoc运行后会生成echo.pb.cc和echo.pb.h文件，你得include echo.pb.h，实现其中的EchoService基类：

#include "echo.pb.h"
...
class MyEchoService : public EchoService  {
public:
    void Echo(::google::protobuf::RpcController* cntl_base,
              const ::example::EchoRequest* request,
              ::example::EchoResponse* response,
              ::google::protobuf::Closure* done) {
        // 这个对象确保在return时自动调用done->Run()
        brpc::ClosureGuard done_guard(done);
         
        brpc::Controller* cntl = static_cast<brpc::Controller*>(cntl_base);
 
        // 填写response
        response->set_message(request->message());
    }
};

Service在插入brpc.Server后才可能提供服务。

当客户端发来请求时，Echo()会被调用。参数的含义分别是：

controller

在brpc中可以静态转为brpc::Controller（前提是代码运行brpc.Server中），包含了所有request和response之外的参数集合，具体接口查阅controller.h

request

请求，只读的，来自client端的数据包。

response

回复。需要用户填充，如果存在required字段没有被设置，该次调用会失败。

done

done由框架创建，递给服务回调，包含了调用服务回调后的后续动作，包括检查response正确性，序列化，打包，发送等逻辑。

不管成功失败，done->Run()必须在请求处理完成后被用户调用一次。

为什么框架不自己调用done->Run()？这是为了允许用户把done保存下来，在服务回调之后的某事件发生时再调用，即实现异步Service。

强烈建议使用ClosureGuard确保done->Run()被调用，即在服务回调开头的那句：

brpc::ClosureGuard done_guard(done);

不管在中间还是末尾脱离服务回调，都会使done_guard析构，其中会调用done->Run()。这个机制称为RAII(Resource Acquisition Is Initialization也称为“资源获取就是初始化”)。没有这个的话你得在每次return前都加上done->Run()，极易忘记。

在异步Service中，退出服务回调时请求未处理完成，done->Run()不应被调用，done应被保存下来供以后调用，乍看起来，这里并不需要用ClosureGuard。但在实践中，异步Service照样会因各种原因跳出回调，如果不使用ClosureGuard，一些分支很可能会在return前忘记done->Run()，所以我们也建议在异步service中使用done_guard，与同步Service不同的是，为了避免正常脱离函数时done->Run()也被调用，你可以调用done_guard.release()来释放其中的done。

一般来说，同步Service和异步Service分别按如下代码处理done：

class MyFooService: public FooService  {
public:
    // 同步服务
    void SyncFoo(::google::protobuf::RpcController* cntl_base,
                 const ::example::EchoRequest* request,
                 ::example::EchoResponse* response,
                 ::google::protobuf::Closure* done) {
         brpc::ClosureGuard done_guard(done);
         ...
    }
 
    // 异步服务
    void AsyncFoo(::google::protobuf::RpcController* cntl_base,
                  const ::example::EchoRequest* request,
                  ::example::EchoResponse* response,
                  ::google::protobuf::Closure* done) {
         brpc::ClosureGuard done_guard(done);
         ...
         done_guard.release();
    }
};

ClosureGuard的接口如下：

// RAII: Call Run() of the closure on destruction.
class ClosureGuard {
public:
    ClosureGuard();
    // Constructed with a closure which will be Run() inside dtor.
    explicit ClosureGuard(google::protobuf::Closure* done);
    
    // Call Run() of internal closure if it's not NULL.
    ~ClosureGuard();
 
    // Call Run() of internal closure if it's not NULL and set it to `done'.
    void reset(google::protobuf::Closure* done);
 
    // Set internal closure to NULL and return the one before set.
    google::protobuf::Closure* release();
};

标记当前调用为失败

调用Controller.SetFailed()可以把当前调用设置为失败，当发送过程出现错误时，框架也会调用这个函数。用户一般是在服务的CallMethod里调用这个函数，比如某个处理环节出错，SetFailed()后确认done->Run()被调用了就可以跳出函数了(若使用了ClosureGuard，跳出函数时会自动调用done，不用手动)。Server端的done的逻辑主要是发送response回client，当其发现用户调用了SetFailed()后，会把错误信息送回client。client收到后，它的Controller::Failed()会为true（成功时为false），Controller::ErrorCode()和Controller::ErrorText()则分别是错误码和错误信息。

用户可以为http访问设置status-code，在server端一般是调用controller.http_response().set_status_code()，标准的status-code定义在http_status_code.h中。如果SetFailed了但没有设置status-code，框架会代为选择和错误码最接近的status-code，实在没有相关的则填brpc::HTTP_STATUS_INTERNAL_SERVER_ERROR（500错误）
获取Client的地址

controller->remote_side()可获得发送该请求的client地址和端口，类型是butil::EndPoint。如果client是nginx，remote_side()是nginx的地址。要获取真实client的地址，可以在nginx里设置proxy_header ClientIp $remote_addr;, 在rpc中通过controller->http_request().GetHeader("ClientIp")获得对应的值。

打印方式：

LOG(INFO) << "remote_side=" << cntl->remote_side(); 
printf("remote_side=%sn", butil::endpoint2str(cntl->remote_side()).c_str());

获取Server的地址

controller->local_side()获得server端的地址，类型是butil::EndPoint。

打印方式：

LOG(INFO) << "local_side=" << cntl->local_side(); 
printf("local_side=%sn", butil::endpoint2str(cntl->local_side()).c_str());

异步Service

即done->Run()在Service回调之外被调用。

有些server以等待后端服务返回结果为主，且处理时间特别长，为了及时地释放出线程资源，更好的办法是把done注册到被等待事件的回调中，等到事件发生后再调用done->Run()。

异步service的最后一行一般是done_guard.release()以确保正常退出CallMethod时不会调用done->Run()。例子请看example/session_data_and_thread_local。

Service和Channel都可以使用done来表达后续的操作，但它们是完全不同的，请勿混淆：

    Service的done由框架创建，用户处理请求后调用done把response发回给client。
    Channel的done由用户创建，待RPC结束后被框架调用以执行用户的后续代码。

在一个会访问下游服务的异步服务中会同时接触两者，容易搞混，请注意区分。
加入Service

默认构造后的Server不包含任何服务，也不会对外提供服务，仅仅是一个对象。

通过如下方法插入你的Service实例。

int AddService(google::protobuf::Service* service, ServiceOwnership ownership);

若ownership参数为SERVER_OWNS_SERVICE，Server在析构时会一并删除Service，否则应设为SERVER_DOESNT_OWN_SERVICE。

插入MyEchoService代码如下：

brpc::Server server;
MyEchoService my_echo_service;
if (server.AddService(&my_echo_service, brpc::SERVER_DOESNT_OWN_SERVICE) != 0) {
    LOG(FATAL) << "Fail to add my_echo_service";
    return -1;
}

Server启动后你无法再修改其中的Service。
启动

调用以下Server的一下接口启动服务。

int Start(const char* ip_and_port_str, const ServerOptions* opt);
int Start(EndPoint ip_and_port, const ServerOptions* opt);
int Start(int port, const ServerOptions* opt);
int Start(const char *ip_str, PortRange port_range, const ServerOptions *opt);  // r32009后增加

"localhost:9000", "cq01-cos-dev00.cq01:8000", “127.0.0.1:7000"都是合法的ip_and_port_str。

options为NULL时所有参数取默认值，如果你要使用非默认值，这么做就行了：

brpc::ServerOptions options;  // 包含了默认值
options.xxx = yyy;
...
server.Start(..., &options);

监听多个端口

一个server只能监听一个端口（不考虑ServerOptions.internal_port），需要监听N个端口就起N个Server。
停止

server.Stop(closewait_ms); // closewait_ms实际无效，出于历史原因未删
server.Join();

Stop()不会阻塞，Join()会。分成两个函数的原因在于当多个Server需要退出时，可以先全部Stop再一起Join，如果一个个Stop/Join，可能得花费Server个数倍的等待时间。

不管closewait_ms是什么值，server在退出时会等待所有正在被处理的请求完成，同时对新请求立刻回复ELOGOFF错误以防止新请求加入。这么做的原因在于只要server退出时仍有处理线程运行，就有访问到已释放内存的风险。如果你的server“退不掉”，很有可能是由于某个检索线程没结束或忘记调用done了。

当client看到ELOGOFF时，会跳过对应的server，并在其他server上重试对应的请求。所以在一般情况下brpc总是“优雅退出”的，重启或上线时几乎不会或只会丢失很少量的流量。

RunUntilAskedToQuit()函数可以在大部分情况下简化server的运转和停止代码。在server.Start后，只需如下代码即会让server运行直到按到Ctrl-C。

// Wait until Ctrl-C is pressed, then Stop() and Join() the server.
server.RunUntilAskedToQuit();
 
// server已经停止了，这里可以写释放资源的代码。

Join()完成后可以修改其中的Service，并重新Start。
被HTTP client访问

使用Protobuf的服务通常可以通过HTTP+json访问，存于http body的json串可与对应protobuf消息相互转化。以echo server为例，你可以用curl访问这个服务。

# -H 'Content-Type: application/json' is optional
$ curl -d '{"message":"hello"}' http://brpc.baidu.com:8765/EchoService/Echo
{"message":"hello"}

注意：也可以指定Content-Type: application/proto用http+protobuf二进制串访问服务，序列化性能更好。
json<=>pb

json字段通过匹配的名字和结构与pb字段一一对应。json中一定要包含pb的required字段，否则转化会失败，对应请求会被拒绝。json中可以包含pb中没有定义的字段，但它们会被丢弃而不会存入pb的unknown字段。转化规则详见json <=> protobuf。

开启选项-pb_enum_as_number后，pb中的enum会转化为它的数值而不是名字，比如在enum MyEnum { Foo = 1; Bar = 2; };中不开启此选项时MyEnum类型的字段会转化为"Foo"或"Bar"，开启后为1或2。此选项同时影响client发出的请求和server返回的回复。由于转化为名字相比数值有更好的前后兼容性，此选项只应用于兼容无法处理enum为名字的老代码。
兼容早期版本client

早期的brpc允许一个pb service被http协议访问时不设置pb请求，即使里面有required字段。一般来说这种service会自行解析http请求和设置http回复，并不会访问pb请求。但这也是非常危险的行为，毕竟这是pb service，但pb请求却是未定义的。

这种服务在升级到新版本rpc时会遇到障碍，因为brpc已不允许这种行为。为了帮助这种服务升级，brpc允许经过一些设置后不把http body自动转化为pb request(从而可自行处理），方法如下：

brpc::ServiceOptions svc_opt;
svc_opt.ownership = ...;
svc_opt.restful_mappings = ...;
svc_opt.allow_http_body_to_pb = false; //关闭http body至pb request的自动转化
server.AddService(service, svc_opt);

如此设置后service收到http请求后不会尝试把body转化为pb请求，所以pb请求总是未定义状态，用户得在cntl->request_protocol() == brpc::PROTOCOL_HTTP认定请求是http时自行解析http body。

相应地，当cntl->response_attachment()不为空且pb回复不为空时，框架不再报错，而是直接把cntl->response_attachment()作为回复的body。这个功能和设置allow_http_body_to_pb与否无关。如果放开自由度导致过多的用户犯错，可能会有进一步的调整。
协议支持

server端会自动尝试其支持的协议，无需用户指定。cntl->protocol()可获得当前协议。server能从一个listen端口建立不同协议的连接，不需要为不同的协议使用不同的listen端口，一个连接上也可以传输多种协议的数据包, 但一般不会这么做(也不建议)，支持的协议有：

    百度标准协议，显示为"baidu_std"，默认启用。

    流式RPC协议，显示为"streaming_rpc", 默认启用。

    http 1.0/1.1，显示为”http“，默认启用。

    RTMP协议，显示为"rtmp", 默认启用。

    hulu-pbrpc的协议，显示为"hulu_pbrpc"，默认启动。

    sofa-pbrpc的协议，显示为”sofa_pbrpc“, 默认启用。

    百盟的协议，显示为”nova_pbrpc“, 默认不启用，开启方式：

    #include <brpc/policy/nova_pbrpc_protocol.h>
    ...
    ServerOptions options;
    ...
    options.nshead_service = new brpc::policy::NovaServiceAdaptor;

    public_pbrpc协议，显示为"public_pbrpc"，默认不启用，开启方式：

    #include <brpc/policy/public_pbrpc_protocol.h>
    ...
    ServerOptions options;
    ...
    options.nshead_service = new brpc::policy::PublicPbrpcServiceAdaptor;

    nshead+mcpack协议，显示为"nshead_mcpack"，默认不启用，开启方式：

    #include <brpc/policy/nshead_mcpack_protocol.h>
    ...
    ServerOptions options;
    ...
    options.nshead_service = new brpc::policy::NsheadMcpackAdaptor;

    顾名思义，这个协议的数据包由nshead+mcpack构成，mcpack中不包含特殊字段。不同于用户基于NsheadService的实现，这个协议使用了mcpack2pb，使得一份代码可以同时处理mcpack和pb两种格式。由于没有传递ErrorText的字段，当发生错误时server只能关闭连接。

    和UB相关的协议请阅读实现NsheadService。

如果你有更多的协议需求，可以联系我们。
设置
版本

Server.set_version(...)可以为server设置一个名称+版本，可通过/version内置服务访问到。虽然叫做"version“，但设置的值请包含服务名，而不仅仅是一个数字版本。
关闭闲置连接

如果一个连接在ServerOptions.idle_timeout_sec对应的时间内没有读取或写出数据，则被视为”闲置”而被server主动关闭。默认值为-1，代表不开启。

打开-log_idle_connection_close后关闭前会打印一条日志。
Name 	Value 	Description 	Defined At
log_idle_connection_close 	false 	Print log when an idle connection is closed 	src/brpc/socket.cpp
pid_file

如果设置了此字段，Server启动时会创建一个同名文件，内容为进程号。默认为空。
在每条日志后打印hostname

此功能只对butil/logging.h中的日志宏有效。

打开-log_hostname后每条日志后都会带本机名称，如果所有的日志需要汇总到一起进行分析，这个功能可以帮助你了解某条日志来自哪台机器。
打印FATAL日志后退出程序

此功能只对butil/logging.h中的日志宏有效，glog默认在FATAL日志时crash。

打开-crash_on_fatal_log后如果程序使用LOG(FATAL)打印了异常日志或违反了CHECK宏中的断言，那么程序会在打印日志后abort，这一般也会产生coredump文件，默认不打开。这个开关可在对程序的压力测试中打开，以确认程序没有进入过严重错误的分支。

    一般的惯例是，ERROR表示可容忍的错误，FATAL代表不可逆转的错误。

最低日志级别

此功能由butil/logging.h和glog各自实现，为同名选项。

只有不低于-minloglevel指定的日志级别的日志才会被打印。这个选项可以动态修改。设置值和日志级别的对应关系：0=INFO 1=NOTICE 2=WARNING 3=ERROR 4=FATAL，默认为0。

未打印日志的开销只是一次if判断，也不会评估参数(比如某个参数调用了函数，日志不打，这个函数就不会被调用）。如果日志最终打印到自定义LogSink，那么还要经过LogSink的过滤。
归还空闲内存至系统

选项-free_memory_to_system_interval表示每过这么多秒就尝试向系统归还空闲内存，<= 0表示不开启，默认值为0，若开启建议设为10及以上的值。此功能支持tcmalloc，之前程序中对MallocExtension::instance()->ReleaseFreeMemory()的定期调用可改成设置此选项。
打印发送给client的错误

server的框架部分一般不针对个别client打印错误日志，因为当大量client出现错误时，可能导致server高频打印日志而严重影响性能。但有时为了调试问题，或就是需要让server打印错误，打开参数-log_error_text即可。
限制最大消息

为了保护server和client，当server收到的request或client收到的response过大时，server或client会拒收并关闭连接。此最大尺寸由-max_body_size控制，单位为字节。

超过最大消息时会打印如下错误日志：

FATAL: 05-10 14:40:05: * 0 src/brpc/input_messenger.cpp:89] A message from 127.0.0.1:35217(protocol=baidu_std) is bigger than 67108864 bytes, the connection will be closed. Set max_body_size to allow bigger messages

protobuf中有类似的限制，出错时会打印如下日志：

FATAL: 05-10 13:35:02: * 0 google/protobuf/io/coded_stream.cc:156] A protocol message was rejected because it was too big (more than 67108864 bytes). To increase the limit (or to disable these warnings), see CodedInputStream::SetTotalBytesLimit() in google/protobuf/io/coded_stream.h.

brpc移除了protobuf中的限制，全交由此选项控制，只要-max_body_size足够大，用户就不会看到错误日志。此功能对protobuf的版本没有要求。
压缩

set_response_compress_type()设置response的压缩方式，默认不压缩。

注意附件不会被压缩。HTTP body的压缩方法见这里。

支持的压缩方法有：

    brpc::CompressTypeSnappy : snanpy压缩，压缩和解压显著快于其他压缩方法，但压缩率最低。
    brpc::CompressTypeGzip : gzip压缩，显著慢于snappy，但压缩率高
    brpc::CompressTypeZlib : zlib压缩，比gzip快10%~20%，压缩率略好于gzip，但速度仍明显慢于snappy。

更具体的性能对比见Client-压缩.
附件

baidu_std和hulu_pbrpc协议支持传递附件，这段数据由用户自定义，不经过protobuf的序列化。站在server的角度，设置在Controller.response_attachment()的附件会被client端收到，Controller.request_attachment()则包含了client端送来的附件。

附件不会被框架压缩。

在http协议中，附件对应message body，比如要返回的数据就设置在response_attachment()中。
验证client身份

如果server端要开启验证功能，需要实现Authenticator中的接口：

class Authenticator {
public:
    // Implement this method to verify credential information `auth_str' from
    // `client_addr'. You can fill credential context (result) into `*out_ctx'
    // and later fetch this pointer from `Controller'.
    // Returns 0 on success, error code otherwise
    virtual int VerifyCredential(const std::string& auth_str,
                                 const base::EndPoint& client_addr,
                                 AuthContext* out_ctx) const = 0;
    }; 

class AuthContext {
public:
    const std::string& user() const;
    const std::string& group() const;
    const std::string& roles() const;
    const std::string& starter() const;
    bool is_service() const;
};

server的验证是基于连接的。当server收到连接上的第一个请求时，会尝试解析出其中的身份信息部分（如baidu_std里的auth字段、HTTP协议里的Authorization头），然后附带client地址信息一起调用VerifyCredential。若返回0，表示验证成功，用户可以把验证后的信息填入AuthContext，后续可通过controller->auth_context()获取，用户不需要关心其分配和释放。否则表示验证失败，连接会被直接关闭，client访问失败。

后续请求默认通过验证么，没有认证开销。

把实现的Authenticator实例赋值到ServerOptions.auth，即开启验证功能，需要保证该实例在整个server运行周期内都有效，不能被析构。
worker线程数

设置ServerOptions.num_threads即可，默认是cpu core的个数（包含超线程的）。

注意: ServerOptions.num_threads仅仅是个提示。

你不能认为Server就用了这么多线程，因为进程内的所有Server和Channel会共享线程资源，线程总数是所有ServerOptions.num_threads和-bthread_concurrency中的最大值。比如一个程序内有两个Server，num_threads分别为24和36，bthread_concurrency为16。那么worker线程数为max(24, 36, 16) = 36。这不同于其他RPC实现中往往是加起来。

Channel没有相应的选项，但可以通过选项-bthread_concurrency调整。

另外，brpc不区分IO线程和处理线程。brpc知道如何编排IO和处理代码，以获得更高的并发度和线程利用率。
限制最大并发

“并发”可能有两种含义，一种是连接数，一种是同时在处理的请求数。这里提到的是后者。

在传统的同步server中，最大并发不会超过工作线程数，设定工作线程数量一般也限制了并发。但brpc的请求运行于bthread中，M个bthread会映射至N个worker中（一般M大于N），所以同步server的并发度可能超过worker数量。另一方面，虽然异步server的并发不受线程数控制，但有时也需要根据其他因素控制并发量。

brpc支持设置server级和method级的最大并发，当server或method同时处理的请求数超过并发度限制时，它会立刻给client回复brpc::ELIMIT错误，而不会调用服务回调。看到ELIMIT错误的client应重试另一个server。这个选项可以防止server出现过度排队，或用于限制server占用的资源。

默认不开启。
为什么超过最大并发要立刻给client返回错误而不是排队？

当前server达到最大并发并不意味着集群中的其他server也达到最大并发了，立刻让client获知错误，并去尝试另一台server在全局角度是更好的策略。
为什么不限制QPS?

QPS是一个秒级的指标，无法很好地控制瞬间的流量爆发。而最大并发和当前可用的重要资源紧密相关："工作线程"，“槽位”等，能更好地抑制排队。

另外当server的延时较为稳定时，限制并发的效果和限制QPS是等价的。但前者实现起来容易多了：只需加减一个代表并发度的计数器。这也是大部分流控都限制并发而不是QPS的原因，比如TCP中的“窗口"即是一种并发度。
计算最大并发数

最大并发度 = 极限QPS * 平均延时 (little's law)

极限QPS和平均延时指的是server在没有严重积压请求的前提下（请求的延时仍能接受时）所能达到的最大QPS和当时的平均延时。一般的服务上线都会有性能压测，把测得的QPS和延时相乘一般就是该服务的最大并发度。
限制server级别并发度

设置ServerOptions.max_concurrency，默认值0代表不限制。访问内置服务不受此选项限制。

Server.ResetMaxConcurrency()可在server启动后动态修改server级别的max_concurrency。
限制method级别并发度

server.MaxConcurrencyOf("...") = ...可设置method级别的max_concurrency。可能的设置方法有：

server.MaxConcurrencyOf("example.EchoService.Echo") = 10;
server.MaxConcurrencyOf("example.EchoService", "Echo") = 10;
server.MaxConcurrencyOf(&service, "Echo") = 10;

此设置一般发生在AddService后，server启动前。当设置失败时（比如对应的method不存在），server会启动失败同时提示用户修正MaxConcurrencyOf设置错误。

当method级别和server级别的max_concurrency都被设置时，先检查server级别的，再检查method级别的。

注意：没有service级别的max_concurrency。
pthread模式

用户代码（客户端的done，服务器端的CallMethod）默认在栈为1MB的bthread中运行。但有些用户代码无法在bthread中运行，比如：

    JNI会检查stack layout而无法在bthread中运行。
    代码中广泛地使用pthread local传递session级别全局数据，在RPC前后均使用了相同的pthread local的数据，且数据有前后依赖性。比如在RPC前往pthread-local保存了一个值，RPC后又读出来希望和之前保存的相等，就会有问题。而像tcmalloc虽然也使用了pthread/LWP local，但每次使用之间没有直接的依赖，是安全的。

对于这些情况，brpc提供了pthread模式，开启**-usercode_in_pthread**后，用户代码均会在pthread中运行，原先阻塞bthread的函数转而阻塞pthread。

打开pthread模式后在性能上的注意点：

    同步RPC都会阻塞worker pthread，server端一般需要设置更多的工作线程(ServerOptions.num_threads)，调度效率会略微降低。
    运行用户代码的仍然是bthread，只是很特殊，会直接使用pthread worker的栈。这些特殊bthread的调度方式和其他bthread是一致的，这方面性能差异很小。
    bthread支持一个独特的功能：把当前使用的pthread worker 让给另一个新创建的bthread运行，以消除一次上下文切换。brpc client利用了这点，从而使一次RPC过程中3次上下文切换变为了2次。在高QPS系统中，消除上下文切换可以明显改善性能和延时分布。但pthread模式不具备这个能力，在高QPS系统中性能会有一定下降。
    pthread模式中线程资源是硬限，一旦线程被打满，请求就会迅速拥塞而造成大量超时。一个常见的例子是：下游服务大量超时后，上游服务可能由于线程大都在等待下游也被打满从而影响性能。开启pthread模式后请考虑设置ServerOptions.max_concurrency以控制server的最大并发。而在bthread模式中bthread个数是软限，对此类问题的反应会更加平滑。

pthread模式可以让一些老代码快速尝试brpc，但我们仍然建议逐渐地把代码改造为使用bthread local或最好不用TLS，从而最终能关闭这个开关。
安全模式

如果你的服务流量来自外部（包括经过nginx等转发），你需要注意一些安全因素：
对外隐藏内置服务

内置服务很有用，但包含了大量内部信息，不应对外暴露。有多种方式可以对外隐藏内置服务：

    设置内部端口。把ServerOptions.internal_port设为一个仅允许内网访问的端口。你可通过internal_port访问到内置服务，但通过对外端口(Server.Start时传入的那个)访问内置服务时将看到如下错误：

    [a27eda84bcdeef529a76f22872b78305] Not allowed to access builtin services, try ServerOptions.internal_port=... instead if you're inside internal network

    http proxy指定转发路径。nginx等可配置URL的映射关系，比如下面的配置把访问/MyAPI的外部流量映射到target-server的/ServiceName/MethodName。当外部流量尝试访问内置服务，比如说/status时，将直接被nginx拒绝。

  location /MyAPI {
      ...
      proxy_pass http://<target-server>/ServiceName/MethodName$query_string   # $query_string是nginx变量，更多变量请查询http://nginx.org/en/docs/http/ngx_http_core_module.html
      ...
  }

请勿在对外服务上开启-enable_dir_service和-enable_threads_service两个选项，它们虽然很方便，但会严重泄露服务器上的其他信息。检查对外的rpc服务是否打开了这两个开关：

curl -s -m 1 <HOSTNAME>:<PORT>/flags/enable_dir_service,enable_threads_service | awk '{if($3=="false"){++falsecnt}else if($3=="Value"){isrpc=1}}END{if(isrpc!=1||falsecnt==2){print "SAFE"}else{print "NOT SAFE"}}'

转义外部可控的URL

可调用brpc::WebEscape()对url进行转义，防止恶意URI注入攻击。
不返回内部server地址

可以考虑对server地址做签名。比如在设置ServerOptions.internal_port后，server返回的错误信息中的IP信息是其MD5签名，而不是明文。
定制/health页面

/health页面默认返回"OK"，若需定制/health页面的内容：先继承HealthReporter，在其中实现生成页面的逻辑（就像实现其他http service那样），然后把实例赋给ServerOptions.health_reporter，这个实例不被server拥有，必须保证在server运行期间有效。用户在定制逻辑中可以根据业务的运行状态返回更多样的状态信息。
线程私有变量

百度内的检索程序大量地使用了thread-local storage (缩写TLS)，有些是为了缓存频繁访问的对象以避免反复创建，有些则是为了在全局函数间隐式地传递状态。你应当尽量避免后者，这样的函数难以测试，不设置thread-local变量甚至无法运行。brpc中有三套机制解决和thread-local相关的问题。
session-local

session-local data与一次server端RPC绑定: 从进入service回调开始，到调用server端的done结束，不管该service是同步还是异步处理。 session-local data会尽量被重用，在server停止前不会被删除。

设置ServerOptions.session_local_data_factory后访问Controller.session_local_data()即可获得session-local数据。若没有设置，Controller.session_local_data()总是返回NULL。

若ServerOptions.reserved_session_local_data大于0，Server会在提供服务前就创建这么多个数据。

示例用法

struct MySessionLocalData {
    MySessionLocalData() : x(123) {}
    int x;
};
 
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_cast<brpc::Controller*>(cntl_base);
 
        // Get the session-local data which is created by ServerOptions.session_local_data_factory
        // and reused between different RPC.
        MySessionLocalData* sd = static_cast<MySessionLocalData*>(cntl->session_local_data());
        if (sd == NULL) {
            cntl->SetFailed("Require ServerOptions.session_local_data_factory to be set with a correctly implemented instance");
            return;
        }
        ...

struct ServerOptions {
    ...
    // The factory to create/destroy data attached to each RPC session.
    // If this field is NULL, Controller::session_local_data() is always NULL.
    // NOT owned by Server and must be valid when Server is running.
    // Default: NULL
    const DataFactory* session_local_data_factory;
 
    // Prepare so many session-local data before server starts, so that calls
    // to Controller::session_local_data() get data directly rather than
    // calling session_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_session_local_data;
};

session_local_data_factory的类型为DataFactory，你需要实现其中的CreateData和DestroyData。

注意：CreateData和DestroyData会被多个线程同时调用，必须线程安全。

class MySessionLocalDataFactory : public brpc::DataFactory {
public:
    void* CreateData() const {
        return new MySessionLocalData;
    }  
    void DestroyData(void* d) const {
        delete static_cast<MySessionLocalData*>(d);
    }  
};
 
int main(int argc, char* argv[]) {
    ...
    MySessionLocalDataFactory session_local_data_factory;
 
    brpc::Server server;
    brpc::ServerOptions options;
    ...
    options.session_local_data_factory = &session_local_data_factory;
    ...

server-thread-local

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_cast<brpc::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-name>

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>

服务器列表放在path所在的文件里，比如"file://conf/local_machine_list"中的“conf/local_machine_list”对应一个文件，其中每行应是一台服务器的地址。当文件更新时, brpc会重新加载。
list://<addr1>,<addr2>...

服务器列表直接跟在list://之后，以逗号分隔，比如"list://db-bce-81-3-186.db01:7000,m1-bce-44-67-72.m1:7000,cp01-rd-cos-006.cp01:7000" 中有三个地址。
http://<url>

连接一个域名下所有的机器, 例如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<MyResponse> response_guard(response);
    std::unique_ptr<brpc::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<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);

如果异步访问中的回调函数特别复杂会有什么影响吗?

没有特别的影响，回调会运行在独立的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=%sn", butil::endpoint2str(cntl->remote_side()).c_str());

获取Client的地址和端口

r31384后通过local_side()方法可在RPC结束后获得发起RPC的地址和端口。

打印方式：

LOG(INFO) << "local_side=" << cntl->local_side(); 
printf("local_side=%sn", 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) 	qps*latency (原理见little's law) 	qps*latency 	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 **********/

option cc_generic_services = true;

message EchoRequest {
      required string message = 1;
};

message EchoResponse {
      required string message = 1;
};

service EchoService {
      rpc Echo(EchoRequest) returns (EchoResponse);
};

#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_cast<brpc::Controller*>(controller);
        response->set_message(request->message());

        LOG(INFO) << "remote_side=" << cntl->remote_side();
        printf("remote_side=%sn", 
               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 : %sn", response.message().c_str());
    }
    else
        LOG(WARNING) << cntl.ErrorText();

    return 0;
}