GRPC 异步调用

文章目录

protobuf原生的异步调用

void DoneCallback(PingMessage *response) {
}
void async_test() {
  RpcClient client("127.0.0.1", 8000);
  PingService::Stub stub(client.Channel());
  if (!client.init()) {
    printf("can't connect\n");
  }
  PingMessage request;
  // get time
  gettimeofday(&aync_starttime, NULL);
  request.set_time(aync_starttime.tv_sec*1000+int(aync_starttime.tv_usec/1000));

  PingMessage* response = new PingMessage();
  Closure* callback = NewCallback(&DoneCallback, response);
  stub.ping(client.Controller(), &request, response, callback);
}

原生的protobuf提供的rpc框架是让我们提供一个回调,当有对应的响应时(发送的消息序列号与接收的相同),调用对应的回调即可;它生成的接口同步和异步是相同的;
与同步调用有相同的问题;由于它只能一个request和response,没有context和status,所以不能带额外的参数;也不能表示调用成功与否;

GRPC异步调用

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

  std::string SayHello(const std::string& user) {
    HelloRequest request;
    request.set_name(user);

    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异步接口向服务器发送请求
    std::unique_ptr<ClientAsyncResponseReader<HelloReply> > rpc(
        stub_->AsyncSayHello(&context, request, &cq));

    // 设置rpc完成时动作,结果参数,状态,和tag(tag用来方便应用层能区分不同的请求,虽然grpc内部有每个请求自己的uid,但是对于应用层却不能用,通过在这里设置一个tag,可以方便的区别,因为异步的可能会多个请求通过rpc发出,接收到后都放到了CompletionQueue中,所以应用层要能区分,就要设置一个tag)
    rpc->Finish(&reply, &status, (void*)1);
    void* got_tag;
    bool ok = false;
    
    // CompletionQueue的next会阻塞到有一个结果为止,got_tag会设置成前面的tag,而ok会设置成代码是否成功的拿到了响应(ok与status不同,status是表示服务器的结果,而ok只是表示阻塞队列中Next的结果)
    GPR_ASSERT(cq.Next(&got_tag, &ok));

    // 通过tag区分不同的响应对应的请求
    GPR_ASSERT(got_tag == (void*)1);
    GPR_ASSERT(ok);

    // 最终返回值
    if (status.ok()) {
      return reply.message();
    } else {
      return "RPC failed";
    }
  }

 private:
  std::unique_ptr<Greeter::Stub> stub_;
};

有三点要注意的 :
1. 上面展示了异步调用的使用方式,但是实际工作中不会这么去用,因为它这样写还是一个同步的;
2. 这种异步的方式是基于CompletionQueue做的,与protobuf自带的rpc基于回调的方式完全不同,如果是protobuf rpc,那么异步回调会由rpc内部线程来做;而grpc在实际工作中,还是得自己有一个线程去调用CompletionQueue的next来等待消息。
3. 这个protobuf rpc方式相差很大,但是思想却是回调由谁来调用引起的,由rpc线程来调用,那就要传回调给内部,如果由用户线程来调用,就要由一个队列先保存结果;

这里有一个疑问:
为什么要自己创建一个reply,而不是让grpc内部自己创建,一方面增加了一个参数,另一方面也增加了管理的复杂性,比如有多次调用异步接口,那么为了有在另一个线程中使用,要把这个reply对象保存起来,再根据tag去读取不同的对象,这样比较复杂,如果是grpc内部创建,那么就么可以把结果保存到CompletionQueue中,通过next来得到结果。如果只是为了保持谁创建谁销毁的原则,如果是内部创建,可以通过返回一个std::unique_ptr来保存可以自动销毁。

所以总的来说,异步调用接口对于应用层来说还不是不太方便,可以像protobuf rpc一样接口一个回调函数(参数就是response, status和tag);这样更方便一些;

GRPC服务器异步

这里的异步服务器是指这个情况:当一个任务需要时间比较长时,不能一直占用工作线程,这时需要启动另一个线程去做,完成之后通知回server的工作线程处理;
在这种情况其实和客户端的需要用异步的情况一样,都是不能阻塞线程,让另一个线程处理。
如果是自己来实现这样的逻辑,会怎么处理呢?
收到异步调用请求-》首先是保存上下文(比如调用参数,client的writer),加入一个队列中-》在新的线程从队列中取出任务处理-》调用writer向client写数据。

grpc的基本流程:

    // 注册一个服务,并启动
    helloworld::Greeter::AsyncService service;
    ServerBuilder builder;
    builder.AddListeningPort("0.0.0.0:50051", InsecureServerCredentials());
    builder.RegisterAsyncService(&service);
    auto cq = builder.AddCompletionQueue();
    auto server = builder.BuildAndStart();
    // 创建一个任务,去等待请求(异步,所以这里会马上返回), 当收到一个sayhello的调用之后,context,request,responder进行初始化,然后加入队列中(根据tag来设置)
    ServerContext context;
    HelloRequest request;
    ServerAsyncResponseWriter<HelloReply> responder;
    service.RequestSayHello(&context, &request, &responder, &cq, &cq, (void*)1);
    // 等待队列结果,如果tag是上面设置的值,说明这个结果就是得到了请求调用,然后就可以开始处理了
    HelloReply reply;
    Status status;
    void* got_tag;
    bool ok = false;
    cq.Next(&got_tag, &ok);
    if (ok && got_tag == (void*)1) {
      // set reply and status
      // 创建完成之后,调用responder.Finish发送结果,如果发送成功,也会以tag为标识加入队列中
      responder.Finish(reply, status, (void*)2);
    }
    // 处理完成之后,删除自己
    void* got_tag;
    bool ok = false;
    cq.Next(&got_tag, &ok);
    if (ok && got_tag == (void*)2) {
      // clean up
    }

从上面的流程和一般的想法有点不一样:
它是先调用RequestSayHello开启处理流程(相当于注册了一个处理器)-》然后异步请求调用到达-》根据RequestSayHello参数加入一个队列-》从队列中取出数据处理-》调用responder.Finish发送-》发送状态又会入队列-》清理这次的处理流程。
和同步调用比起来,没有去注册service,而就直接注册对应的方法;并且由于是异步处理,所以服务器不主动调用,所以它会把收到的请求等信息放入队列,我们自己要去遍历;

上面的是基本的流程,下面是一个完整的例子,通过CallData对象,封装了一个异步请求的所以动作(这个对象封装得很好,一种异步请求一个类型,外界不用关注它的内部实现);

grpc的服务器异步调用完整代码:

class ServerImpl final {
 public:
  ~ServerImpl() {
    server_->Shutdown();
    // Always shutdown the completion queue after the server.
    cq_->Shutdown();
  }

  // 开始注册service,并运行
  void Run() {
    std::string server_address("0.0.0.0:50051");

    ServerBuilder builder;
    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:
  // 这个类封装了异步请求处理所需要的状态与逻辑
  // Class encompasing the state and logic needed to serve a request.
  class CallData {
   public:
    // 创建一个calldata对象,然后开始调用调用对应的函数;这个对象封装了请求,ctx(用于发送消息给军客户端)
    CallData(Greeter::AsyncService* service, ServerCompletionQueue* cq)
        : service_(service), cq_(cq), responder_(&ctx_), status_(CREATE) {
      Proceed();
    }

    void Proceed() {
      if (status_ == CREATE) {
        // 开始注册SayHello异步处理流程,当收到一个sayhello的请求调用后,会加入队列
        status_ = PROCESS;
        service_->RequestSayHello(&ctx_, &request_, &responder_, cq_, cq_,
                                  this);
      } else if (status_ == PROCESS) {
        // 这里又新创建了处理流程,不然可能在处理过程中有请求来就不能及时处理
        new CallData(service_, cq_);
        // 从队列中拿到请求,真正的处理逻辑,这个会在一个新线程中运行
        // The actual processing.
        std::string prefix("Hello ");
        reply_.set_message(prefix + request_.name());
        // 设置状态
        status_ = FINISH;
        responder_.Finish(reply_, Status::OK, this);
      } else {
        // 发送完成之后入队列中的数据
        GPR_ASSERT(status_ == FINISH);
        // Once in the FINISH state, deallocate ourselves (CallData).
        delete this;
      }
    }

   private:
    // 异步service
    Greeter::AsyncService* service_;
    // 一个生产者消费者队列
    ServerCompletionQueue* cq_;
    ServerContext ctx_;
    HelloRequest request_;
    HelloReply reply_;
    // 用于消息回复的方法
    ServerAsyncResponseWriter<HelloReply> responder_;

    // Let's implement a tiny state machine with the following states.
    enum CallStatus { CREATE, PROCESS, FINISH };
    CallStatus status_;  // The current serving state.
  };

  // This can be run in multiple threads if needed.
  void HandleRpcs() {
    // 创建一个新的,由于新创建的status为create,所以它马上会开始service的RequestSayHello方法
    new CallData(&service_, cq_.get());
    void* tag;  // uniquely identifies a request.
    bool ok;
    while (true) {
      // Block waiting to read the next event from the completion queue. The
      // event is uniquely identified by its tag, which in this case is the
      // memory address of a CallData instance.
      // The return value of Next should always be checked. This return value
      // tells us whether there is any kind of event or cq_ is shutting down.
      // 从
      GPR_ASSERT(cq_->Next(&tag, &ok));
      GPR_ASSERT(ok);
      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;
}
原文链接:,转发请注明来源!

发表评论