【gRPC】C++异步服务端客户端API实例及代码解析

2022/9/7 14:23:22

本文主要是介绍【gRPC】C++异步服务端客户端API实例及代码解析,对大家解决编程问题具有一定的参考价值,需要的程序猿们随着小编来一起学习吧!

对于同步API而言,程序的吞吐量并不高。因为在每次发送一个gRPC请求时,会阻塞整个线程,必须等待服务端的ack回到客户端才能继续运行或者发送下一个请求,因此异步API是提升程序吞吐量的必要手段。

gRPC异步操作依赖于完成队列CompletionQueue
官网教程:https://grpc.io/docs/languages/cpp/async/
参考博客1:https://www.luozhiyun.com/archives/671
参考博客2:https://blog.miigon.net/posts/cn-so-difference-between-sync-and-async-grpc/

整体思路概述:

  • 将一个完成队列CompletionQueue绑定到RPC调用
  • 在客户端与服务器两端执行写入或读取之类的操作,同时带有唯一的void*标签
  • 调用ComletionQueue::Next以等待操作完成。如果出现标签,则表示相应的操作完成

异步客户端:

要点

  • 就像同步客户端一样,我们需要创建一个存根stub用于进行gRPC方法的调用,但是此时我们需要调用的是异步方法,需要为其绑定一个完成队列
//客户端的类实例在初始化时就需要创建Channel和Stub了,具体看官方实例代码中的greeter_async_client.cc
CompletionQueue cq;//创建完成队列
std::unique_ptr<ClientAsyncResponseReader<HelloReply> > rpc(
    stub_->AsyncSayHello(&context, request, &cq));//将完成队列绑定到存根,进而创建出客户端异步响应读取器
  • rpc->StartCall()初始化RPC请求
  • rpc->Finish()有三个参数,分别是gRPC响应、最终状态、唯一标签。一旦RPC请求完成,响应的结果、最终状态会被封装到前两个传入的参数中,同时会附带唯一标签。唯一标签可以使用RPC请求的地址,这样一来,当最终的响应到来时,我们可以拿着该地址进行相应处理。
  • 异步处理响应,拿着唯一标签可以获取对应RPC请求的地址,进而进行相关处理。
  • 为了使得请求的发出与响应的处理相互之间不阻塞,需要把响应的处理独立出一个线程

代码示例

class GreeterClient {
 public:
  explicit GreeterClient(std::shared_ptr<Channel> channel)
      : stub_(Greeter::NewStub(channel)) {}

  // 客户端SayHello
  void SayHello(const std::string& user) {
    // RPC请求数据封装
    HelloRequest request;
    request.set_name(user);

    // 异步客户端请求,存储请求响应的状态和数据的结构体等,在下方进行的定义
    AsyncClientCall* call = new AsyncClientCall;

    // 初始化response_reader
    // stub_->PrepareAsyncSayHello()创建一个RPC对象,但是不会立即启动RPC调用
    call->response_reader =
        stub_->PrepareAsyncSayHello(&call->context, request, &cq_);

    // StartCall()方法发起真正的RPC请求
    call->response_reader->StartCall();

    // Finish()方法前两个参数用于指定响应数据的存储位置,第三个参数指定了该次RPC异步请求的地址
    call->response_reader->Finish(&call->reply, &call->status, (void*)call);
  }

  // 不断循环监听完成队列,对响应进行处理
  void AsyncCompleteRpc() {
    void* got_tag;
    bool ok = false;

    // 在队列为空时阻塞,队列中有响应结果时读取到got_tag和ok两个参数
    // 前者是结果对应的RPC请求的地址,后者是响应的状态
    while (cq_.Next(&got_tag, &ok)) {
      // 类型转换,获取到的实际上是此响应结果对应的RPC请求的地址,在这个地址下保存了实际的响应结果数据
      AsyncClientCall* call = static_cast<AsyncClientCall*>(got_tag);

      // 验证请求是否真的完成了
      GPR_ASSERT(ok);

      if (call->status.ok())
        std::cout << "Greeter received: " << call->reply.message() << std::endl;
      else
        std::cout << "RPC failed" << std::endl;

      // 完成了响应的处理后,清除该RPC请求
      delete call;
    }
  }

 private:
  // 异步客户端通话,存储一次RPC通话的信息,里面包含响应的状态和数据的结构
  struct AsyncClientCall {
    // 服务器返回的响应数据
    HelloReply reply;

    // 客户端的上下文信息,可以被用于向服务器传达额外信息或调整某些RPC行为
    ClientContext context;

    // RPC响应的状态
    Status status;

    // 客户端异步响应读取器
    std::unique_ptr<ClientAsyncResponseReader<HelloReply>> response_reader;
  };

  // 存根,在我们的视角里就是服务器端暴露的服务接口
  std::unique_ptr<Greeter::Stub> stub_;

  // 完成队列,一个用于gRPC异步处理的生产者消费者队列
  CompletionQueue cq_;
};

int main(int argc, char** argv) {
  // 实例化一个客户端,需要一个信道,第二个参数表明该通道未经过身份验证
  GreeterClient greeter(grpc::CreateChannel(
      "localhost:50051", grpc::InsecureChannelCredentials()));

  // 独立的异步响应处理线程
  // 由于该方法是客户端类内的非静态方法,所以需要传入客户端类的实例表明归属
  std::thread thread_ = std::thread(&GreeterClient::AsyncCompleteRpc, &greeter);

  //发送异步的请求SayHello()
  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();  //永远会阻塞,因为异步响应处理线程永远不会停止,必须ctrl+c才能退出

  return 0;
}

异步服务器

要点

  • 客户端在发起请求时附带了标签(此次RPC请求会话的地址),因此服务器端也需要将该标签妥善处理再返回
  • 官方API中是准备一个CallData对象作为容器,gRPC通过ServerCompletionQueue将各种事件发送到CallData对象中,然后让这儿对象根据自身的状态进行处理。处理完成后还需要再手动创建一个CallData对象,这个对象是为下一个Client请求准备的,整个过程就像流水线一样
    • CREATE:CallData对象被创建处理之前处于CREATE状态
    • PROCESS:请求到达后,转换为PROCESS状态
    • FINISH:响应完成后,转换为FINISH状态
  • 整体服务器端流程
    • 启动服务时,预分配 一个 CallData 实例供未来客户端请求使用。
    • 该 CallData 对象构造时,service_->RequestSayHello(&ctx_, &request_, &responder_, cq_, cq_, this) 将被调用,通知gRPC开始准备接收恰好是一个SayHello请求。
    • 这时候我们还不知道请求会由谁发出,何时到达,我们只是告诉 gRPC 说我们已经准备好接收了,让 gRPC 在真的接收到时通知我们。
    • 供给 RequestSayHello的参数告诉了gRPC将上下文信息、请求体以及回复器放在哪里、使用哪个完成队列来通知、以及通知的时候,用于鉴别请求的 tag(在这个例子中,this 被作为 tag 使用)。
      HandleRpcs() 运行到 cq->Next() 并阻塞。等待下一个事件发生
    • 客户端发送一个 SayHello 请求到服务器,gRPC 开始接收并解码该请求(IO 操作)
    • 一段时间后….
    • gRPC接收请求完成了。它将请求体放入CallData对象的request_成员中(通过我们之前提供的指针),然后创建一个事件(使用指向CallData 对象的指针作为 tag),并 将该事件放到完成队列 cq_ 中.
    • HandleRpcs() 中的循环接收到了该事件(之前阻塞住的 cq->Next() 调用此时也返回),并调用 CallData::Proceed() 来处理请求。
    • CallData 的 status_ 属性此时是 PROCESS,它做了如下事情:
      • 创建一个新的 CallData 对象,这样在这个请求后的新请求才能被新对象处理。
      • 生成当前请求的回复,告诉 gRPC 我们处理完成了,将该回复发送回客户端
      • gRPC 开始回复的传输 (IO 操作)
      • HandleRpcs() 中的循环迭代一次,再次阻塞在 cq->Next(),等待新事件的发生。
    • 一段时间后….
    • gRPC 完成了回复的传输,再次通过在完成队列里放入一个以 CallData 指针为 tag 的事件的方式通知我们。
    • cq->Next() 接收到该事件并返回,CallData::Proceed() 将 CallData 对象释放(使用 delete this;)。HandleRpcs() 循环并重新阻塞在 cq->Next() 上,等待新事件的发生。
      整个过程看似和同步 API 很相似,只是多了对完成队列的控制。然而,通过这种方式,每一个 一段时间后.... (通常是在等待 IO 操作的完成或等待一个请求出现) cq->Next() 不仅可以接收到当前处理的请求的完成事件,还可以接收到其他请求的事件。
      所以假设第一个请求正在等待它的回复数据传输完成时,一个新的请求到达了,cq->Next() 可以获得新请求产生的事件,并开始并行处理新请求,而不用等待第一个请求的传输完成

代码示例

//服务器实现类
class ServerImpl final {
 public:
  ~ServerImpl() {
    server_->Shutdown();
    // 关闭服务器后也要关闭完成队列
    cq_->Shutdown();
  }

  // There is no shutdown handling in this code.
  void Run() {
    // 服务器地址和端口
    std::string server_address("0.0.0.0:50051");
    // 服务器构建器
    ServerBuilder builder;
    // 服务器IP与端口指定,第二个参数表示该通道未经过身份验证
    builder.AddListeningPort(server_address, grpc::InsecureServerCredentials());
    // 注册服务
    builder.RegisterService(&service_);
    // 为当前服务器创建完成队列
    cq_ = builder.AddCompletionQueue();
    // 构建并启动服务器
    server_ = builder.BuildAndStart();
    std::cout << "Server listening on " << server_address << std::endl;

    // 运行服务器的主流程
    HandleRpcs();
  }

 private:
  // 处理一个请求所需要保存的状态、逻辑、数据被封装成了CallData类
  class CallData {
   public:
    // 传入service实例和服务器端的完成队列,创建后status_是CREATE状态
    CallData(Greeter::AsyncService* service, ServerCompletionQueue* cq)
        : service_(service), cq_(cq), responder_(&ctx_), status_(CREATE) {
      // 立即调用服务逻辑
      Proceed();
    }

    void Proceed() {
      if (status_ == CREATE) {
        // 转换为PROCESS状态
        status_ = PROCESS;

        // 作为初始化的一部分,请求系统开始处理SayHello请求。
        // 此处的this指代此CallData实例
        service_->RequestSayHello(&ctx_, &request_, &responder_, cq_, cq_,
                                  this);
      } else if (status_ == PROCESS) {
        // 在执行当前CallData的任务时,创建一个新的CallData实例去为新的请求服务
        // 当前CallData会在FINISH阶段自行销毁
        new CallData(service_, cq_);

        // 实际的逻辑处理
        std::string prefix("Hello ");
        reply_.set_message(prefix + request_.name());

        // 在完成后将状态置为FINISH。使用当前CallData的地址作为该事件的标签
        status_ = FINISH;
        responder_.Finish(reply_, Status::OK, this);
      } else {
        GPR_ASSERT(status_ == FINISH);
        // 销毁当前CallData
        delete this;
      }
    }

   private:
    // 异步服务
    Greeter::AsyncService* service_;
    //服务器端的完成队列,是一个生产者-消费者队列
    ServerCompletionQueue* cq_;
    // 服务器端上下文信息,可以被用于向客户端传达额外信息、数据或调整某些RPC行为
    ServerContext ctx_;

    // 客户端发来的请求
    HelloRequest request_;
    // 服务端的响应
    HelloReply reply_;

    // 发送服务端响应的工具
    ServerAsyncResponseWriter<HelloReply> responder_;

    // 状态机定义
    enum CallStatus { CREATE, PROCESS, FINISH };
    CallStatus status_;  // 当前的状态
  };

  // 如果有需求的话,服务器的处理可以是多线程的
  void HandleRpcs() {
    // 创建一个新的CallData,将完成队列中的数据封装进去
    new CallData(&service_, cq_.get());
    void* tag;  // 请求特有的标签,实际上是请求的RPC会话对象在客户端的地址信息
    bool ok;
    while (true) {
      // 阻塞等待读取完成队列中的事件,每个事件使用一个标签进行标识,该标签是CallData实例的地址
      // 完成队列的Next()方法应该每次都检查返回值,来确保事件和完成队列的状态正常
      GPR_ASSERT(cq_->Next(&tag, &ok));
      GPR_ASSERT(ok);
      // 从标签转换为CallData*类型,进而访问CallData中的方法
      static_cast<CallData*>(tag)->Proceed();
    }
  }

  // 当前服务器的完成队列
  std::unique_ptr<ServerCompletionQueue> cq_;
  // 当前服务器的异步服务
  Greeter::AsyncService service_;
  // 服务器实例
  std::unique_ptr<Server> server_;
};

int main(int argc, char** argv) {
  ServerImpl server;
  server.Run();
  return 0;
}



这篇关于【gRPC】C++异步服务端客户端API实例及代码解析的文章就介绍到这儿,希望我们推荐的文章对大家有所帮助,也希望大家多多支持为之网!


扫一扫关注最新编程教程