rpc.md

September 6, 2024 · View on GitHub

基础功能对比

RPC	IDL	通信	网络数据	压缩	Attachement	半同步	异步	Streaming
Thrift Binary Framed	Thrift	tcp	二进制	不支持	不支持	支持	不支持	不支持
Thrift Binary HttpTransport	Thrift	http	二进制	不支持	不支持	支持	不支持	不支持
GRPC	PB	http2	二进制	gzip/zlib/lz4/snappy	支持	不支持	支持	支持
BRPC Std	PB	tcp	二进制	gzip/zlib/lz4/snappy	支持	不支持	支持	支持
SRPC Std	PB/Thrift	tcp	二进制/JSON	gzip/zlib/lz4/snappy	支持	支持	支持	不支持
SRPC Std Http	PB/Thrift	http	二进制/JSON	gzip/zlib/lz4/snappy	支持	支持	支持	不支持

基础概念

通信层：TCP/TPC_SSL/HTTP/HTTPS/HTTP2
协议层：Thrift-binary/BRPC-std/SRPC-std/SRPC-http/tRPC-std/tRPC-http
压缩层：不压缩/gzip/zlib/lz4/snappy
数据层：PB binary/Thrift binary/Json string
IDL序列化层：PB/Thrift serialization
RPC调用层：Service/Client IMPL

RPC Global

获取srpc版本号srpc::SRPCGlobal::get_instance()->get_srpc_version()

RPC Status Code

name	value	含义
RPCStatusUndefined	0	未定义
RPCStatusOK	1	正确/成功
RPCStatusServiceNotFound	2	找不到Service名
RPCStatusMethodNotFound	3	找不到RPC函数名
RPCStatusMetaError	4	Meta错误/解析失败
RPCStatusReqCompressSizeInvalid	5	请求压缩大小错误
RPCStatusReqDecompressSizeInvalid	6	请求解压大小错误
RPCStatusReqCompressNotSupported	7	请求压缩类型不支持
RPCStatusReqDecompressNotSupported	8	请求解压类型不支持
RPCStatusReqCompressError	9	请求压缩失败
RPCStatusReqDecompressError	10	请求解压失败
RPCStatusReqSerializeError	11	请求IDL序列化失败
RPCStatusReqDeserializeError	12	请求IDL反序列化失败
RPCStatusRespCompressSizeInvalid	13	回复压缩大小错误
RPCStatusRespDecompressSizeInvalid	14	回复解压大小错误
RPCStatusRespCompressNotSupported	15	回复压缩类型不支持
RPCStatusRespDecompressNotSupported	16	回复解压类型不支持
RPCStatusRespCompressError	17	回复压缩失败
RPCStatusRespDecompressError	18	回复解压失败
RPCStatusRespSerializeError	19	回复IDL序列化失败
RPCStatusRespDeserializeError	20	回复IDL反序列化失败
RPCStatusIDLSerializeNotSupported	21	不支持IDL序列化
RPCStatusIDLDeserializeNotSupported	22	不支持IDL反序列化
RPCStatusURIInvalid	30	URI非法
RPCStatusUpstreamFailed	31	Upstream全熔断
RPCStatusSystemError	100	系统错误
RPCStatusSSLError	101	SSL错误
RPCStatusDNSError	102	DNS错误
RPCStatusProcessTerminated	103	程序退出&终止

RPC IDL

描述文件
前后兼容
Protobuf/Thrift

示例

下面我们通过一个具体例子来呈现

我们拿pb举例，定义一个ServiceName为Example的example.proto文件
rpc接口名为Echo，输入参数为EchoRequest，输出参数为EchoResponse
EchoRequest包括两个string：message和name
EchoResponse包括一个string：message

syntax="proto2";

message EchoRequest {
    optional string message = 1;
    optional string name = 2;
};

message EchoResponse {
    optional string message = 1;
};

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

RPC Service

组成SRPC服务的基本单元
每一个Service一定由某一种IDL生成
Service由IDL决定，与网络通信具体协议无关

示例

下面我们通过一个具体例子来呈现

沿用上面的example.protoIDL描述文件
执行官方的protoc example.proto --cpp_out=./ --proto_path=./获得example.pb.h和example.pb.cpp两个文件
执行SRPC的srpc_generator protobuf ./example.proto ./获得example.srpc.h文件
我们派生Example::Service来实现具体的rpc业务逻辑，这就是一个RPC Service
注意这个Service没有任何网络、端口、通信协议等概念，仅仅负责完成实现从EchoRequest输入到输出EchoResponse的业务逻辑

class ExampleServiceImpl : public Example::Service
{
public:
    void Echo(EchoRequest *request, EchoResponse *response, RPCContext *ctx) override
    {
        response->set_message("Hi, " + request->name());

        printf("get_req:\n%s\nset_resp:\n%s\n",
                request->DebugString().c_str(),
                response->DebugString().c_str());
    }
};

RPC Server

每一个Server对应一个端口
每一个Server对应一个确定的网络通信协议
一个Service可以添加到任意的Server里
一个Server可以拥有任意多个Service，但在当前Server里ServiceName必须唯一
不同IDL的Service是可以放进同一个Server中的

示例

下面我们通过一个具体例子来呈现

沿用上面的ExampleServiceImplService
首先，我们创建1个RPC Server，并确定proto文件的内容
然后，我们可以创建任意个数的Service实例、任意不同IDL proto形成的Service，把这些Service通过add_service()接口添加到Server里
最后，通过Server的start()或者serve()开启服务，处理即将到来的rpc请求
想像一下，我们也可以从Example::Service派生出多个Service，而它们的rpcEcho实现的功能可以不同
想像一下，我们可以在N个不同的端口创建N个不同的RPC Server，代表着不同的协议
想像一下，我们可以把同一个ServiceIMPL实例add_service()到不同的Server上，我们也可以把不同的ServiceIMPL实例add_service到同一个Server上
想像一下，我们可以用同一个ExampleServiceImpl，在三个不同端口、同时服务于BRPC-STD、SRPC-STD、SRPC-Http
甚至，我们可以将1个Protobuf IDL相关的ExampleServiceImpl和1个Thrift IDL相关的AnotherThriftServiceImpl，add_service到同一个SRPC-STD Server，两种IDL在同一个端口上完美工作！

int main()
{
    SRPCServer server_srpc;
    SRPCHttpServer server_srpc_http;
    BRPCServer server_brpc;
    ThriftServer server_thrift;
    TRPCServer server_trpc;
    TRPCHttpServer server_trpc_http;

    ExampleServiceImpl impl_pb;
    AnotherThriftServiceImpl impl_thrift;

    server_srpc.add_service(&impl_pb);
    server_srpc.add_service(&impl_thrift);
    server_srpc_http.add_service(&impl_pb);
    server_srpc_http.add_service(&impl_thrift);
    server_brpc.add_service(&impl_pb);
    server_thrift.add_service(&impl_thrift);
    server_trpc.add_service(&impl_pb);
    server_trpc_http.add_service(&impl_pb);

    server_srpc.start(1412);
    server_srpc_http.start(8811);
    server_brpc.start(2020);
    server_thrift.start(9090);
    server_trpc.start(2022);
    server_trpc_http.start(8822);

    getchar();
    server_trpc_http.stop();
    server_trpc.stop();
    server_thrift.stop();
    server_brpc.stop();
    server_srpc_http.stop();
    server_srpc.stop();

    return 0;
}

RPC Client

每一个Client对应着一个确定的目标/一个确定的集群
每一个Client对应着一个确定的网络通信协议
每一个Client对应着一个确定的IDL

示例

下面我们通过一个具体例子来呈现

沿用上面的例子，client相对简单，直接调用即可
通过Example::XXXClient创建某种RPC的client实例，需要目标的ip+port或url
利用client实例直接调用rpc函数Echo即可，这是一次异步请求，请求完成后会进入回调函数
具体的RPC Context用法请看下一个段落: RPC Context)

#include <stdio.h>
#include "example.srpc.h"
#include "workflow/WFFacilities.h"

using namespace srpc;

int main()
{
    Example::SRPCClient client("127.0.0.1", 1412);
    EchoRequest req;
    req.set_message("Hello!");
    req.set_name("SRPCClient");

    WFFacilities::WaitGroup wait_group(1);

    client.Echo(&req, [&wait_group](EchoResponse *response, RPCContext *ctx) {
        if (ctx->success())
            printf("%s\n", response->DebugString().c_str());
        else
            printf("status[%d] error[%d] errmsg:%s\n",
                    ctx->get_status_code(), ctx->get_error(), ctx->get_errmsg());
        wait_group.done();
    });

    wait_group.wait();
    return 0;
}

RPC Context

RPCContext专门用来辅助异步接口，Service和Client通用
每一个异步接口都会提供Context，用来给用户提供更高级的功能，比如获取对方ip、获取连接seqid等
Context上一些功能是Server或Client独有的，比如Server可以设置回复数据的压缩方式，Client可以获取请求成功或失败
Context上可以通过get_series()获得所在的series，与workflow的异步模式无缝结合

RPCContext API - Common

`long long get_seqid() const;`

请求+回复视为1次完整通信，获得当前socket连接上的通信sequence id，seqid=0代表第1次

`std::string get_remote_ip() const;`

获得对方IP地址，支持ipv4/ipv6

`int get_peer_addr(struct sockaddr addr, socklen_t addrlen) const;`

获得对方地址，in/out参数为更底层的数据结构sockaddr

`const std::string& get_service_name() const;`

获取RPC Service Name

`const std::string& get_method_name() const;`

获取RPC Methode Name

`SeriesWork *get_series() const;`

获取当前ServerTask/ClientTask所在series

`bool get_http_header(const std::string& name, std::string& value);`

如果通讯使用HTTP协议，则根据name获取HTTP header中的value

RPCContext API - Only for client done

`bool success() const;`

client专用。这次请求是否成功

`int get_status_code() const;`

client专用。这次请求的rpc status code

`const char *get_errmsg() const;`

client专用。这次请求的错误信息

`int get_error() const;`

client专用。这次请求的错误码

`void *get_user_data() const;`

client专用。获取ClientTask的user_data。如果用户通过create_xxx_task()接口产生task，则可以通过user_data域记录上下文，在创建task时设置，在回调函数中拿回。

RPCContext API - Only for server process

`void set_data_type(RPCDataType type);`

Server专用。设置数据打包类型

RPCDataProtobuf
RPCDataThrift
RPCDataJson

`void set_compress_type(RPCCompressType type);`

Server专用。设置数据压缩类型(注：Client的压缩类型在Client或Task上设置)

RPCCompressNone
RPCCompressSnappy
RPCCompressGzip
RPCCompressZlib
RPCCompressLz4

`void set_attachment_nocopy(const char *attachment, size_t len);`

Server专用。设置attachment附件。

`bool get_attachment(const char **attachment, size_t *len) const;`

Server专用。获取attachment附件。

`void set_reply_callback(std::function<void (RPCContext *ctx)> cb);`

Server专用。设置reply callback，操作系统写入socket缓冲区成功后被调用。

`void set_send_timeout(int timeout);`

Server专用。设置发送超时，单位毫秒。-1代表无限。

`void set_keep_alive(int timeout);`

Server专用。设置连接保活时间，单位毫秒。-1代表无限。

`bool set_http_code(int code);`

Server专用。如果通讯使用HTTP协议，则可以设置http status code返回码。仅在框架层能正确响应时有效。

`bool set_http_header(const std::string& name, const std::string& value);`

Server专用。如果通讯使用HTTP协议，可以在回复中设置HTTP header，如果name被设置过会覆盖旧value。

`bool add_http_header(const std::string& name, const std::string& value);`

Server专用。如果通讯使用HTTP协议，可以在回复中添加HTTP header，如果有重复name，会保留多个value。

`void log(const RPCLogVector& fields);`

Server专用。透传数据相关，请参考OpenTelemetry数据协议中的log语义。

`void baggage(const std::string& key, const std::string& value);`

Server专用。透传数据相关，参考OpenTelemetry数据协议中的baggage语义。

`void set_json_add_whitespace(bool on);`

Server专用。JsonPrintOptions相关，可设置增加json空格等。

`void set_json_always_print_enums_as_ints(bool flag);`

Server专用。JsonPrintOptions相关，可设置用int打印enum名。

`void set_json_preserve_proto_field_names(bool flag);`

Server专用。JsonPrintOptions相关，可设置保留原始字段名字。

`void set_json_always_print_fields_with_no_presence(bool flag);`

Server专用。JsonPrintOptions相关，可设置带上所有默认的proto数据中的域。

RPC Options

Server Params

name	默认	含义
max_connections	2000	Server的最大连接数，默认2000个
peer_response_timeout	10 * 1000	每一次IO的读超时，默认10秒
receive_timeout	-1	每一条完整消息的读超时，默认无限
keep_alive_timeout	60 * 1000	空闲连接保活，-1代表永远不断开，0代表短连接，默认长连接保活60秒
request_size_limit	2LL * 1024 * 1024 * 1024	请求包大小限制，最大2GB
ssl_accept_timeout	10 * 1000	SSL连接超时，默认10秒

Client Params

name	默认	含义
host	""	目标host，可以是ip、域名
port	1412	目标端口号，默认1412
is_ssl	false	ssl开关，默认关闭
url	""	当host为空，url设置才有效。url将屏蔽host/port/is_ssl三项
task_params	TASK默认配置	见下方

Task Params

name	默认	含义
send_timeout	-1	发送写超时，默认无限
receive_timeout	-1	回复超时，默认无限
watch_timeout	0	对方第一次回复的超时，默认0不设置
keep_alive_timeout	30 * 1000	空闲连接保活，-1代表永远不断开，默认30s
retry_max	0	最大重试次数，默认0不重试
compress_type	RPCCompressNone	压缩类型，默认不压缩
data_type	RPCDataUndefined	网络包数据类型，默认与RPC默认值一致，SRPC-Http协议为json，其余为对应IDL的类型

与workflow异步框架的结合

1. Server

下面我们通过一个具体例子来呈现

Echo RPC在接收到请求时，向下游发起一次http请求
对下游请求完成后，我们将http response的body信息填充到response的message里，回复给客户端
我们不希望阻塞/占据着Handler的线程，所以对下游的请求一定是一次异步请求
首先，我们通过Workflow框架的工厂WFTaskFactory::create_http_task创建一个异步任务http_task
然后，我们利用RPCContext的ctx->get_series()获取到ServerTask所在的SeriesWork
最后，我们使用SeriesWork的push_back接口将http_task放到SeriesWork的后面

class ExampleServiceImpl : public Example::Service
{
public:
    void Echo(EchoRequest *request, EchoResponse *response, RPCContext *ctx) override
    {
        auto *http_task = WFTaskFactory::create_http_task("https://www.sogou.com", 0, 0,
            [request, response](WFHttpTask *task) {
                if (task->get_state() == WFT_STATE_SUCCESS)
                {
                    const void *data;
                    size_t len;
                    task->get_resp()->get_parsed_body(&data, &len);
                    response->mutable_message()->assign((const char *)data, len);
                }
                else
                    response->set_message("Error: " + std::to_string(task->get_error()));

                printf("Server Echo()\nget_req:\n%s\nset_resp:\n%s\n",
                                            request->DebugString().c_str(),
                                            response->DebugString().c_str());
            });

        ctx->get_series()->push_back(http_task);
    }
};

2. Client

下面我们通过一个具体例子来呈现

我们并行发出两个请求，1个是rpc请求，1个是http请求
两个请求都结束后，我们再发起一次计算任务，计算两个数的平方和
首先，我们通过RPC Client的create_Echo_task创建一个rpc异步请求的网络任务rpc_task
然后，我们通过Workflow框架的工厂WFTaskFactory::create_http_task和WFTaskFactory::create_go_task分别创建异步网络任务http_task，和异步计算任务calc_task
最后，我们利用串并联流程图，乘号代表并行、大于号代表串行，将3个异步任务组合起来执行start()

void calc(int x, int y)
{
    int z = x * x + y * y;

    printf("calc result: %d\n", z);
}

int main()
{
    Example::SRPCClient client("127.0.0.1", 1412);

    auto *rpc_task = client.create_Echo_task([](EchoResponse *response, RPCContext *ctx) {
        if (ctx->success())
            printf("%s\n", response->DebugString().c_str());
        else
            printf("status[%d] error[%d] errmsg:%s\n",
                    ctx->get_status_code(), ctx->get_error(), ctx->get_errmsg());
    });

    auto *http_task = WFTaskFactory::create_http_task("https://www.sogou.com", 0, 0, [](WFHttpTask *task) {
        if (task->get_state() == WFT_STATE_SUCCESS)
        {
            std::string body;
            const void *data;
            size_t len;

            task->get_resp()->get_parsed_body(&data, &len);
            body.assign((const char *)data, len);
            printf("%s\n\n", body.c_str());
        }
        else
            printf("Http request fail\n\n");
    });

    auto *calc_task = WFTaskFactory::create_go_task(calc, 3, 4);

    EchoRequest req;
    req.set_message("Hello!");
    req.set_name("1412");
    rpc_task->serialize_input(&req);

    WFFacilities::WaitGroup wait_group(1);

    SeriesWork *series = Workflow::create_series_work(http_task, [&wait_group](const SeriesWork *) {
        wait_group.done();
    });
    series->push_back(rpc_task);
    series->push_back(calc_task);
    series->start();

    wait_group.wait();
    return 0;
}

3. Upstream

SRPC可以直接使用Workflow的任何组件，最常用的就是Upstream，SRPC的任何一种client都可以使用Upstream。

我们通过参数来看看如何构造可以使用Upstream的client：

#include "workflow/UpstreamManager.h"

int main()
{
    // 1. 创建upstream并添加实例
    UpstreamManager::upstream_create_weighted_random("echo_server", true);
    UpstreamManager::upstream_add_server("echo_server", "127.0.0.1:1412");
    UpstreamManager::upstream_add_server("echo_server", "192.168.10.10");
    UpstreamManager::upstream_add_server("echo_server", "internal.host.com");

    // 2. 构造参数，填上upstream的名字
    RPCClientParams client_params = RPC_CLIENT_PARAMS_DEFAULT;
    client_params.host = "srpc::echo_server"; // 这个scheme只用于upstream URI解析
    client_params.port = 1412; // 这个port只用于upstream URI解析，不影响具体实例的选取

    // 3. 用参数创建client，其他用法与示例类似
    Example::SRPCClient client(&client_params);

    ...

如果使用了ConsistentHash或者Manual方式创建upstream，则我们往往需要对不同的task进行区分、以供选取算法使用。这时候可以使用client task上的int set_uri_fragment(const std::string& fragment);接口，设置请求级相关的信息。

这个域的是URI里的fragment，语义请参考RFC3689 3.5-Fragment，任何需要用到fragment的功能（如其他选取策略里附带的其他信息），都可以利用这个域。