grpc c ++中的异步模型

Question

我的团队正在设计一个具有微服务架构的可扩展解决方案，并计划使用 gRPC 作为层之间的传输通信。我们决定使用异步 grpc 模型。如果我扩展 RPC 方法的数量，example(greeter_async_server.cc) 提供的设计似乎不可行，因为我必须为每个 RPC 方法创建一个新类，并像这样在HandleRpcs() 中创建它们的对象. Pastebin（简短示例代码）。

   void HandleRpcs() {
            new CallDataForRPC1(&service_, cq_.get());
            new CallDataForRPC2(&service_, cq_.get());
            new CallDataForRPC3(&service, cq_.get());
            // so on...
    }

它将被硬编码，所有的灵活性都将丢失。

我有大约 300-400 个 RPC 方法来实现，而拥有 300-400 个类会很麻烦当我必须处理超过 100K 的 RPC 请求/秒时，这个解决方案是非常糟糕的设计。我无法承受在每个请求上以这种方式创建对象的开销。有人可以为我提供一个解决方法吗？ async grpc c++ 不能像它的同步伴侣那样简单吗？

编辑：为了让情况更清晰，对于那些可能难以掌握这个异步示例流程的人，我写的是我目前所理解的内容，请如果某处有错，请纠正我。

在异步 grpc 中，每次我们必须将唯一标记与完成队列绑定，这样当我们轮询时，服务器可以在特定的 RPC 将被客户端命中时将其返回给我们，我们推断来自返回的关于调用类型的唯一标签。

service_->RequestRPC2(&ctx_, &request_, &responder_, cq_, cq_,this); 这里我们使用当前对象的地址作为唯一标签。这就像在完成队列上注册我们的 RPC 调用。然后我们在HandleRPCs() 中向下轮询以查看客户端是否命中了 RPC，如果是，则cq_->Next(&tag, &OK) 将填充标签。轮询代码sn -p：

while (true) {
          GPR_ASSERT(cq_->Next(&tag, &ok));
          GPR_ASSERT(ok);
          static_cast<CallData*>(tag)->Proceed();
        }

因为，我们注册到队列中的唯一标签是 CallData 对象的地址，所以我们可以调用Proceed()。这对于一个逻辑在Proceed() 中的 RPC 来说很好。但是每次都有更多的 RPC，我们会将它们全部放在 CallData 中，然后在轮询时，我们将调用唯一的一个 Proceed()，它将包含（比如）RPC1（postgres 调用）、RPC2（mongodb 调用）的逻辑）， .. 很快。这就像在一个函数中编写我的所有程序一样。因此，为了避免这种情况，我使用了 GenericCallData 类和 virtual void Proceed() 并从中创建了派生类，每个 RPC 一个类，在它们自己的 Proceed() 中具有自己的逻辑。这是一个可行的解决方案，但我想避免编写很多类。

我尝试的另一个解决方案是将所有 RPC-function-logics 排除在 proceed() 之外并放入它们自己的函数中，并维护一个全局 std::map<long, std::function</*some params*/>> 。因此，每当我将带有唯一标签的 RPC 注册到队列中时，我都会存储其相应的逻辑函数（我肯定会将其硬编码到语句中并绑定所有需要的参数），然后将唯一标签作为键。在轮询时，当我得到&tag 时，我会在地图中查找此键并调用相应的保存函数。现在，还有一个障碍，我必须在函数逻辑中执行此操作：

// pseudo code
void function(reply, responder, context, service)
{
    // register this RPC with another unique tag so to serve new incoming request of the same type on the completion queue
     service_->RequestRPC1(/*params*/, new_unique_id);
    // now again save this new_unique_id and current function into the map, so when tag will be returned we can do lookup
     map.emplace(new_unique_id, function);

    // now you're free to do your logic
    // do your logic
}

您看到了，代码现在已经传播到另一个模块中，并且它是基于 RPC 的。 希望它能清除情况。 我想如果有人能以更简单的方式实现这种类型的服务器。

Answer 1

这篇文章现在已经很老了，但我还没有看到任何关于这个的答案或例子，所以我将向任何其他读者展示我是如何解决它的。我有大约 30 个 RPC 调用，并且正在寻找一种在添加和删除 RPC 调用时减少占用空间的方法。我花了一些时间来找出解决问题的好方法。

所以我从 (g)RPC 库中获取 RPC 请求的接口是接收方需要实现的回调接口。界面如下所示：

class IRpcRequestHandler
{
public:
    virtual ~IRpcRequestHandler() = default;
    virtual void onZigbeeOpenNetworkRequest(const smarthome::ZigbeeOpenNetworkRequest& req,
                                            smarthome::Response& res) = 0;
    virtual void onZigbeeTouchlinkDeviceRequest(const smarthome::ZigbeeTouchlinkDeviceRequest& req,
                                                smarthome::Response& res) = 0;
    ...
};

还有一些用于在 gRPC 服务器启动后设置/注册每个 RPC 方法的代码：

void ready() 
{
    SETUP_SMARTHOME_CALL("ZigbeeOpenNetwork", // Alias that is used for debug messages
                         smarthome::Command::AsyncService::RequestZigbeeOpenNetwork,  // Generated gRPC service method for async.
                         smarthome::ZigbeeOpenNetworkRequest, // Generated gRPC service request message
                         smarthome::Response, // Generated gRPC service response message
                         IRpcRequestHandler::onZigbeeOpenNetworkRequest); // The callback method to call when request has arrived.

    SETUP_SMARTHOME_CALL("ZigbeeTouchlinkDevice",
                         smarthome::Command::AsyncService::RequestZigbeeTouchlinkDevice,
                         smarthome::ZigbeeTouchlinkDeviceRequest,
                         smarthome::Response,
                         IRpcRequestHandler::onZigbeeTouchlinkDeviceRequest);
    ...
}

这是您在添加和删除 RPC 方法时需要关心的全部内容。

SETUP_SMARTHOME_CALL 是一个自制的宏，如下所示：

#define SETUP_SMARTHOME_CALL(ALIAS, SERVICE, REQ, RES, CALLBACK_FUNC) \
  new ServerCallData<REQ, RES>(                                       \
      ALIAS,                                                          \
      std::bind(&SERVICE,                                             \
                &mCommandService,                                     \
                std::placeholders::_1,                                \
                std::placeholders::_2,                                \
                std::placeholders::_3,                                \
                std::placeholders::_4,                                \
                std::placeholders::_5,                                \
                std::placeholders::_6),                               \
      mCompletionQueue.get(),                                         \
      std::bind(&CALLBACK_FUNC, requestHandler, std::placeholders::_1, std::placeholders::_2))

我认为 ServerCallData 类看起来像 gRPC 示例中的类，但做了一些修改。 ServerCallData 派生自具有抽象函数 void proceed(bool ok) 的非模板类，用于 CompletionQueue::Next() 处理。当创建 ServerCallData 时，它将调用 SERVICE 方法在 CompletionQueue 上注册自己，并且在每次第一次 proceed(ok) 调用时，它将克隆自己，这将注册另一个实例。如果有人感兴趣，我也可以发布一些示例代码。

编辑：在下面添加了更多示例代码。

GrpcServer

class GrpcServer
{
 public:
  explicit GrpcServer(std::vector<grpc::Service*> services);
  virtual ~GrpcServer();

  void run(const std::string& sslKey,
           const std::string& sslCert,
           const std::string& password,
           const std::string& listenAddr,
           uint32_t port,
           uint32_t threads = 1);

 private:
  virtual void ready();  // Called after gRPC server is created and before polling CQ.
  void handleRpcs();  // Function that polls from CQ, can be run by multiple threads. Casts object to CallData and calls CallData::proceed().

  std::unique_ptr<ServerCompletionQueue> mCompletionQueue;
  std::unique_ptr<Server> mServer;
  std::vector<grpc::Service*> mServices;
  std::list<std::shared_ptr<std::thread>> mThreads;
  ...
}

以及CallData对象的主要部分：

template <typename TREQUEST, typename TREPLY>
class ServerCallData : public ServerCallMethod
{
 public:
  explicit ServerCallData(const std::string& methodName,
                          std::function<void(ServerContext*,
                                             TREQUEST*,
                                             ::grpc::ServerAsyncResponseWriter<TREPLY>*,
                                             ::grpc::CompletionQueue*,
                                             ::grpc::ServerCompletionQueue*,
                                             void*)> serviceFunc,
                          grpc::ServerCompletionQueue* completionQueue,
                          std::function<void(const TREQUEST&, TREPLY&)> callback,
                          bool first = false)
      : ServerCallMethod(methodName),
        mResponder(&mContext),
        serviceFunc(serviceFunc),
        completionQueue(completionQueue),
        callback(callback)
  {
    requestNewCall();
  }

  void proceed(bool ok) override
  {
    if (!ok)
    {
      delete this;
      return;
    }

    if (callStatus() == ServerCallMethod::PROCESS)
    {    
      callStatus() = ServerCallMethod::FINISH;
      new ServerCallData<TREQUEST, TREPLY>(callMethodName(), serviceFunc, completionQueue, callback);

      try
      {
        callback(mRequest, mReply);
      }
      catch (const std::exception& e)
      {
        mResponder.Finish(mReply, Status::CANCELLED, this);
        return;
      }

      mResponder.Finish(mReply, Status::OK, this);
    }
    else
    {    
      delete this;
    }
  }

 private:
  void requestNewCall()
  {
    serviceFunc(
        &mContext, &mRequest, &mResponder, completionQueue, completionQueue, this);
  }

  ServerContext mContext;
  TREQUEST mRequest;
  TREPLY mReply;
  ServerAsyncResponseWriter<TREPLY> mResponder;
  std::function<void(ServerContext*,
                     TREQUEST*,
                     ::grpc::ServerAsyncResponseWriter<TREPLY>*,
                     ::grpc::CompletionQueue*,
                     ::grpc::ServerCompletionQueue*,
                     void*)>
      serviceFunc;
  std::function<void(const TREQUEST&, TREPLY&)> callback;
  grpc::ServerCompletionQueue* completionQueue;
};

Answer 2

虽然线程很旧，但我想分享一个我目前正在实施的解决方案。它主要由继承 CallData 的模板类组成，以实现可扩展性。这样，每个新的 rpc 将只需要专门化所需 CallData 方法的模板。

Calldata 标头：

class CallData {
    protected:
        enum Status { CREATE, PROCESS, FINISH };
        Status status;

        virtual void treat_create() = 0;
        virtual void treat_process() = 0;

    public:
        void Proceed();
};

CallData 继续执行：

void CallData::Proceed() {
    switch (status) {
        case CREATE:
            status = PROCESS;
            treat_create();
            break;
        case PROCESS:
            status = FINISH;
            treat_process();
            break;
        case FINISH:
            delete this;
    }
}

继承自CallData 标头（简化）：

template <typename Request, typename Reply>
class CallDataTemplated : CallData {
    static_assert(std::is_base_of<google::protobuf::Message, Request>::value, 
        "Request and reply must be protobuf messages");
    static_assert(std::is_base_of<google::protobuf::Message, Reply>::value,
        "Request and reply must be protobuf messages");

    private:
        Service,Cq,Context,ResponseWriter,...
        Request request;
        Reply reply;

    protected:
        void treat_create() override;
        void treat_process() override;

    public:
        ...
};

然后，对于特定的 rpc，理论上您应该能够执行以下操作：

template<>
void CallDataTemplated<HelloRequest, HelloReply>::treat_process() {
     ...
}

从我的角度来看，它有很多模板化方法，但比为每个 rpc 创建一个类更可取。