异步 grpc 服务器处理多种类型的请求

Question

有没有更好的例子来说明如何实现多个异步请求。

 // The greeting service definition.
service Greeter {
  // Sends a greeting
  rpc ListSayHello1 (HelloRequest1) returns (stream HelloReply1) {}
  rpc ListSayHello2 (HelloRequest2) returns (stream HelloReply2) {}
}

// The request message containing the user's name.
message HelloRequest1 {
  string name = 1;
}

message HelloRequest2 {
  string name = 1;
}

// The response message containing the greetings
message HelloReply1 {
  string message = 1;
}

message HelloReply2 {
  string message = 1;
}

我正在使用下面的异步模式来处理SayHello1，但现在我想添加对SayHello2. 应该采取什么方法？

 new CallData(&service_, cq_.get(), *this);
    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();
    }

在CallData，我这样做

void CallData::Proceed()
{
    if (status_ == CREATE) {

        status_ = PROCESS;
        service_->RequestListSayHello1(&ctx_, &request_, &writer_, cq_, cq_,
            this);
    }
... 
}

Answer 1

没有必要创建额外的完成队列。我们只需要知道如何处理完成队列返回的内容。我们可以通过创建一个公共基类来解决这个问题，该基类能够执行必要的功能：

class CallDataBase
{
protected:
    virtual void WaitForRequest() = 0;
    virtual void HandleRequest() = 0;
public:
    virtual void Proceed() = 0;
    CallDataBase() {}
};

CallDataBase 的每个专业化都知道如何进行、WaitForRequest 和 HandleRequest。其中一些对于所有请求都是常见的，因此使用模板化类很方便：

template < class RequestType, class ReplyType>
class CallDataT : CallDataBase
{
protected:
    enum CallStatus { CREATE, PROCESS, FINISH };
    CallStatus status_;

    Greeter::AsyncService* service_;
    ServerCompletionQueue* completionQueue_;
    RequestType request_;
    ReplyType reply_;
    ServerAsyncResponseWriter<ReplyType> responder_;
    ServerContext serverContext_;

    // When we handle a request of this type, we need to tell
    // the completion queue to wait for new requests of the same type.
    virtual void AddNextToCompletionQueue() = 0;
public:
    CallDataT(Greeter::AsyncService* service, ServerCompletionQueue* completionQueue) :
        status_(CREATE),
        service_(service),
        completionQueue_(completionQueue),
        responder_(&serverContext_)
    {
    }
public:
    virtual void Proceed() override
    {
        if (status_ == CREATE)
        {
            status_ = PROCESS;
            WaitForRequest();
        }
        else if (status_ == PROCESS)
        {
            AddNextToCompletionQueue();
            HandleRequest();
            status_ = FINISH;
            responder_.Finish(reply_, Status::OK, this);
        }
        else
        {
            // We're done! Self-destruct!
            if (status_ != FINISH)
            {
                // Log some error message
            }
            delete this;
        }
    }
};

最后，消息类型的实际实现：

class CallDataHello : CallDataT<HelloRequest, HelloReply>
{
public:
    CallDataHello(Greeter::AsyncService* service, ServerCompletionQueue* completionQueue) : CallDataT(service, completionQueue)
    {
        Proceed();
    }
protected:
    virtual void AddNextToCompletionQueue() override
    {
        new CallDataHello(service_, completionQueue_);
    }
    virtual void WaitForRequest() override
    {
        service_->RequestSayHello(&serverContext_, &request_, &responder_, completionQueue_, completionQueue_, this);
    }
    virtual void HandleRequest() override
    {
        reply_.set_message(std::string("Hello ") + request_.name());
    }
};

class CallDataHelloAgain : CallDataT<HelloAgainRequest, HelloAgainReply>
{
public:
    CallDataHelloAgain(Greeter::AsyncService* service, ServerCompletionQueue* completionQueue) : CallDataT(service, completionQueue)
    {
        Proceed();
    }
protected:
    virtual void AddNextToCompletionQueue() override
    {
        new CallDataHelloAgain(service_, completionQueue_);
    }
    virtual void WaitForRequest() override
    {
        service_->RequestSayHelloAgain(&serverContext_, &request_, &responder_, completionQueue_, completionQueue_, this);
    }
    virtual void HandleRequest() override
    {
        reply_.set_message(std::string("Hello again ") + request_.name());
    }
};

最后，在GRPC服务器实现中，我们可以统一处理不同的请求：

void HandleRpcs() 
{
    new CallDataHello(&service_, completionQueue.get());
    new CallDataHelloAgain(engine_, &service_, completionQueue.get());
    void* tag;
    bool ok;
    while (true) {
        bool ret = completionQueue->Next(&tag, &ok);
        if (ok == false || ret == false)
        {
            return;
        }
        static_cast<CallDataBase*>(tag)->Proceed();
    }
}

这是可行的，因为我们添加到completionQueue 的所有CallData 都是基类CallDataBase，因此我们可以安全地对它们调用Proceed()。这种方法使得添加新请求相对容易。所有请求都由同一个完成队列处理。然而，所有的请求都将被串行处理，所以如果你需要更多的并行处理，我认为你必须创建多个完成队列。