标签：CallData 请求 gRPC C++ 队列 API cq 客户端

对于同步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;
}

标签：CallData,请求,gRPC,C++,队列,API,cq,客户端
来源： https://www.cnblogs.com/jinyunshaobing/p/16663772.html

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。