gRPC异步使用入门(C++)
2016-12-14 11:27
1791 查看
gRPC 1.0的正式发布,正好赶上我们新项目的开始。出于Google的招牌以及“1.0”所代表的信心,在阅读了其特性列表,确定能够满足项目需求的情况下,我们哼哧哼哧的用上了。
在gRPC之前,我在实际项目中大规模使用的是ZeroC出品的ICE,那是一个功能非常丰富、文档和工具也非常完备的RPC框架。不过一方面其是商业产品,虽然源代码开放,但是用于商用需要支付一笔不菲的费用;另一方面,由于功能特性很多,显得有些过于重量级,部分常用功能的学习成本相对较高。gRPC不存在这两个问题,同时由于后发优势,在部分功能细节上能够提供更多的便利。比如gRPC的每个消息都可以通过DebugString方法文本化,在日志记录的时候就非常方便。
关于gRPC的入门和使用,参考其官方文档http://www.grpc.io/docs/,网络上还有其中文翻译版本http://doc.oschina.net/grpc?t=58008。这篇文章要分享的,是gRPC异步的正确使用方式。
然而这个例子仅仅是告诉了我们,异步模式下请求一个服务接口所使用的gRPC函数与同步模式下的区别,并没有实现一般情况下我们使用异步模式的目的。
为什么这样说呢?我们阅读这段代码,首先可以得到一个结论,对于SayHello函数的调用者来说,与使用同步模式版本的SayHello函数没有区别。SayHello仍然需要等到请求被传送到服务端,服务端处理请求并返回响应后才能够返回。而一般的情况下,我们如果使用异步模式,是希望SayHello函数能够在调用发送请求的指令后立即返回,在得到响应后异步处理的。
实际上,gRPC示例HelloWorld中的另一个文件greeter_async_client2.cc中,展示了如何实现一般情况下的异步客户端。
其实我也是在写这篇文章,需要从greeter_async_client.cc中拷贝代码时,才注意到这个greeter_async_client2.cc。Google其实应该在文档中再多花一点点笔墨,照顾到那些异步网络通信编程经验并不是特别丰富的人群。后者至少要提示greeter_async_client.cc仅仅展示了gRPC的异步调用API,完整的异步客户端应该阅读greeter_async_client2.cc。
我本意是打算将自己的异步客户端代码拷贝一部分到文章里,但是既然发现了这greeter_async_client2.cc,不妨就使用它来说明gRPC异步客户端的正确使用姿势。先上代码:
在greeter_async_client2.cc中,使用了GreeterClient类来封装SayHello的请求及处理响应。与greeter_aync_client.cc中不同,GreeterClient类在一个新的线程中异步的处理服务端发回的响应。调用者调用SayHello函数时,在Finish函数被调用之后即刻返回了。而在新的线程中,处理函数AsyncCompleteRpc循环从完成队列中取出服务端响应,并做处理。从这样一段简单的代码中,我们可以总结出以下几个关键信息:
gRPC的异步依托于CompletionQueue完成队列(消息队列)来实现。
在异步客户端中,通过gRPC stub的异步方法调用,获取ClientAsyncResponseReader的实例。
在异步客户端中,ClientAsyncResponseReader的Finish方法向CompletionQueue注册了响应消息处理器和响应消息体的存储容器。
当服务器响应消息到来时,响应消息体被填充到注册的容器中,而响应消息处理器则被push到CompletionQueue中。
从CompletionQueue中获取到响应消息处理器,对响应消息进行处理。
弄明白这么几点之后,我们不难写出适合所需的异步gRPC Client了。只需要做以下几步改进:
将AsyncClientCall提升为一个纯虚类,定义一个纯虚函数用于处理服务器消息。将ClientAsyncResponseReader和xxxReplay成员变量下沉到其派生类中。
从AsyncClientCall派生不同消息的处理器,根据消息不同,其成员变量response_reader和replay的类型不同,并实现AsyncClientCall定义的消息处理纯虚函数。
参照SayHello函数编写其它消息的调用函数。
在AsyncCompleteRpc方法中将打印消息调用状态的代码替换为调用AsyncClientCall的消息处理函数。
这样我们就利用C++的多态性在AsyncCompleteRpc中实现了对不同消息响应的处理。
这里想要讨论一下在服务端是否有必要在gRPC本身提供的CompletionQueue之外,再使用消息队列将消息处理器对其它内部模块的调用异步化。一般来说,这是不必要的。因为gRPC本身提供了异步机制,已经可以实现对请求的异步处理了。
另外一个问题,是否可以利用多线程并行处理消息请求?答案当然是肯定的,我们可以使用不同的CompletionQueue在不同线程内并行处理不同消息。至于有没有必要,以及能否提高效率和处理随之引入的内部模块锁竞争问题,就需要具体情况具体分析了。
在gRPC之前,我在实际项目中大规模使用的是ZeroC出品的ICE,那是一个功能非常丰富、文档和工具也非常完备的RPC框架。不过一方面其是商业产品,虽然源代码开放,但是用于商用需要支付一笔不菲的费用;另一方面,由于功能特性很多,显得有些过于重量级,部分常用功能的学习成本相对较高。gRPC不存在这两个问题,同时由于后发优势,在部分功能细节上能够提供更多的便利。比如gRPC的每个消息都可以通过DebugString方法文本化,在日志记录的时候就非常方便。
关于gRPC的入门和使用,参考其官方文档http://www.grpc.io/docs/,网络上还有其中文翻译版本http://doc.oschina.net/grpc?t=58008。这篇文章要分享的,是gRPC异步的正确使用方式。
客户端
grpc官方文档和示例HelloWorld的greeter_async_client.cc中,关于客户端异步的示例代码是这样的:std::string SayHello(const std::string& user) { // Data we are sending to the server. HelloRequest request; request.set_name(user); // Container for the data we expect from the server. HelloReply reply; // Context for the client. It could be used to convey extra information to // the server and/or tweak certain RPC behaviors. ClientContext context; // The producer-consumer queue we use to communicate asynchronously with the // gRPC runtime. CompletionQueue cq; // Storage for the status of the RPC upon completion. Status status; // stub_->AsyncSayHello() performs the RPC call, returning an instance we // store in "rpc". Because we are using the asynchronous API, we need to // hold on to the "rpc" instance in order to get updates on the ongoing RPC. std::unique_ptr<ClientAsyncResponseReader<HelloReply> > rpc( stub_->AsyncSayHello(&context, request, &cq)); // Request that, upon completion of the RPC, "reply" be updated with the // server's response; "status" with the indication of whether the operation // was successful. Tag the request with the integer 1. rpc->Finish(&reply, &status, (void*)1); void* got_tag; bool ok = false; // Block until the next result is available in the completion queue "cq". // The return value of Next should always be checked. This return value // tells us whether there is any kind of event or the cq_ is shutting down. GPR_ASSERT(cq.Next(&got_tag, &ok)); // Verify that the result from "cq" corresponds, by its tag, our previous // request. GPR_ASSERT(got_tag == (void*)1); // ... and that the request was completed successfully. Note that "ok" // corresponds solely to the request for updates introduced by Finish(). GPR_ASSERT(ok); // Act upon the status of the actual RPC. if (status.ok()) { return reply.message(); } else { return "RPC failed"; } }
然而这个例子仅仅是告诉了我们,异步模式下请求一个服务接口所使用的gRPC函数与同步模式下的区别,并没有实现一般情况下我们使用异步模式的目的。
为什么这样说呢?我们阅读这段代码,首先可以得到一个结论,对于SayHello函数的调用者来说,与使用同步模式版本的SayHello函数没有区别。SayHello仍然需要等到请求被传送到服务端,服务端处理请求并返回响应后才能够返回。而一般的情况下,我们如果使用异步模式,是希望SayHello函数能够在调用发送请求的指令后立即返回,在得到响应后异步处理的。
实际上,gRPC示例HelloWorld中的另一个文件greeter_async_client2.cc中,展示了如何实现一般情况下的异步客户端。
其实我也是在写这篇文章,需要从greeter_async_client.cc中拷贝代码时,才注意到这个greeter_async_client2.cc。Google其实应该在文档中再多花一点点笔墨,照顾到那些异步网络通信编程经验并不是特别丰富的人群。后者至少要提示greeter_async_client.cc仅仅展示了gRPC的异步调用API,完整的异步客户端应该阅读greeter_async_client2.cc。
我本意是打算将自己的异步客户端代码拷贝一部分到文章里,但是既然发现了这greeter_async_client2.cc,不妨就使用它来说明gRPC异步客户端的正确使用姿势。先上代码:
class GreeterClient { public: explicit GreeterClient(std::shared_ptr<Channel> channel) : stub_(Greeter::NewStub(channel)) {} // Assembles the client's payload and sends it to the server. void SayHello(const std::string& user) { // Data we are sending to the server. HelloRequest request; request.set_name(user); // Call object to store rpc data AsyncClientCall* call = new AsyncClientCall; // stub_->AsyncSayHello() performs the RPC call, returning an instance to // store in "call". Because we are using the asynchronous API, we need to // hold on to the "call" instance in order to get updates on the ongoing RPC. call->response_reader = stub_->AsyncSayHello(&call->context, request, &cq_); // Request that, upon completion of the RPC, "reply" be updated with the // server's response; "status" with the indication of whether the operation // was successful. Tag the request with the memory address o 4000 f the call object. call->response_reader->Finish(&call->reply, &call->status, (void*)call); } // Loop while listening for completed responses. // Prints out the response from the server. void AsyncCompleteRpc() { void* got_tag; bool ok = false; // Block until the next result is available in the completion queue "cq". while (cq_.Next(&got_tag, &ok)) { // The tag in this example is the memory location of the call object AsyncClientCall* call = static_cast<AsyncClientCall*>(got_tag); // Verify that the request was completed successfully. Note that "ok" // corresponds solely to the request for updates introduced by Finish(). GPR_ASSERT(ok); if (call->status.ok()) std::cout << "Greeter received: " << call->reply.message() << std::endl; else std::cout << "RPC failed" << std::endl; // Once we're complete, deallocate the call object. delete call; } } private: // struct for keeping state and data information struct AsyncClientCall { // Container for the data we expect from the server. HelloReply reply; // Context for the client. It could be used to convey extra information to // the server and/or tweak certain RPC behaviors. ClientContext context; // Storage for the status of the RPC upon completion. Status status; std::unique_ptr<ClientAsyncResponseReader<HelloReply>> response_reader; }; // Out of the passed in Channel comes the stub, stored here, our view of the // server's exposed services. std::unique_ptr<Greeter::Stub> stub_; // The producer-consumer queue we use to communicate asynchronously with the // gRPC runtime. CompletionQueue cq_; }; int main(int argc, char** argv) { // Instantiate the client. It requires a channel, out of which the actual RPCs // are created. This channel models a connection to an endpoint (in this case, // localhost at port 50051). We indicate that the channel isn't authenticated // (use of InsecureChannelCredentials()). GreeterClient greeter(grpc::CreateChannel( "localhost:50051", grpc::InsecureChannelCredentials())); // Spawn reader thread that loops indefinitely std::thread thread_ = std::thread(&GreeterClient::AsyncCompleteRpc, &greeter); for (int i = 0; i < 100; i++) { std::string user("world " + std::to_string(i)); greeter.SayHello(user); // The actual RPC call! } std::cout << "Press control-c to quit" << std::endl << std::endl; thread_.join(); //blocks forever return 0; }
在greeter_async_client2.cc中,使用了GreeterClient类来封装SayHello的请求及处理响应。与greeter_aync_client.cc中不同,GreeterClient类在一个新的线程中异步的处理服务端发回的响应。调用者调用SayHello函数时,在Finish函数被调用之后即刻返回了。而在新的线程中,处理函数AsyncCompleteRpc循环从完成队列中取出服务端响应,并做处理。从这样一段简单的代码中,我们可以总结出以下几个关键信息:
gRPC的异步依托于CompletionQueue完成队列(消息队列)来实现。
在异步客户端中,通过gRPC stub的异步方法调用,获取ClientAsyncResponseReader的实例。
在异步客户端中,ClientAsyncResponseReader的Finish方法向CompletionQueue注册了响应消息处理器和响应消息体的存储容器。
当服务器响应消息到来时,响应消息体被填充到注册的容器中,而响应消息处理器则被push到CompletionQueue中。
从CompletionQueue中获取到响应消息处理器,对响应消息进行处理。
弄明白这么几点之后,我们不难写出适合所需的异步gRPC Client了。只需要做以下几步改进:
将AsyncClientCall提升为一个纯虚类,定义一个纯虚函数用于处理服务器消息。将ClientAsyncResponseReader和xxxReplay成员变量下沉到其派生类中。
从AsyncClientCall派生不同消息的处理器,根据消息不同,其成员变量response_reader和replay的类型不同,并实现AsyncClientCall定义的消息处理纯虚函数。
参照SayHello函数编写其它消息的调用函数。
在AsyncCompleteRpc方法中将打印消息调用状态的代码替换为调用AsyncClientCall的消息处理函数。
这样我们就利用C++的多态性在AsyncCompleteRpc中实现了对不同消息响应的处理。
服务端
gRPC服务端异步与客户端的原理类似,并且参考其官方文档和示例代码,跟客户端扩展处理不同消息的做法一样,就可以实现处理不同消息的异步服务端。这里想要讨论一下在服务端是否有必要在gRPC本身提供的CompletionQueue之外,再使用消息队列将消息处理器对其它内部模块的调用异步化。一般来说,这是不必要的。因为gRPC本身提供了异步机制,已经可以实现对请求的异步处理了。
另外一个问题,是否可以利用多线程并行处理消息请求?答案当然是肯定的,我们可以使用不同的CompletionQueue在不同线程内并行处理不同消息。至于有没有必要,以及能否提高效率和处理随之引入的内部模块锁竞争问题,就需要具体情况具体分析了。
相关文章推荐
- Lex和Yacc从入门到精通(4)-能够使用C++的Lex和Yacc框架
- 学习ajax很好的入门教程--使用 JavaScript 和 Ajax 发出异步请求
- C++ STL入门教程(2)——list(双向链表)的使用(附完整程序代码)
- C++ STL入门教程(4)——stack,queue,priority_queue的使用(附完整程序代码)
- C++ STL入门教程(3)——deque(双向队列)的使用(附完整程序代码)
- Lex和Yacc从入门到精通(4)-能够使用C++的Lex和Yacc框架
- Google C++测试框架系列入门篇:第一章 介绍:为什么使用GTest?
- C++ STL使用入门
- 快速入门:触摸输入(使用 C#/VB/C++ 和 XAML 的 Windows 应用商店应用)
- C++ STL入门教程(6)——set(集合)的使用(附完整程序代码)
- C/C++代码覆盖率工具:gcov和lcov的使用入门
- 【windows8开发】异步编程入门篇之 Concurrency::create_async(C++)
- 【windows8开发】异步编程入门篇之 Concurrency::task(C++)
- C++ STL入门教程(1)——vector(向量容器)的使用(附完整程序代码)
- C/C++代码覆盖率工具:gcov和lcov的使用入门 .
- jQuery入门学习二:使用jQuery方式实现异步登录验证
- Protocol Buffers的安装使用和C++入门示例
- Google V8 编程入门(二) - 使用c++访问js脚本对象
- C++ STL入门教程(4)——stack(栈),queue(队列),priority_queue(优先队列)的使用(附完整程序代码)
- C++ STL入门教程(7)——multimap(一对多索引),multiset(多元集合)的使用(附完整程序代码)