Skip to content

B
brpc
Commit c6ae71e4 authored by gejun's avatar gejun Committed by iCode
Browse files
Options

Merge "+ Add more docs for client"

parents 9809ab9d 8d51edb8
Hide whitespace changes
Inline Side-by-side
Showing with 1165 additions and 0 deletions
docs/backup_request.md 0 → 100644
View file @ c6ae71e4
有时为了保证可用性，需要同时访问两路服务，哪个先返回就取哪个。在baidu-rpc中，这有多种做法：
# 当后端server可以挂在一个名字服务内时
Channel开启backup request。这个Channel会先向其中一个server发送请求，如果在ChannelOptions.backup_request_ms后还没回来，再向另一个server发送。之后哪个先回来就取哪个。在设置了合理的backup_request_ms后，大部分时候只会发一个请求，对后端服务只有一倍压力。
示例代码见[example/backup_request_c++](https://svn.baidu.com/public/trunk/baidu-rpc/example/backup_request_c++)。这个例子中，client设定了在2ms后发送backup request，server在碰到偶数位的请求后会故意睡眠20ms以触发backup request。
运行后，client端和server端的日志分别如下，“index”是请求的编号。可以看到server端在收到第一个请求后会故意sleep 20ms，client端之后发送另一个同样index的请求，最终的延时并没有受到故意sleep的影响。
![img](http://wiki.baidu.com/download/attachments/160281427/image2015-12-28%2019%3A48%3A54.png?version=1&modificationDate=1451303334000&api=v2)
![img](http://wiki.baidu.com/download/attachments/160281427/image2015-12-28%2019%3A48%3A2.png?version=1&modificationDate=1451303282000&api=v2)
/rpcz也显示client在2ms后触发了backup超时并发出了第二个请求。
![img](http://wiki.baidu.com/download/attachments/160281427/image2015-12-28%2019%3A54%3A22.png?version=1&modificationDate=1451303662000&api=v2)
## 选择合理的backup_request_ms
可以观察baidu-rpc默认提供的latency_cdf图，或自行添加。cdf图的y轴是延时（默认微秒），x轴是小于y轴延时的请求的比例。在下图中，选择backup_request_ms=2ms可以大约覆盖95.5%的请求，选择backup_request_ms=10ms则可以覆盖99.99%的请求。
![img](http://wiki.baidu.com/download/attachments/160281427/image2015-12-28%2021%3A23%3A48.png?version=1&modificationDate=1451309036000&api=v2)
自行添加的方法：
```c++
#include <bvar/bvar.h>
#include <base/time.h>
...
bvar::LatencyRecorder my_func_latency("my_func");
...
base::Timer tm;
tm.start();
my_func();
tm.stop();
my_func_latency << tm.u_elapsed(); // u代表微秒，还有s_elapsed(), m_elapsed(), n_elapsed()分别对应秒，毫秒，纳秒。
// 好了，在/vars中会显示my_func_qps, my_func_latency, my_func_latency_cdf等很多计数器。
```
# 当后端server不能挂在一个名字服务内时
【推荐】建立一个开启backup request的SelectiveChannel，其中包含两个sub channel。访问这个SelectiveChannel和上面的情况类似，会先访问一个sub channel，如果在ChannelOptions.backup_request_ms后没返回，再访问另一个sub channel。如果一个sub channel对应一个集群，这个方法就是在两个集群间做互备。SelectiveChannel的例子见[example/selective_echo_c++](https://svn.baidu.com/public/trunk/baidu-rpc/example/selective_echo_c++)，具体做法请参考上面的过程。
【不推荐】发起两个异步RPC后Join它们，它们的done内是相互取消的逻辑。示例代码见[example/cancel_c++](https://svn.baidu.com/public/trunk/baidu-rpc/example/cancel_c++)。这种方法的问题是总会发两个请求，对后端服务有两倍压力，这个方法怎么算都是不经济的，你应该尽量避免用这个方法。
\ No newline at end of file
docs/dummy_server.md 0 → 100644
View file @ c6ae71e4
如果你的程序只使用了baidu-rpc的client或根本没有使用baidu-rpc，但你也想使用baidu-rpc的内置服务，只要在程序中启动一个空的server就行了，这种server我们称为**dummy server**。
# 使用了baidu-rpc的client
只要在程序运行目录建立dummy_server.port文件，填入一个端口号（比如8888），程序会马上在这个端口上启动一个dummy server。在浏览器中访问它的内置服务，便可看到同进程内的所有bvar。
![img](http://wiki.baidu.com/download/attachments/71337189/image2015-12-25%2017%3A46%3A20.png?version=1&modificationDate=1451036781000&api=v2)
![img](http://wiki.baidu.com/download/attachments/71337189/image2015-12-25%2017%3A47%3A30.png?version=1&modificationDate=1451036850000&api=v2)
![img](http://wiki.baidu.com/download/attachments/71337189/image2015-12-25%2017%3A48%3A24.png?version=1&modificationDate=1451036904000&api=v2)
# 没有使用baidu-rpc
你必须手动加入dummy server。你得先查看[Getting Started](http://wiki.baidu.com/display/RPC/Getting+Started)如何下载和编译baidu-rpc，然后在程序入口处加入如下代码片段：
```c++
#include <baidu/rpc/server.h>
...
int main() {
...
baidu::rpc::Server dummy_server;
baidu::rpc::ServerOptions dummy_server_options;
dummy_server_options.num_threads = 0; // 不要改变寄主程序的线程数。
if (dummy_server.Start(8888/*port*/, &dummy_server_options) != 0) {
LOG(FATAL) << "Fail to start dummy server";
return -1;
}
...
}
```
r31803之后加入dummy server更容易了，只要一行：
```c++
#include <baidu/rpc/server.h>
...
int main() {
...
baidu::rpc::StartDummyServerAt(8888/*port*/);
...
}
```
\ No newline at end of file
docs/http_client.md 0 → 100644
View file @ c6ae71e4
http client的例子见[example/http_c++](https://svn.baidu.com/public/trunk/baidu-rpc/example/http_c++/http_client.cpp)
# 创建Channel
baidu::rpc::Channel可访问HTTP服务，ChannelOptions.protocol须指定为PROTOCOL_HTTP。
设定为HTTP协议后，`Channel::Init`的第一个参数可为任意合法的URL。注意：允许任意URL是为了省去用户取出host和port的麻烦，`Channel::Init`只用其中的host及port，其他部分都会丢弃。
```c++
baidu::rpc::ChannelOptions options;
options.protocol = baidu::rpc::PROTOCOL_HTTP;
if (channel.Init("www.baidu.com" /*any url*/, &options) != 0) {
LOG(ERROR) << "Fail to initialize channel";
return -1;
}
```
http channel也支持bns地址。
# GET
```c++
baidu::rpc::Controller cntl;
cntl.http_request().uri() = "www.baidu.com/index.html"; // 设置为待访问的URL
channel.CallMethod(NULL, &cntl, NULL, NULL, NULL/*done*/);
```
HTTP和protobuf无关，所以除了Controller和done，CallMethod的其他参数均为NULL。如果要异步操作，最后一个参数传入done。
`cntl.response_attachment()`是回复的body，类型也是base::IOBuf。注意IOBuf转化为std::string（通过to_string()接口）是需要分配内存并拷贝所有内容的，如果关注性能，你的处理过程应该尽量直接支持IOBuf，而不是要求连续内存。
# POST
默认的HTTP Method为GET，如果需要做POST，则需要设置。待POST的数据应置入request_attachment()，它([base::IOBuf](https://svn.baidu.com/public/trunk/iobuf/base/iobuf.h))可以直接append std::string或char*
```c++
baidu::rpc::Controller cntl;
cntl.http_request().uri() = "..."; // 设置为待访问的URL
cntl.http_request().set_method(baidu::rpc::HTTP_METHOD_POST);
cntl.request_attachment().append("{\"message\":\"hello world!\"}");
channel.CallMethod(NULL, &cntl, NULL, NULL, NULL/*done*/);
```
需要大量打印过程的body建议使用base::IOBufBuilder，它的用法和std::ostringstream是一样的。对于有大量对象要打印的场景，IOBufBuilder会简化代码，并且效率也更高。
```c++
baidu::rpc::Controller cntl;
cntl.http_request().uri() = "..."; // 设置为待访问的URL
cntl.http_request().set_method(baidu::rpc::HTTP_METHOD_POST);
base::IOBufBuilder os;
os << "A lot of printing" << printable_objects << ...;
os.move_to(cntl.request_attachment());
channel.CallMethod(NULL, &cntl, NULL, NULL, NULL/*done*/);
```
# URL
URL的一般形式如下图：
```c++
// URI scheme : http://en.wikipedia.org/wiki/URI_scheme
//
// foo://username:password@example.com:8042/over/there/index.dtb?type=animal&name=narwhal#nose
// \_/ \_______________/ \_________/ \__/ \___/ \_/ \______________________/ \__/
// | | | | | | | |
// | userinfo host port | | query fragment
// | \________________________________/\_____________|____|/ \__/ \__/
// schema | | | | | |
// authority | | | | |
// path | | interpretable as keys
// | |
// \_______________________________________________|____|/ \____/ \_____/
// | | | | |
// hierarchical part | | interpretable as values
// | |
// interpretable as filename |
// |
// |
// interpretable as extension
```
问题：Channel为什么不直接利用Init时传入的URL，而需要给uri()再设置一次？
确实，在简单使用场景下，这两者有所重复，但在复杂场景中，两者差别很大，比如：
- 访问挂在bns下的多个http server。此时Channel.Init传入的是bns节点名称，对uri()的赋值则是包含Host的完整URL（比如"www.foo.com/index.html?name=value"），BNS下所有的http server都会看到"Host: [www.foo.com](http://www.foo.com/)"；uri()也可以是只包含路径的URL，比如"/index.html?name=value"，框架会以目标server的ip和port为Host，地址为10.46.188.39:8989的http server将会看到"Host: 10.46.188.39:8989"。
- 通过http proxy访问目标server。此时Channel.Init传入的是proxy server的地址，但uri()填入的是目标server的URL。
# 常见设置
**以http request为例（对response的操作自行替换）**，常见操作方式如下表所示：
| 操作 | 方法 |
| -------------- | ---------------------------------------- |
| 访问名为Foo的header | `const std::string* value = cntl->http_request().GetHeader("Foo");` |
| | // 不存在为NULL |
| 设置名为Foo的header | `cntl->http_request().SetHeader("Foo", "value")` |
| 访问名为Foo的query | `const std::string* value = cntl->http_request().uri().GetQuery("Foo");` |
| | // 不存在为NULL |
| 设置名为Foo的query | `cntl->http_request().uri().SetQuery("Foo", "value")` |
| 设置HTTP方法 | `cntl->http_request().set_method(baidu::rpc::HTTP_METHOD_POST);` |
| 设置url | `cntl->http_request().uri() = "http://www.baidu.com";` |
| 设置content-type | `cntl->http_request().set_content_type("text/plain");` |
| 访问body | `base::IOBuf& buf = cntl->request_attachment();` |
| | `std::string str = cntl->request_attachment().to_string();` // 有拷贝 |
| 设置body | `cntl->request_attachment().append("....");` |
| | `base::IOBufBuilder os;``os << "....";` |
| | `os.move_to(cntl->request_attachment());` |
Notes on http header:
- 根据 HTTP 协议[规定](http://www.w3.org/Protocols/rfc2616/rfc2616-sec4.html#sec4.2)， header 的 field_name部分不区分大小写。在r33861前，field_name都转为了小写，在r33861后，大小写能保持不变（仍然支持大小写不敏感）。
- 如果 HTTP 头中出现了相同的 field_name, 根据协议[规定](http://www.w3.org/Protocols/rfc2616/rfc2616-sec4.html#sec4.2)，value将被合并到一起， 中间用逗号(,) 分隔， 具体value如何理解，需要用户自己确定.
- query之间用"&"分隔, key和value之间用"="分隔, value可以省略，比如key1=value1&key2&key3=value3中key2是合理的query，值为空字符串。
# 查看client发出的请求和收到的回复
打开[-http_verbose](http://brpc.baidu.com:8765/flags/http_verbose)即可在stderr看到所有的http request和response，注意这应该只用于线下调试，而不是线上程序。
# HTTP的错误处理
当Server返回的http status code不是2xx时，该次http访问即视为失败，client端会设置对应的ErrorCode：
- 在r31923前，1xx和3xx错误对应EPROTONOSUPPORT，4xx错误对应EREQUEST，其余错误对应EINTERNAL。http body不会置入`cntl->response_attachment()`。
- 在r31923后，所有错误被统一为EHTTP。如果用户发现`cntl->ErrorCode()`为EHTTP，那么可以检查`cntl->http_response().status_code()`以获得具体的http错误。同时http body会置入`cntl->response_attachment()`，用户可以把代表错误的html或json传递回来。
# 压缩request body
在r33877后，调用`Controller::set_request_compress_type(baidu::rpc::COMPRESS_TYPE_GZIP)`可将http body用gzip压缩，并设置"Content-Encoding"为"gzip"。
# 解压response body
出于通用性考虑且解压代码不复杂，baidu-rpc不会自动解压response body，用户可以自己做，方法如下：
```c++
#include <baidu/rpc/policy/gzip_compress.h>
...
const std::string* encoding = cntl->http_response().GetHeader("Content-Encoding");
if (encoding != NULL && *encoding == "gzip") {
base::IOBuf uncompressed;
if (!baidu::rpc::policy::GzipDecompress(cntl->response_attachment(), &uncompressed)) {
LOG(ERROR) << "Fail to un-gzip response body";
return;
}
cntl->response_attachment().swap(uncompressed);
}
// cntl->response_attachment()中已经是解压后的数据了
```
# 持续下载
r33796前baidu-rpc client在下载一个超长的body时，需要一直等待直到body完整才会视作RPC结束，这个过程中超长body都会存在内存中，如果body是无限长的（比如直播用的flv文件），那么内存会持续增长，直到超时。换句话说，r33796前的baidu-rpc client不适合下载大文件。
r33796后baidu-rpc client支持在读取完body前就结束RPC，让用户在RPC结束后再读取持续增长的body。注意这个功能不等同于“支持http chunked mode”，baidu-rpc的http实现一直支持解析chunked mode，这里的问题是如何让用户处理超长或无限长的body，和body是否以chunked mode传输无关。
使用方法如下：
1. 首先实现ProgressiveReader，接口如下：
```c++
#include <baidu/rpc/progressive_reader.h>
...
class ProgressiveReader {
public:
// Called when one part was read.
// Error returned is treated as *permenant* and the socket where the
// data was read will be closed.
// A temporary error may be handled by blocking this function, which
// may block the HTTP parsing on the socket.
virtual base::Status OnReadOnePart(const void* data, size_t length) = 0;
// Called when there's nothing to read anymore. The `status' is a hint for
// why this method is called.
// - status.ok(): the message is complete and successfully consumed.
// - otherwise: socket was broken or OnReadOnePart() failed.
// This method will be called once and only once. No other methods will
// be called after. User can release the memory of this object inside.
virtual void OnEndOfMessage(const base::Status& status) = 0;
};
```
OnReadOnePart在每读到一段数据时被调用，OnEndOfMessage在数据结束或连接断开时调用，实现前仔细阅读注释。
2. 发起RPC前设置`cntl.response_will_be_read_progressively();`
这告诉baidu-rpc在读取http response时只要读完header部分RPC就可以结束了。
3. RPC结束后调用`cntl.ReadProgressiveAttachmentBy(new MyProgressiveReader);`
MyProgressiveReader就是用户实现ProgressiveReader的实例。用户可以在这个实例的OnEndOfMessage接口中删除这个实例。
# 持续上传
目前Post的数据必须是完整生成好的，不适合POST超长的body。
# 访问带认证的Server
根据Server的认证方式生成对应的auth_data，并设置为http header "Authorization"的值。比如用的是curl，那就加上选项`-H "Authorization : <auth_data>"。`查询[giano文档](http://doc.noah.baidu.com/new/baas/base_tool.md)了解如何在Shell中生成auth_data。
\ No newline at end of file
docs/memcached_client.md 0 → 100644
View file @ c6ae71e4
[memcached](http://memcached.org/)是常用的缓存服务，为了使用户更快捷地访问memcached并充分利用bthread的并发能力，baidu-rpc直接支持memcache协议。示例程序：<https://svn.baidu.com/public/trunk/baidu-rpc/example/memcache_c++/>
> 注意：baidu-rpc只支持memcache的二进制协议。memcached在1.3前只有文本协议，但在当前看来支持的意义甚微。如果你的memcached早于1.3，升级版本。
相比使用[libmemcached](http://libmemcached.org/libMemcached.html)(官方client)的优势有：
- 线程安全。用户不需要为每个线程建立独立的client。
- 支持同步、异步、批量同步、批量异步等访问方式，能使用ParallelChannel等组合访问方式。
- 有明确的request和response。而libmemcached是没有的，收到的消息不能直接和发出的消息对应上，用户需要自己做维护工作。
- 支持多种[连接方式](http://wiki.baidu.com/pages/viewpage.action?pageId=213828702#id-访问Memcached-连接方式)。支持超时、backup request、取消、tracing、内置服务等一系列RPC基本福利。
当前实现充分利用了RPC的并发机制并尽量避免了拷贝。一个client可以轻松地把一个同机memcached实例([版本1.4.15](https://svn.baidu.com/third-64/tags/memcached/memcached_1-4-15-100_PD_BL/))压到极限：单连接9万，多连接33万。在大部分情况下，baidu-rpc应该能充分发挥memcached的性能。
# 访问单台memcached
创建一个访问memcached的Channel：
```c++
#include <baidu/rpc/memcache.h>
#include <baidu/rpc/channel.h>
ChannelOptions options;
options.protocol = baidu::rpc::PROTOCOL_MEMCACHE;
if (channel.Init("0.0.0.0:11211", &options) != 0) { // 11211是memcached的默认端口
LOG(FATAL) << "Fail to init channel to memcached";
return -1;
}
...
```
往memcached中设置一份数据。
```c++
// 写入key="hello" value="world" flags=0xdeadbeef，10秒失效，无视cas。
baidu::rpc::MemcacheRequest request;
baidu::rpc::MemcacheResponse response;
baidu::rpc::Controller cntl;
if (!request.Set("hello", "world", 0xdeadbeef/*flags*/, 10/*expiring seconds*/, 0/*ignore cas*/)) {
LOG(FATAL) << "Fail to SET request";
return -1;
}
channel.CallMethod(NULL, &cntl, &request, &response, NULL/*done*/);
if (cntl.Failed()) {
LOG(FATAL) << "Fail to access memcached, " << cntl.ErrorText();
return -1;
}
if (!response.PopSet(NULL)) {
LOG(FATAL) << "Fail to SET memcached, " << response.LastError();
return -1;
}
...
```
上述的代码有如下注意点：
- 请求类型必须为MemcacheRequest，回复类型必须为MemcacheResponse，否则CallMethod会失败。不需要stub，直接调用channel.CallMethod，method填NULL。
- 调用request.XXX()增加操作，本例XXX=Set，一个request多次调用不同的操作，这些操作会被同时送到memcached（常被称为pipeline模式）。
- 依次调用response.PopXXX()弹出操作结果，本例XXX=Set，成功返回true，失败返回false，调用response.LastError()可获得错误信息。XXX必须和request的依次对应，否则失败。本例中若用PopGet就会失败，错误信息为“not a GET response"。
- Pop结果独立于RPC结果。即使Set失败，RPC可能还是成功的。RPC失败意味着连接断开，超时之类的。“不能把某个值设入memcached”对于RPC来说还是成功的。如果业务上认为要成功操作才算成功，那么你不仅要判RPC成功，还要判PopXXX是成功的。
目前支持的请求操作有：
```c++
bool Set(const Slice& key, const Slice& value, uint32_t flags, uint32_t exptime, uint64_t cas_value);
bool Add(const Slice& key, const Slice& value, uint32_t flags, uint32_t exptime, uint64_t cas_value);
bool Replace(const Slice& key, const Slice& value, uint32_t flags, uint32_t exptime, uint64_t cas_value);
bool Append(const Slice& key, const Slice& value, uint32_t flags, uint32_t exptime, uint64_t cas_value);
bool Prepend(const Slice& key, const Slice& value, uint32_t flags, uint32_t exptime, uint64_t cas_value);
bool Delete(const Slice& key);
bool Flush(uint32_t timeout);
bool Increment(const Slice& key, uint64_t delta, uint64_t initial_value, uint32_t exptime);
bool Decrement(const Slice& key, uint64_t delta, uint64_t initial_value, uint32_t exptime);
bool Touch(const Slice& key, uint32_t exptime);
bool Version();
```
对应的回复操作：
```c++
// Call LastError() of the response to check the error text when any following operation fails.
bool PopGet(IOBuf* value, uint32_t* flags, uint64_t* cas_value);
bool PopGet(std::string* value, uint32_t* flags, uint64_t* cas_value);
bool PopSet(uint64_t* cas_value);
bool PopAdd(uint64_t* cas_value);
bool PopReplace(uint64_t* cas_value);
bool PopAppend(uint64_t* cas_value);
bool PopPrepend(uint64_t* cas_value);
bool PopDelete();
bool PopFlush();
bool PopIncrement(uint64_t* new_value, uint64_t* cas_value);
bool PopDecrement(uint64_t* new_value, uint64_t* cas_value);
bool PopTouch();
bool PopVersion(std::string* version);
```
# 访问memcached集群
建立一个使用c_md5负载均衡算法的channel，每个MemcacheRequest只包含一个操作或确保所有的操作始终落在同一台server，就能访问挂载在对应名字服务下的memcached集群了。如果request包含了多个操作，在当前实现下这些操作总会送向同一个server。比方说一个request中包含了多个Get操作，而对应的key分布在多个server上，那么结果就肯定不对了，这个情况下你必须把一个request分开为多个。
或者你可以沿用常见的[twemproxy](https://github.com/twitter/twemproxy)方案。这个方案虽然需要额外部署proxy，还增加了延时，但client端仍可以像访问单点一样的访问它。
\ No newline at end of file
docs/redis_client.md 0 → 100644
View file @ c6ae71e4
[redis](http://redis.io/)是最近几年比较火的缓存服务，相比memcached在server端提供了更多的数据结构和操作方法，简化了用户的开发工作，在百度内有比较广泛的应用。为了使用户更快捷地访问redis并充分利用bthread的并发能力，baidu-rpc直接支持redis协议。示例程序：<https://svn.baidu.com/public/trunk/baidu-rpc/example/redis_c++/>
相比使用[hiredis](https://github.com/redis/hiredis)(官方client)的优势有：
- 线程安全。用户不需要为每个线程建立独立的client。
- 支持同步、异步、批量同步、批量异步等访问方式，能使用ParallelChannel等组合访问方式。
- 支持多种[连接方式](http://wiki.baidu.com/pages/viewpage.action?pageId=213828685#id-创建和访问Client-连接方式)。支持超时、backup request、取消、tracing、内置服务等一系列RPC基本福利。
- 一个进程和一个redis-server只有一个连接。多个线程同时访问一个redis-server时更高效（见[性能](http://wiki.baidu.com/pages/viewpage.action?pageId=213828705#id-访问Redis-性能)）。无论reply的组成多复杂，内存都会连续成块地分配，并支持短串优化(SSO)。
像http一样，baidu-rpc保证在最差情况下解析redis reply的时间复杂度也是O(N)，N是reply的字节数，而不是O(N^2)。当reply是个较大的数组时，这是比较重要的。
r32037后加上[-redis_verbose](http://brpc.baidu.com:8765/flags/redis_verbose;redis_verbose_crlf2space)后会在stderr上打印出所有的redis request和response供调试。
# 访问单台redis
创建一个访问redis的Channel：
```c++
#include <baidu/rpc/redis.h>
#include <baidu/rpc/channel.h>
baidu::rpc::ChannelOptions options;
options.protocol = baidu::rpc::PROTOCOL_REDIS;
baidu::rpc::Channel redis_channel;
if (redis_channel.Init("0.0.0.0:6379", &options) != 0) { // 6379是redis-server的默认端口
LOG(ERROR) << "Fail to init channel to redis-server";
return -1;
}
...
```
执行SET后再INCR：
```c++
std::string my_key = "my_key_1";
int my_number = 1;
...
// 执行"SET <my_key> <my_number>"
baidu::rpc::RedisRequest set_request;
baidu::rpc::RedisResponse response;
baidu::rpc::Controller cntl;
set_request.AddCommand("SET %s %d", my_key.c_str(), my_number);
redis_channel.CallMethod(NULL, &cntl, &set_request, &response, NULL/*done*/);
if (cntl.Failed()) {
LOG(ERROR) << "Fail to access redis-server";
return -1;
}
if (response.reply(0).is_error()) {
LOG(ERROR) << "Fail to set";
return -1;
}
// 你可以通过response.reply(0).c_str()访问到值或多种方式打印出结果。
LOG(INFO) << response.reply(0).c_str() // OK
<< response.reply(0) // OK
<< response; // OK
...
// 执行"INCR <my_key>"
baidu::rpc::RedisRequest incr_request;
incr_request.AddCommand("INCR %s", my_key.c_str());
response.Clear();
cntl.Reset();
redis_channel.CallMethod(NULL, &cntl, &incr_request, &response, NULL/*done*/);
if (cntl.Failed()) {
LOG(ERROR) << "Fail to access redis-server";
return -1;
}
if (response.reply(0).is_error()) {
LOG(ERROR) << "Fail to incr";
return -1;
}
// 返回了incr后的值，你可以通过response.reply(0).integer()访问到值，或以多种方式打印结果。
LOG(INFO) << response.reply(0).integer() // 2
<< response.reply(0) // (integer) 2
<< response; // (integer) 2
```
批量执行incr或decr
```c++
baidu::rpc::RedisRequest request;
baidu::rpc::RedisResponse response;
baidu::rpc::Controller cntl;
request.AddCommand("INCR counter1");
request.AddCommand("DECR counter1");
request.AddCommand("INCRBY counter1 10");
request.AddCommand("DECRBY counter1 20");
redis_channel.CallMethod(NULL, &cntl, &get_request, &response, NULL/*done*/);
if (cntl.Failed()) {
LOG(ERROR) << "Fail to access redis-server";
return -1;
}
CHECK_EQ(4, response.reply_size());
for (int i = 0; i < 4; ++i) {
CHECK(response.reply(i).is_integer());
CHECK_EQ(baidu::rpc::REDIS_REPLY_INTEGER, response.reply(i).type());
}
CHECK_EQ(1, response.reply(0).integer());
CHECK_EQ(0, response.reply(1).integer());
CHECK_EQ(10, response.reply(2).integer());
CHECK_EQ(-10, response.reply(3).integer());
```
# RedisRequest
一个[RedisRequest](https://svn.baidu.com/public/trunk/baidu-rpc/src/baidu/rpc/redis.h)可包含多个Command，调用AddCommand*增加命令，成功返回true，失败返回false并会打印调用处的栈。
```c++
bool AddCommand(const char* fmt, ...);
bool AddCommandV(const char* fmt, va_list args);
bool AddCommandByComponents(const base::StringPiece* components, size_t n);
```
格式和hiredis基本兼容：即%b对应二进制数据（指针+length)，其他和printf的参数类似。对一些细节做了改进：当某个字段包含空格时，使用单引号或双引号包围起来会被视作一个字段。比如AddCommand("Set 'a key with space' 'a value with space as well'")中的key是a key with space，value是a value with space as well。在hiredis中必须写成redisvCommand(..., "SET %s %s", "a key with space", "a value with space as well");
AddCommandByComponents类似hiredis中的redisCommandArgv，用户通过数组指定命令中的每一个部分。这个方法不是最快捷的，但效率最高，且对AddCommand和AddCommandV可能发生的转义问题免疫，如果你在使用AddCommand和AddCommandV时出现了“引号不匹配”，“无效格式”等问题且无法定位，可以试下这个方法。
如果AddCommand*失败，后续的AddCommand*和CallMethod都会失败。一般来说不用判AddCommand*的结果，失败后自然会通过RPC失败体现出来。
command_size()可获得（成功）加入的命令个数。
调用Clear()后可重用RedisReques
# RedisResponse
[RedisResponse](https://svn.baidu.com/public/trunk/baidu-rpc/src/baidu/rpc/redis.h)可能包含一个或多个[RedisReply](https://svn.baidu.com/public/trunk/baidu-rpc/src/baidu/rpc/redis_reply.h)，reply_size()可获得reply的个数，reply(i)可获得第i个reply的引用（从0计数）。注意在hiredis中，如果请求包含了N个command，获取结果也要调用N次redisGetReply。但在baidu-rpc中这是不必要的，RedisResponse已经包含了N个reply，通过reply(i)获取就行了。只要RPC成功，response.reply_size()应与request.command_size()相等，除非redis-server有bug（redis-server工作的基本前提就是response和request按序一一对应）
每个reply可能是：
- REDIS_REPLY_NIL：redis中的NULL，代表值不存在。可通过is_nil()判定。
- REDIS_REPLY_STATUS：在redis文档中称为Simple String。一般是操作的返回值，比如SET返回的OK。可通过is_string()判定（status和string暂时不让用户区分），c_str()或data()获得值。
- REDIS_REPLY_STRING：在redis文档中称为Bulk String。大多数值都是这个类型，包括那些可以incr的。可通过is_string()判定，c_str()或data()获得值。
- REDIS_REPLY_ERROR：操作出错时的返回值，包含一段错误信息。可通过is_error()判定，error_message()获得错误信息。
- REDIS_REPLY_INTEGER：一个64位有符号数。可通过is_integer()判定，integer()获得值。
- REDIS_REPLY_ARRAY：另一些reply的数组。可通过is_array()判定，size()获得数组大小，[i]获得对应的子reply引用。
比如response包含三个reply，类型分别是integer，string和array (size=2)。那么可以分别这么获得值：response.reply(0).integer()，response.reply(1).c_str(), repsonse.reply(2)[0]和repsonse.reply(2)[1]。如果类型对不上，调用处的栈会被打印出来，并返回一个undefined的值。
> response中的所有reply的ownership属于response。当response析构时，reply也析构了。相应地，RedisReply被禁止拷贝。
调用Clear()后RedisResponse可以重用。
# 访问redis集群
暂时请沿用常见的[twemproxy](https://github.com/twitter/twemproxy)方案，像访问单点一样访问proxy。如果你之前用hiredis访问[BDRP](http://wiki.baidu.com/pages/viewpage.action?pageId=40197196)（使用了twemproxy），那把client更换成baidu-rpc就行了。通过client（一致性哈希）直接访问redis集群虽然能降低延时，但同时也（可能）意味着无法直接利用BDRP的托管服务，这一块还不是很确定。
如果你自己维护了redis集群，和memcache类似，应该是可以用一致性哈希访问的。但每个RedisRequest应只包含一个command或确保所有的command始终落在同一台server。如果request包含了多个command，在当前实现下总会送向同一个server。比方说一个request中包含了多个Get，而对应的key分布在多个server上，那么结果就肯定不对了，这个情况下你必须把一个request分开为多个。
# 查看发出的请求和收到的回复
打开[-redis_verbose](http://brpc.baidu.com:8765/flags/redis_verbose)即可在stderr看到所有的redis request和response，注意这应该只用于线下调试，而不是线上程序。
打开[-redis_verbose_crlf2space](http://brpc.baidu.com:8765/flags/redis_verbose_crlf2space)可让打印内容中的CRLF (\r\n)变为空格，方便阅读。
# 性能
redis版本：<https://svn.baidu.com/third-64/tags/redis/redis_2-6-14-100_PD_BL/> （不是最新的3.0）
分别使用1，50，200个bthread同步压测同机redis-server，延时单位均为微秒。
```
$ ./client -use_bthread -thread_num 1
TRACE: 02-13 19:42:04: * 0 client.cpp:180] Accessing redis server at qps=18668 latency=50
TRACE: 02-13 19:42:05: * 0 client.cpp:180] Accessing redis server at qps=17043 latency=52
TRACE: 02-13 19:42:06: * 0 client.cpp:180] Accessing redis server at qps=16520 latency=54
$ ./client -use_bthread -thread_num 50
TRACE: 02-13 19:42:54: * 0 client.cpp:180] Accessing redis server at qps=301212 latency=164
TRACE: 02-13 19:42:55: * 0 client.cpp:180] Accessing redis server at qps=301203 latency=164
TRACE: 02-13 19:42:56: * 0 client.cpp:180] Accessing redis server at qps=302158 latency=164
$ ./client -use_bthread -thread_num 200
TRACE: 02-13 19:43:48: * 0 client.cpp:180] Accessing redis server at qps=411669 latency=483
TRACE: 02-13 19:43:49: * 0 client.cpp:180] Accessing redis server at qps=411679 latency=483
TRACE: 02-13 19:43:50: * 0 client.cpp:180] Accessing redis server at qps=412583 latency=482
```
200个线程后qps基本到极限了。这里的极限qps比hiredis高很多，原因在于baidu-rpc默认以单链接访问redis-server，多个线程在写出时会[以wait-free的方式合并](http://wiki.baidu.com/display/RPC/IO#IO-发消息)，从而让redis-server就像被批量访问一样，每次都能从那个连接中读出一批请求，从而获得远高于非批量时的qps。下面通过连接池访问redis-server时qps的大幅回落是另外一个证明。
分别使用1，50，200个bthread一次发送10个同步压测同机redis-server，延时单位均为微秒。
```
$ ./client -use_bthread -thread_num 1 -batch 10
TRACE: 02-13 19:46:45: * 0 client.cpp:180] Accessing redis server at qps=15880 latency=59
TRACE: 02-13 19:46:46: * 0 client.cpp:180] Accessing redis server at qps=16945 latency=57
TRACE: 02-13 19:46:47: * 0 client.cpp:180] Accessing redis server at qps=16728 latency=57
$ ./client -use_bthread -thread_num 50 -batch 10
TRACE: 02-13 19:47:14: * 0 client.cpp:180] Accessing redis server at qps=38082 latency=1307
TRACE: 02-13 19:47:15: * 0 client.cpp:180] Accessing redis server at qps=38267 latency=1304
TRACE: 02-13 19:47:16: * 0 client.cpp:180] Accessing redis server at qps=38070 latency=1305
PID USER PR NI VIRT RES SHR S %CPU %MEM TIME+ COMMAND
16878 gejun 20 0 48136 2436 1004 R 93.8 0.0 12:48.56 redis-server // thread_num=50
$ ./client -use_bthread -thread_num 200 -batch 10
TRACE: 02-13 19:49:09: * 0 client.cpp:180] Accessing redis server at qps=29053 latency=6875
TRACE: 02-13 19:49:10: * 0 client.cpp:180] Accessing redis server at qps=29163 latency=6855
TRACE: 02-13 19:49:11: * 0 client.cpp:180] Accessing redis server at qps=29271 latency=6838
PID USER PR NI VIRT RES SHR S %CPU %MEM TIME+ COMMAND
16878 gejun 20 0 48136 2508 1004 R 99.9 0.0 13:36.59 redis-server // thread_num=200
```
注意redis-server实际处理的qps要乘10。乘10后也差不多在40万左右。另外在thread_num为50或200时，redis-server的CPU已打满。注意redis-server是[单线程reactor](http://wiki.baidu.com/display/RPC/Threading+Overview#ThreadingOverview-单线程reactor)，一个核心打满就意味server到极限了。
使用50个bthread通过连接池方式同步压测同机redis-server。
```
$ ./client -use_bthread -connection_type pooled
TRACE: 02-13 18:07:40: * 0 client.cpp:180] Accessing redis server at qps=75986 latency=654
TRACE: 02-13 18:07:41: * 0 client.cpp:180] Accessing redis server at qps=75562 latency=655
TRACE: 02-13 18:07:42: * 0 client.cpp:180] Accessing redis server at qps=75238 latency=657
PID USER PR NI VIRT RES SHR S %CPU %MEM TIME+ COMMAND
16878 gejun 20 0 48136 2520 1004 R 99.9 0.0 9:52.33 redis-server
```
可以看到qps相比单链接时有大幅回落，同时redis-server的CPU打满了。原因在于redis-server每次只能从一个连接中读到一个请求，IO开销大幅增加。
# Command Line Interface
example/redis_c++/redis_cli是一个类似于官方CLI的命令行工具，以展示baidu-rpc对redis协议的处理能力。当使用baidu-rpc访问redis-server出现不符合预期的行为时，也可以使用这个CLI进行交互式的调试。
```
$ ./redis_cli
__ _ __
/ /_ ____ _(_)___/ /_ __ _________ _____
/ __ \/ __ `/ / __ / / / /_____/ ___/ __ \/ ___/
/ /_/ / /_/ / / /_/ / /_/ /_____/ / / /_/ / /__
/_.___/\__,_/_/\__,_/\__,_/ /_/ / .___/\___/
/_/
This command-line tool mimics the look-n-feel of official redis-cli, as a
demostration of baidu-rpc's capability of talking to redis server. The
output and behavior is not exactly same with the official one.
redis 127.0.0.1:6379> mset key1 foo key2 bar key3 17
OK
redis 127.0.0.1:6379> mget key1 key2 key3
["foo", "bar", "17"]
redis 127.0.0.1:6379> incrby key3 10
(integer) 27
redis 127.0.0.1:6379> client setname baidu-rpc-cli
OK
redis 127.0.0.1:6379> client getname
"baidu-rpc-cli"
```
和官方CLI类似，redis_cli <command>也可以直接运行命令，-server参数可以指定redis-server的地址。
\ No newline at end of file
docs/streaming_rpc.md 0 → 100644
View file @ c6ae71e4
# 概述
在一些应用场景中， client或server需要像对面发送大量数据，这些数据非常大或者持续地在产生以至于无法放在一个RPC的附件中。比如一个分布式系统的不同节点间传递replica或snapshot。client/server之间虽然可以通过多次RPC把数据切分后传输过去，但存在如下问题：
- 如果这些RPC是并行的，无法保证接收端有序地收到数据，拼接数据的逻辑相当复杂。
- 如果这些RPC是串行的，每次传递都得等待一次网络RTT+处理数据的延时，特别是后者的延时可能是难以预估的。
为了让大块数据以流水线的方式在client/server之间传递， 我们提供了Streaming RPC这种交互模型。Streaming RPC让用户能够在client/service之间建立用户态连接，称为Stream, 同一个TCP连接之上能同时存在多个Stream。 Stream的传输数据以消息为基本单位， 输入端可以源源不断的往Stream中写入消息， 接收端会按输入端写入顺序收到消息。
Streaming RPC保证：
- 有消息边界。
- 接收消息的顺序和发送消息的顺序严格一致。
- 全双工。
- 支持流控。
- 提供超时提醒
目前的实现还没有自动切割过大的消息，同一个tcp连接上的多个Stream之间可能有[Head-of-line blocking](https://en.wikipedia.org/wiki/Head-of-line_blocking)问题，请尽量避免过大的单个消息，实现自动切割后我们会告知并更新文档。
例子见[example/streaming_echo_c++](https://svn.baidu.com/public/trunk/baidu-rpc/example/streaming_echo_c++/)。
# 建立Stream
目前Stream都由Client端建立。Client先在本地创建一个Stream，再通过一次RPC（必须使用标准协议）与指定的Service建立一个Stream，如果Service在收到请求之后选择接受这个Stream， 那在response返回Client后Stream就会建立成功。过程中的任何错误都把RPC标记为失败，同时也意味着Stream创建失败。用linux下建立连接的过程打比方，Client先创建[socket](http://linux.die.net/man/7/socket)（创建Stream），再调用[connect](http://linux.die.net/man/2/connect)尝试与远端建立连接（通过RPC建立Stream），远端[accept](http://linux.die.net/man/2/accept)后连接就建立了（service接受后创建成功）。
> 如果Client尝试向不支持Streaming RPC的老Server建立Stream，将总是失败。
程序中我们用StreamId代表一个Stream，对Stream的读写，关闭操作都将作用在这个Id上。
```c++
struct StreamOptions
// The max size of unconsumed data allowed at remote side.
// If |max_buf_size| <= 0, there's no limit of buf size
// default: 2097152 (2M)
int max_buf_size;
// Notify user when there's no data for at least |idle_timeout_ms|
// milliseconds since the last time that on_received_messages or on_idle_timeout
// finished.
// default: -1
long idle_timeout_ms;
// How many messages at most passed to handler->on_received_messages
// default: 1
size_t max_messages_size;
// Handle input message, if handler is NULL, the remote side is not allowd to
// write any message, who will get EBADF on writting
// default: NULL
StreamInputHandler* handler;
};
// [Called at the client side]
// Create a Stream at client-side along with the |cntl|, which will be connected
// when receiving the response with a Stream from server-side. If |options| is
// NULL, the Stream will be created with default options
// Return 0 on success, -1 otherwise
int StreamCreate(StreamId* request_stream, Controller &cntl, const StreamOptions* options);
```
# 接受Stream
如果client在RPC上附带了一个Stream， service在收到RPC后可以通过调用StreamAccept接受。接受后Server端对应产生的Stream存放在response_stream中，Server可通过这个Stream向Client发送数据。
```c++
// [Called at the server side]
// Accept the Stream. If client didn't create a Stream with the request
// (cntl.has_remote_stream() returns false), this method would fail.
// Return 0 on success, -1 otherwise.
int StreamAccept(StreamId* response_stream, Controller &cntl, const StreamOptions* options);
```
# 读取Stream
在建立或者接受一个Stream的时候， 用户可以继承StreamInputHandler并把这个handler填入StreamOptions中. 通过这个handler，用户可以处理对端的写入数据，连接关闭以及idle timeout
```c++
class StreamInputHandler {
public:
// 当接收到消息后被调用
virtual int on_received_messages(StreamId id, base::IOBuf *const messages[], size_t size) = 0;
// 当Stream上长时间没有数据交互后被调用
virtual void on_idle_timeout(StreamId id) = 0;
// 当Stream被关闭时被调用
virtual void on_closed(StreamId id) = 0;
};
```
>***第一次收到请求的时间***
>
>在client端，如果建立过程是一次同步RPC， 那在等待的线程被唤醒之后，on_received_message就可能会被调用到。 如果是异步RPC请求， 那等到这次请求的done->Run() 执行完毕之后， on_received_message就可能会被调用。
>
>在server端， 当框架传入的done->Run()被调用完之后， on_received_message就可能会被调用。
# 写入Stream
```c++
// Write |message| into |stream_id|. The remote-side handler will received the
// message by the written order
// Returns 0 on success, errno otherwise
// Errno:
// - EAGAIN: |stream_id| is created with positive |max_buf_size| and buf size
// which the remote side hasn't consumed yet excceeds the number.
// - EINVAL: |stream_id| is invalied or has been closed
int StreamWrite(StreamId stream_id, const base::IOBuf &message);
```
# 流控
当存在较多已发送但未接收的数据时，发送端的Write操作会立即失败(返回EAGAIN）， 这时候可以通过同步或异步的方式等待对端消费掉数据。
```c++
// Wait util the pending buffer size is less than |max_buf_size| or error occurs
// Returns 0 on success, errno otherwise
// Errno:
// - ETIMEDOUT: when |due_time| is not NULL and time expired this
// - EINVAL: the Stream was close during waiting
int StreamWait(StreamId stream_id, const timespec* due_time);
// Async wait
void StreamWait(StreamId stream_id, const timespec *due_time,
void (*on_writable)(StreamId stream_id, void* arg, int error_code),
void *arg);
```
# 关闭Stream
```c++
// Close |stream_id|, after this function is called:
// - All the following |StreamWrite| would fail
// - |StreamWait| wakes up immediately.
// - Both sides |on_closed| would be notifed after all the pending buffers have
// been received
// This function could be called multiple times without side-effects
int StreamClose(StreamId stream_id);
```
docs/ub_client.md 0 → 100644
View file @ c6ae71e4
baidu-rpc可通过多种方式访问用ub搭建的服务。
# ubrpc (by protobuf)
r31687后，baidu-rpc支持通过protobuf访问ubrpc，不需要baidu-rpc-ub，也不依赖idl-compiler。（也可以让protobuf服务被ubrpc client访问，方法见[使用ubrpc的服务](http://wiki.baidu.com/pages/viewpage.action?pageId=213828733#id-实现NsheadService-使用ubrpc的服务)）。
**步骤：**
1. 用[public/mcpack2pb/idl2proto](https://svn.baidu.com/public/trunk/mcpack2pb/idl2proto)把idl文件转化为proto文件，老版本idl2proto不会转化idl中的service，需要手动转化。
```protobuf
// Converted from echo.idl by public/mcpack2pb/idl2proto
import "idl_options.proto";
option (idl_support) = true;
option cc_generic_services = true;
message EchoRequest {
required string message = 1;
}
message EchoResponse {
required string message = 1;
}
// 对于idl中多个request或response的方法，要建立一个包含所有request或response的消息。
// 这个例子中就是MultiRequests和MultiResponses。
message MultiRequests {
required EchoRequest req1 = 1;
required EchoRequest req2 = 2;
}
message MultiResponses {
required EchoRequest res1 = 1;
required EchoRequest res2 = 2;
}
service EchoService {
// 对应idl中的void Echo(EchoRequest req, out EchoResponse res);
rpc Echo(EchoRequest) returns (EchoResponse);
// 对应idl中的uint32_t EchoWithMultiArgs(EchoRequest req1, EchoRequest req2, out EchoResponse res1, out EchoResponse res2);
rpc EchoWithMultiArgs(MultiRequests) returns (MultiResponses);
}
```
原先的echo.idl文件：
```idl
struct EchoRequest {
string message;
};
struct EchoResponse {
string message;
};
service EchoService {
void Echo(EchoRequest req, out EchoResponse res);
uint32_t EchoWithMultiArgs(EchoRequest req1, EchoRequest req2, out EchoResponse res1, out EchoResponse res2);
};
```
2. 在COMAKE或BCLOUD中插入如下片段以使用代码生成插件。
注意--mcpack_out要和--cpp_out一致，你可以先设成--mcpack_out=.，执行comake2或bcloud后看错误信息中的--cpp_out的值，再把--mcpack_out设成一样的。
BCLOUD中要把`/public/mcpack2pb/protoc-gen-mcpack`替换成`/public/mcpack2pb/protoc-gen-mcpack**.forbcloud**`，并把ENV.WorkRoot()替换为WORKROOT的实际值。
```python
PROTOC(ENV.WorkRoot()+"/third-64/protobuf/bin/protoc")
PROTOFLAGS("--plugin=protoc-gen-mcpack=" + ENV.WorkRoot() + "/public/mcpack2pb/protoc-gen-mcpack --mcpack_out=.")
PROTOFLAGS('--proto_path=' + ENV.WorkRoot() + 'public/mcpack2pb/')
PROTOFLAGS('--proto_path=' + ENV.WorkRoot() + 'third-64/protobuf/include/')
```
3. 用channel发起访问。
idl不同于pb，允许有多个请求，我们先看只有一个请求的情况，和普通的pb访问基本上是一样的。
```c++
#include <baidu/rpc/channel.h>
#include "echo.pb.h"
...
baidu::rpc::Channel channel;
baidu::rpc::ChannelOptions opt;
opt.protocol = baidu::rpc::PROTOCOL_UBRPC_COMPACK; // or "ubrpc_compack";
if (channel.Init(..., &opt) != 0) {
LOG(ERROR) << "Fail to init channel";
return -1;
}
EchoService_Stub stub(&channel);
...
EchoRequest request;
EchoResponse response;
baidu::rpc::Controller cntl;
request.set_message("hello world");
stub.Echo(&cntl, &request, &response, NULL);
if (cntl.Failed()) {
LOG(ERROR) << "Fail to send request, " << cntl.ErrorText();
return;
}
// 取response中的字段
// [idl] void Echo(EchoRequest req, out EchoResponse res);
// ^
// response.message();
```
多个请求要设置一下set_idl_names。
```c++
#include <baidu/rpc/channel.h>
#include "echo.pb.h"
...
baidu::rpc::Channel channel;
baidu::rpc::ChannelOptions opt;
opt.protocol = baidu::rpc::PROTOCOL_UBRPC_COMPACK; // or "ubrpc_compack";
if (channel.Init(..., &opt) != 0) {
LOG(ERROR) << "Fail to init channel";
return -1;
}
EchoService_Stub stub(&channel);
...
MultiRequests multi_requests;
MultiResponses multi_responses;
baidu::rpc::Controller cntl;
multi_requests.mutable_req1()->set_message("hello");
multi_requests.mutable_req2()->set_message("world");
cntl.set_idl_names(baidu::rpc::idl_multi_req_multi_res);
stub.EchoWithMultiArgs(&cntl, &multi_requests, &multi_responses, NULL);
if (cntl.Failed()) {
LOG(ERROR) << "Fail to send request, " << cntl.ErrorText();
return;
}
// 取response中的字段
// [idl] uint32_t EchoWithMultiArgs(EchoRequest req1, EchoRequest req2,
// ^ out EchoResponse res1, out EchoResponse res2);
// | ^ ^
// | multi_responses.res1().message(); |
// | multi_responses.res2().message();
// cntl.idl_result();
```
例子详见[example/echo_c++_ubrpc_compack](https://svn.baidu.com/public/trunk/baidu-rpc/example/echo_c++_ubrpc_compack/)。
# ubrpc (by baidu-rpc-ub)
server端由public/ubrpc搭建，request/response使用idl文件描述字段，序列化格式是compack或mcpack_v2。
**步骤：**
1. 依赖[public/baidu-rpc-ub](https://svn.baidu.com/public/trunk/baidu-rpc-ub)模块，在COMAKE中增加依赖：`CONFIGS('public/baidu-rpc-ub@ci-base')。`这个模块是baidu-rpc的扩展，不需要的用户不会依赖idl/mcpack/compack等模块。baidu-rpc-ub只包含扩展代码，baidu-rpc中的新特性会自动体现在这个模块中。
2. 编写一个proto文件，其中定义了service，名字和idl中的相同，但请求类型必须是baidu.rpc.UBRequest，回复类型必须是baidu.rpc.UBResponse。这两个类型定义在baidu/rpc/ub.proto中，使用时得import。
```protobuf
import "baidu/rpc/ub.proto"; // UBRequest, UBResponse
option cc_generic_services = true;
// Define UB service. request/response must be UBRequest/UBResponse
service EchoService {
rpc Echo(baidu.rpc.UBRequest) returns (baidu.rpc.UBResponse);
};
```
3. 在COMAKE包含baidu-rpc-ub/src路径。
```python
# baidu/rpc/ub.proto的包含路径
PROTOC(ENV.WorkRoot()+"third-64/protobuf/bin/protoc")
PROTOFLAGS("--proto_path=" + ENV.WorkRoot() + "public/baidu-rpc-ub/src/")
```
4. 用法和访问其他协议类似：创建Channel，ChannelOptions.protocol为**baidu::rpc::PROTOCOL_NSHEAD_CLIENT**或**"nshead_client"**。request和response对象必须是baidu-rpc-ub提供的类型
```c++
#include <baidu/rpc/ub_call.h>
...
baidu::rpc::Channel channel;
baidu::rpc::ChannelOptions opt;
opt.protocol = baidu::rpc::PROTOCOL_NSHEAD_CLIENT; // or "nshead_client";
if (channel.Init(..., &opt) != 0) {
LOG(ERROR) << "Fail to init channel";
return -1;
}
EchoService_Stub stub(&channel);
...
const int BUFSIZE = 1024 * 1024; // 1M
char* buf_for_mempool = new char[BUFSIZE];
bsl::xmempool pool;
if (pool.create(buf_for_mempool, BUFSIZE) != 0) {
LOG(FATAL) << "Fail to create bsl::xmempool";
return -1;
}
// 构造UBRPC的request/response，idl结构体作为模块参数传入。为了构造idl结构，需要传入一个bsl::mempool
baidu::rpc::UBRPCCompackRequest<example::EchoService_Echo_params> request(&pool);
baidu::rpc::UBRPCCompackResponse<example::EchoService_Echo_response> response(&pool);
// 设置字段
request.mutable_req()->set_message("hello world");
// 发起RPC
baidu::rpc::Controller cntl;
stub.Echo(&cntl, &request, &response, NULL);
if (cntl.Failed()) {
LOG(ERROR) << "Fail to Echo, " << cntl.ErrorText();
return;
}
// 取回复中的字段
response.result_params().res().message();
...
```
具体example代码可以参考[echo_c++_compack_ubrpc](https://svn.baidu.com/public/trunk/baidu-rpc-ub/example/echo_c++_compack_ubrpc/)，类似的还有[echo_c++_mcpack_ubrpc](https://svn.baidu.com/public/trunk/baidu-rpc-ub/example/echo_c++_mcpack_ubrpc/)。
# nshead+idl
server端是由public/ub搭建，通讯包组成为nshead+idl::compack/idl::mcpack(v2)
由于不需要指定service和method，无需编写proto文件，直接使用Channel.CallMethod方法发起RPC即可。请求包中的nshead可以填也可以不填，框架会补上正确的magic_num和body_len字段：
```c++
#include <baidu/rpc/ub_call.h>
...
baidu::rpc::Channel channel;
baidu::rpc::ChannelOptions opt;
opt.protocol = baidu::rpc::PROTOCOL_NSHEAD_CLIENT; // or "nshead_client";
if (channel.Init(..., &opt) != 0) {
LOG(ERROR) << "Fail to init channel";
return -1;
}
...
// 构造UB的request/response，完全类似构造原先idl结构，传入一个bsl::mempool（变量pool）
// 将类型作为模板传入，之后在使用上可以直接使用对应idl结构的接口
baidu::rpc::UBCompackRequest<example::EchoRequest> request(&pool);
baidu::rpc::UBCompackResponse<example::EchoResponse> response(&pool);
// Set `message' field of `EchoRequest'
request.set_message("hello world");
// Set fields of the request nshead struct if needed
request.mutable_nshead()->version = 99;
baidu::rpc::Controller cntl;
channel.CallMethod(NULL, &cntl, &request, &response, NULL); // 假设channel已经通过之前所述方法Init成功
// Get `message' field of `EchoResponse'
response.message();
```
具体example代码可以参考[echo_c++_mcpack_ub](https://svn.baidu.com/public/trunk/baidu-rpc-ub/example/echo_c++_mcpack_ub/)，compack情况类似，不再赘述
# nshead+mcpack(非idl产生的)
server端是由public/ub搭建，通讯包组成为nshead+mcpack包，但不是idl编译器生成的，RPC前需要先构造RawBuffer将其传入，然后获取mc_pack_t并按之前手工填写mcpack的方式操作：
```c++
#include <baidu/rpc/ub_call.h>
...
baidu::rpc::Channel channel;
baidu::rpc::ChannelOptions opt;
opt.protocol = baidu::rpc::PROTOCOL_NSHEAD_CLIENT; // or "nshead_client";
if (channel.Init(..., &opt) != 0) {
LOG(ERROR) << "Fail to init channel";
return -1;
}
...
// 构造RawBuffer，一次RPC结束后RawBuffer可以复用，类似于bsl::mempool
const int BUFSIZE = 10 * 1024 * 1024;
baidu::rpc::RawBuffer req_buf(BUFSIZE);
baidu::rpc::RawBuffer res_buf(BUFSIZE);
// 传入RawBuffer来构造request和response
baidu::rpc::UBRawMcpackRequest request(&req_buf);
baidu::rpc::UBRawMcpackResponse response(&res_buf);
// Fetch mc_pack_t and fill in variables
mc_pack_t* req_pack = request.McpackHandle();
int ret = mc_pack_put_str(req_pack, "mystr", "hello world");
if (ret != 0) {
LOG(FATAL) << "Failed to put string into mcpack: "
<< mc_pack_perror((long)req_pack) << (void*)req_pack;
break;
}
// Set fields of the request nshead struct if needed
request.mutable_nshead()->version = 99;
baidu::rpc::Controller cntl;
channel.CallMethod(NULL, &cntl, &request, &response, NULL); // 假设channel已经通过之前所述方法Init成功
// Get response from response buffer
const mc_pack_t* res_pack = response.McpackHandle();
mc_pack_get_str(res_pack, "mystr");
```
具体example代码可以参考[echo_c++_raw_mcpack](https://svn.baidu.com/public/trunk/baidu-rpc-ub/example/echo_c++_raw_mcpack/)。
# nshead+blob
r32897后baidu-rpc直接支持用nshead+blob访问老server（而不用依赖baidu-rpc-ub）。example代码可以参考[nshead_extension_c++](https://svn.baidu.com/public/trunk/baidu-rpc/example/nshead_extension_c++/client.cpp)。
```c++
#include <baidu/rpc/nshead_message.h>
...
baidu::rpc::Channel;
baidu::rpc::ChannelOptions opt;
opt.protocol = baidu::rpc::PROTOCOL_NSHEAD; // or "nshead"
if (channel.Init(..., &opt) != 0) {
LOG(ERROR) << "Fail to init channel";
return -1;
}
...
baidu::rpc::NsheadMessage request;
baidu::rpc::NsheadMessage response;
// Append message to `request'
request.body.append("hello world");
// Set fields of the request nshead struct if needed
request.head.version = 99;
baidu::rpc::Controller cntl;
channel.CallMethod(NULL, &cntl, &request, &response, NULL);
if (cntl.Failed()) {
LOG(ERROR) << "Fail to access the server: " << cntl.ErrorText();
return -1;
}
// response.head and response.body contains nshead_t and blob respectively.
```
或者用户也可以使用baidu-rpc-ub中的UBRawBufferRequest和UBRawBufferResponse来访问。example代码可以参考[echo_c++_raw_buffer](https://svn.baidu.com/public/trunk/baidu-rpc-ub/example/echo_c++_raw_buffer/)。
```c++
baidu::rpc::Channel channel;
baidu::rpc::ChannelOptions opt;
opt.protocol = baidu::rpc::PROTOCOL_NSHEAD_CLIENT; // or "nshead_client"
if (channel.Init(..., &opt) != 0) {
LOG(ERROR) << "Fail to init channel";
return -1;
}
...
// 构造RawBuffer，一次RPC结束后RawBuffer可以复用，类似于bsl::mempool
const int BUFSIZE = 10 * 1024 * 1024;
baidu::rpc::RawBuffer req_buf(BUFSIZE);
baidu::rpc::RawBuffer res_buf(BUFSIZE);
// 传入RawBuffer来构造request和response
baidu::rpc::UBRawBufferRequest request(&req_buf);
baidu::rpc::UBRawBufferResponse response(&res_buf);
// Append message to `request'
request.append("hello world");
// Set fields of the request nshead struct if needed
request.mutable_nshead()->version = 99;
baidu::rpc::Controller cntl;
channel.CallMethod(NULL, &cntl, &request, &response, NULL); // 假设channel已经通过之前所述方法Init成功
// Process response. response.data() is the buffer, response.size() is the length.
```
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment