|
|
发表于 2022-10-22 11:41:24
|
显示全部楼层
自荐一手自己写的 rpc 框架，不算什么优秀作品，但我想对于从事linux后台开发的人员来说绝对值得一看的。
开源地址：
https://github.com/Gooddbird/tinyrpc
TinyRPC 就是一个基于c++11开发的rpc框架，他能帮助你非常简单的就搭建一个高性能 rpc 服务，而且代码量不多，非常适合学习。
TinyRPC 的特点如下：
- 最方便的异步rpc调用，无须任何回调函数。通过引入协程，用同步的写法就能实现异步RPC调用的性能。简言之就是调用rpc时协程自动让出cpu，让其他协程执行。当rpc调用返回数据时自动切换回来。
- 主从Reactor架构，底层使用 epoll 进行套接字管理。多线程支持。性能不是问题，目前http echo 单机qps可以达到10w左右
- 快速构建rpc服务，只需要一份简单配置文件以及几行代码即可。
- 支持简单的 HTTP 协议；也支持基于Protobuf 序列化的协议。
TinyRPC的开源地址如下：
https://github.com/Gooddbird/tinyrpc
有兴趣的可以给个star，respect！
下面简单介绍下tinyrpc框架的使用方法：
1. 概述
上次介绍了如何使用 TinyRPC 框架搭建 HTTP 协议的 RPC 服务。
这次将带来基于 Protobuf 的 RPC 服务搭建。实际上在现在的互联网大厂中，通常会使用这种 Protobuf 这种格式来实现 RPC ，而不是 HTTP。 HTTP 通常用来对外提供接口。
TinyRPC 框架是我目前使用 C++ 开发的一个基于协程的高性能RPC框架。目前差不多进入开发尾声了，基本的功能已经实现和测试过了。当然，最基本的是 RPC 功能，我将在本篇文章中介绍如何搭建 RPC 服务。
TinyRPC 的项目地址为：
GitHub - Gooddbird/tinyrpc: c++ rpc
如果你觉得项目不错，麻烦给个 Star 支持下，万分感谢。
此篇文章为应用篇，涉及的原理知识不多。更多关于 TinyRPC 框架的原理性文章请参考，我会持续更新：
2. 搭建 RPC 服务
2.1 TinyPB 协议
TinyRPC 框架自定义了一种基于 protobuf 的协议报文格式，即 TinyPB (Tiny protobuf protocal) 协议。这个协议的格式参考了陈硕的《Linux 多线程编程--使用 muduo 网络库》一节内容，并扩充了一些必要的字段。
TinyPB 单个协议包报文用 c++ 伪代码描述如下：
/*** min of package is: 1 + 4 + 4 + 4 + 4 + 4 + 4 + 1 = 26 bytes***/char start; // 代表报文的开始， 一般是 0x02int32_t pk_len {0}; // 整个包长度，单位 byteint32_t msg_req_len {0}; // msg_req 字符串长度std::string msg_req; // msg_req,标识一个 rpc 请求或响应。 一般来说 请求 和 响应使用同一个 msg_req.int32_t service_name_len {0}; // service_name 长度std::string service_full_name; // 完整的 rpc 方法名， 如 QueryService.query_nameint32_t err_code {0}; // 框架级错误代码. 0 代表调用正常，非 0 代表调用失败int32_t err_info_len {0}; // err_info 长度std::string err_info; // 详细错误信息， err_code 非0时会设置该字段值std::string pb_data; // 业务 protobuf 数据，由 google 的 protobuf 序列化后得到int32_t check_num {0}; // 包检验和，用于检验包数据是否有损坏char end; // 代表报文结束，一般是 0x03
最小的数据包报文长度为 26 字节。
注释信息已经很完整了。另外几个需要特殊说明的字段如下：
err_code: err_code 是框架级别的错误码，即代表调用 RPC 过程中发生的错误，如对端关闭、调用超时等。为0 代表此次 RPC 调用正常，即正常发送数据且接收到回包。非 0 值代表调用失败，此时会设置 err_info 为详细的错误信息。
service_full_name : 是指的调用的完整方法名。即 servicename.methodname。一般来说，我们需要提前注册一个 Service (这里的 Service 指的继承了google::protobuf::Service 的类)，而一个 Service 下包含多个方法。
pb_data：这是最重要的数据，存储的是结果 Protobuf 序列化后的数据。
TinyPB 协议报文中包含了多个 len 字段，这主要是为了用空间换时间，接收方在提前知道长度的情况下，更方便解码各个字段，从而提升了 decode 效率。
可以看到，这个协议非常简单，只是个基本格式。 如果要正式运用在生产环境，还需要自己扩充字段才可。
2.2 定义 Protobuf 文件
前面介绍的 TinyPB 协议只需要简单知道即可， TinyRPC 内部已经实现了编码和解码。
当然，定义一个 Protobuf 文件是必须的。
如 tinypb.proto
syntax = "proto3";option cc_generic_services = true;message queryAgeReq { int32 req_no = 1; int32 id = 2;}message queryAgeRes { int32 ret_code = 1; string res_info = 2; int32 req_no = 3; int32 id = 4; int32 age = 5;}message queryNameReq { int32 req_no = 1; int32 id = 2; int32 type = 3;}message queryNameRes { int32 ret_code = 1; string res_info = 2; int32 req_no = 3; int32 id = 4; string name = 5;}service QueryService { // rpc method name rpc query_name(queryNameReq) returns (queryNameRes); // rpc method name rpc query_age(queryAgeReq) returns (queryAgeRes);}
2.3 生成 pb 桩文件
没什么好说的，感谢 google 的 ProtoBuf。直接使用命令行生成即可：
protoc --cpp_out=./ tinypb.proto
执行完后可以发现当前路径下新增了两个文件：
tnypb.pb.cc tinypb.pb.h
这是 Protobuf 自动根据我们的proto 文件生成的 C++ 代码。这两个文件包含了一些 RPC 主要类，如: queryAgeReq、QueryService、QueryService_Stub。
2.4 实现 RPC 方法
从生成的代码可以看到，QueryService 是个抽象基类，因此必须要继承然后重写 "query_name" 和 "query_age" 这两个方法才行。这个时候仅仅需要在这两个方法里面实现业务逻辑就行了。例如：
class QueryServiceImpl : public QueryService { public: QueryServiceImpl() {} ~QueryServiceImpl() {} void query_name(google::protobuf::RpcController* controller, const ::queryNameReq* request, ::queryNameRes* response, ::google::protobuf::Closure* done) { // DebugLog << "========================"; // DebugLog << "this is query_name func"; // DebugLog << "first begin to sleep 6s"; // sleep_hook(6); // DebugLog << "sleep 6s end"; response->set_ret_code(0); response->set_res_info("OK"); tinyrpc::MySQLInstase* instase = tinyrpc::MySQLInstaseFactroy::GetThreadMySQLFactory()->GetMySQLInstase("test_db_key1"); if (!instase->isInitSuccess()) { response->set_ret_code(-1); response->set_res_info("faild to init mysql"); ErrorLog << "mysql instase init failed"; return; } char query_sql[512]; sprintf(query_sql, "select user_id, user_name, user_gender from user_db.t_user_information where user_id = '%s';", std::to_string(request->id()).c_str()); int rt = instase->query(std::string(query_sql)); if (rt != 0) { response->set_ret_code(-1); response->set_res_info(instase->getMySQLErrorInfo()); ErrorLog << "failed to excute sql" << std::string(query_sql); return; } MYSQL_RES* res = instase->storeResult(); MYSQL_ROW row = instase->fetchRow(res); if (row) { int i = 0; DebugLog << "query success"; response->set_id(std::atoi(row[i++])); response->set_name(std::string(row[i++])); } else { DebugLog << "query empty"; response->set_ret_code(-1); response->set_res_info("this user not exist"); } if (done) { done->Run(); } } void query_age(google::protobuf::RpcController* controller, const ::queryAgeReq* request, ::queryAgeRes* response, ::google::protobuf::Closure* done) { DebugLog << "========================"; DebugLog << "this is query_age func"; response->set_ret_code(0); response->set_res_info("OK"); response->set_req_no(request->req_no()); response->set_id(request->id()); response->set_age(20); DebugLog << "========================"; done->Run(); }};
这里很简单，就是根据 request 的 id 参数去 MySQL 数据库里面查询用户信息，然后返回到 response 对象。
2.5 配置文件
跟前一篇文章一样的，一个配置文件是必须的。
<?xml version="1.0" encoding="UTF-8" ?><root> <!--log config--> <log> <!--日志文件路径，这个是相对运行时候 server 所在文件的路径--> <log_path>./</log_path> <!--日志文件前缀，一般跟 服务的名字相同--> <log_prefix>test_rpc_server1</log_prefix> <!--identify max size of single log file, MB--> <log_max_file_size>5</log_max_file_size> <!--log level: DEBUG < INFO < WARN < ERROR--> <log_level>DEBUG</log_level> <!--inteval that put log info to async logger, s--> <log_sync_inteval>1</log_sync_inteval> </log> <coroutine> <!--coroutine stack size (KB)--> <coroutine_stack_size>128</coroutine_stack_size> <!--default coroutine pool size--> <coroutine_pool_size>5000</coroutine_pool_size> </coroutine> <msg_req_len>20</msg_req_len> <!--max time when call connect, s--> <max_connect_timeout>75</max_connect_timeout> <!--count of io threads, at least 1--> <iothread_num>2</iothread_num> <time_wheel> <bucket_num>6</bucket_num> <!--inteval that destroy bad TcpConnection, s--> <inteval>10</inteval> </time_wheel> <server> <!--服务器绑定的 ip--> <ip>192.168.245.7</ip> <!--服务器监听的端口--> <port>39999</port> <!--服务器协议--> <protocal>TinyPB</protocal> </server> <database> <!--这里需要换成自己的 MySQL 配置信息--> <db_key name="test_db_key1"> <!-- <ip>127.0.0.1</ip> --> <ip>192.168.245.7</ip> <port>3306</port> <user>root</user> <passwd>Ikerli20220517!!</passwd> <select_db></select_db> <char_set>utf8mb4</char_set> </db_key> </database></root>
注意这里 protocal 选择的是 TinyPB 协议，而不是 HTTP。 说明这个 RPC 服务的协议类型是 TinyPB。
另外要注意下 database 节点，因为这里使用了 MySQL, 需要将其配置信息写入到此文件中。
2.6 实现 RPC 服务
没什么好说的，几行代码即可：
int main(int argc, char* argv[]) { if (argc != 2) { printf("Start TinyRPC server error, input argc is not 2!"); printf("Start TinyRPC server like this: \n"); printf("./server a.xml\n"); return 0; } tinyrpc::InitConfig(argv[1]); tinyrpc::GetServer()->registerService(std::make_shared<QueryServiceImpl>()); tinyrpc::StartRpcServer(); return 0;}
注意这里需要将 QueryServiceImpl 注册到 RPC 中，否则无法调用这个方法。
然后编译启动即可：
nohup ./test_rpc_server1 ../conf/test_rpc_server1.xml &
此时，就成功建立了 一个基于 TinyPB 协议的 RPC 服务了。那么怎么测试呢，还记得上一节的 HTTP 服务吗，修改下代码，让 HTTP 服务调用这个 RPC 服务。
2.7 修改 HTTP 服务实现异步 RPC 调用
void handle(tinyrpc::HttpRequest* req, tinyrpc::HttpResponse* res) { DebugLog << "success recive http request, now to get http response"; setHttpCode(res, tinyrpc::HTTP_OK); setHttpContentType(res, "text/html;charset=utf-8"); queryNameReq rpc_req; queryNameRes rpc_res; DebugLog << "now to call QueryServer TinyRPC server to query who's id is " << req->m_query_maps["id"]; rpc_req.set_id(std::atoi(req->m_query_maps["id"].c_str())); tinyrpc::TinyPbRpcChannel channel(std::make_shared<tinyrpc::IPAddress>("127.0.0.1", 39999)); QueryService_Stub stub(&channel); tinyrpc::TinyPbRpcController rpc_controller; rpc_controller.SetTimeout(5000); stub.query_name(&rpc_controller, &rpc_req, &rpc_res, NULL); if (rpc_controller.ErrorCode() != 0) { ErrorLog << "failed to call QueryServer rpc server"; char buf[512]; sprintf(buf, html, "failed to call QueryServer rpc server"); setHttpBody(res, std::string(buf)); return; } if (rpc_res.ret_code() != 0) { std::stringstream ss; ss << "QueryServer rpc server return bad result, ret = " << rpc_res.ret_code() << ", and res_info = " << rpc_res.res_info(); ErrorLog << ss.str(); char buf[512]; sprintf(buf, html, ss.str().c_str()); setHttpBody(res, std::string(buf)); return; } std::stringstream ss; ss << "Success!! Your name is " << rpc_res.name() << ", and Your id is " << rpc_res.id(); char buf[512]; sprintf(buf, html, ss.str().c_str()); setHttpBody(res, std::string(buf)); }
这一段代码，几乎算是 RPC 调用的标准模板了。在 TinyRPC 框架中，调用 RPC 服务就仅仅只需要这几行代码即可。真正核心的调用原理都被封装在 RpcChannel 里面了。
此外，这里调用 RPC 是异步的，而不是传统的阻塞式的调用。 为什么说是异步的，其实在：
stub.query_name(&rpc_controller, &rpc_req, &rpc_res, NULL);
这一行代码中，由于 RPC 调用肯定是需要耗时的，这一段代码不可能立即返回。但是由于协程的特性，TinyRPC 框架在内部封装了这些细节，会主动 yield 当前协程，释放 CPU 资源，线程转而切换到其他协程去执行。而当调用完成后，会自动 Resume 当前协程，然后 query_name 函数返回，继续往下执行。
也就是说，这里同步的写法确实现了异步的性能。这也是 TinyRPC 框架的最大亮点。
想一下，如果是以前，要做到异步 RPC 我们要怎么做，怕不是只能用回调函数，类似这样：
std::function call_back = [](){ // .......... // .......... }stub.query_name(&rpc_controller, &rpc_req, &rpc_res, call_back);
我们要把 query_name 返回后所有的剩下代码写到回调函数 call_back 里面去，这代码看上去确实挺难受的。毕竟谁都不想看这种异步代码，不容易看出执行顺序。此外，有可能回调函数又嵌套回调函数，最终陷入回调地狱，代码支离破碎，难以理解。
这里只是提一嘴，关于异步 RPC 的细节原理，将在下一篇文章介绍。
注意，这里的调用链路。从 test_http_sever 调用了 test_rpc_server1 服务的 query_name RPC 方法，tets_rpc_srver1 的 query_name 方法从 MySQL 中查询到用户数据并且返回给 test_http_server。调用链路如图：
2.8 简单测试
ok，先将两个 RPC 服务运行起来，然后同样访问 url
http://192.168.245.7:19999/user?id=1100110001
同样能得到正确的测试结果，读者可自行测试。本文代码和文档位于：
tinyrpc/quick_rpc_test.md at main · Gooddbird/tinyrpc
欢迎自行测试，有问题请提 issue。
TinyRPC的开源地址如下：
https://github.com/Gooddbird/tinyrpc
有兴趣的可以给个star，respect！
<hr>如果您觉得对你有所帮助，请点个赞加关注支持一下。同时 GitHub 项目也麻烦给个 Star 支持下开源精神，感谢各位大佬！！！
更多内容欢迎关注我的个人公众号，微信搜索 ikerli |
|
