gRPC C++异步模型多RPC方法扩展优化方案问询

你遇到的痛点非常真实——官方异步示例那种「一个RPC对应一个CallData派生类」的模式，在300-400个RPC的场景下完全不可行：不仅要写几百个重复结构的类，高频请求下的对象创建销毁也会带来不必要的性能损耗。下面给你一套能让异步代码接近同步模型简洁性的解决方案，同时兼顾扩展性和性能。

核心思路：通用模板类+回调逻辑 #

我们可以用模板类封装通用的CallData逻辑，把每个RPC的请求注册方法、业务处理逻辑作为参数传入，彻底避免为每个RPC写单独的派生类。本质是把「每个RPC的差异化逻辑」从类继承结构中抽离出来，用函数对象（lambda/绑定函数）的方式注入。

第一步：定义通用的异步CallData基类与模板类 #

先写一个抽象基类统一Proceed接口，再用模板类处理不同RPC的请求/响应类型：

#include <grpcpp/grpcpp.h>
#include <functional>
#include <memory>
#include <mutex>
#include <queue>

// 抽象基类，用于统一处理完成队列的tag回调
class AsyncCallDataBase {
public:
    virtual ~AsyncCallDataBase() = default;
    virtual void Proceed() = 0;
};

// 通用模板类，适配任意Request/Response类型的RPC
template <typename RequestType, typename ResponseType>
class AsyncCallData : public AsyncCallDataBase {
public:
    // 定义gRPC服务Request方法的函数签名
    using RequestFunc = void (grpc::Service::*)(
        grpc::ServerContext*, RequestType*,
        grpc::ServerAsyncResponseWriter<ResponseType>*,
        grpc::CompletionQueue*, grpc::CompletionQueue*, void*);

    // 业务处理逻辑的回调签名：输入上下文、请求，输出响应和状态
    using HandlerFunc = std::function<grpc::Status(
        const grpc::ServerContext&, const RequestType&, ResponseType*)>;

    // 构造函数：传入服务实例、Request方法、处理逻辑、完成队列
    AsyncCallData(grpc::Service* service, RequestFunc request_func,
                  HandlerFunc handler, grpc::CompletionQueue* cq)
        : service_(service), request_func_(request_func),
          handler_(std::move(handler)), cq_(cq), responder_(&ctx_) {
        // 初始化时立即注册第一个请求到完成队列
        Proceed();
    }

    void Proceed() override {
        if (status_ == CallStatus::CREATE) {
            // 注册下一个同类型RPC请求到完成队列，用当前对象作为唯一tag
            (service_->*request_func_)(&ctx_, &request_, &responder_, cq_, cq_, this);
            status_ = CallStatus::PROCESS;
        } else if (status_ == CallStatus::PROCESS) {
            // 立即创建新的CallData实例，确保能处理后续同类型请求（不阻塞当前请求处理）
            new AsyncCallData(service_, request_func_, handler_, cq_);

            // 执行业务逻辑（复用同步模型的写法）
            grpc::Status status = handler_(ctx_, request_, &response_);

            // 发送响应，完成后触发FINISH状态
            responder_.Finish(response_, status, this);
            status_ = CallStatus::FINISH;
        } else {
            // FINISH状态：销毁当前对象（如果用对象池可以改成放回池）
            GPR_ASSERT(status_ == CallStatus::FINISH);
            delete this;
        }
    }

private:
    enum class CallStatus { CREATE, PROCESS, FINISH };
    CallStatus status_ = CallStatus::CREATE;

    grpc::Service* service_;
    RequestFunc request_func_;
    HandlerFunc handler_;
    grpc::CompletionQueue* cq_;

    // 每个请求的上下文、请求、响应对象
    grpc::ServerContext ctx_;
    RequestType request_;
    ResponseType response_;
    grpc::ServerAsyncResponseWriter<ResponseType> responder_;
};

第二步：注册所有RPC并启动服务 #

现在你可以像写同步服务一样，把每个RPC的业务逻辑用lambda或独立函数实现，然后在HandleRpcs里一行代码注册一个RPC，完全不需要写几百个派生类：

// 假设你的proto生成的服务类是MyService
class MyService final : public MyProto::MyService::Service {
    // 同步模型里的override方法现在换成回调逻辑
};

void HandleRpcs(std::unique_ptr<grpc::CompletionQueue> cq, MyService* service) {
    // 注册RPC1：传入Request方法和业务逻辑lambda
    new AsyncCallData<MyProto::RPC1Request, MyProto::RPC1Response>(
        service, &MyService::RequestRPC1,
        [](const grpc::ServerContext& ctx, const MyProto::RPC1Request& req, MyProto::RPC1Response* resp) {
            // 这里写RPC1的业务逻辑，和同步模型写法一致
            resp->set_result(req.input() * 2);
            return grpc::Status::OK;
        }, cq.get());

    // 注册RPC2：同理
    new AsyncCallData<MyProto::RPC2Request, MyProto::RPC2Response>(
        service, &MyService::RequestRPC2,
        [](const grpc::ServerContext& ctx, const MyProto::RPC2Request& req, MyProto::RPC2Response* resp) {
            resp->set_message("Hello " + req.name());
            return grpc::Status::OK;
        }, cq.get());

    // ... 剩下的300+个RPC都用这种方式注册，复制粘贴改参数即可

    // 轮询完成队列处理请求
    void* tag;
    bool ok;
    while (cq->Next(&tag, &ok)) {
        GPR_ASSERT(ok);
        static_cast<AsyncCallDataBase*>(tag)->Proceed();
    }
}

性能优化：对象池减少内存分配开销 #

如果你的QPS达到10万+，每次请求创建销毁AsyncCallData对象会带来内存分配的开销，可以用对象池复用实例：

template <typename RequestType, typename ResponseType>
class AsyncCallDataPool {
public:
    AsyncCallDataPool(grpc::Service* service, typename AsyncCallData<RequestType, ResponseType>::RequestFunc request_func,
                      typename AsyncCallData<RequestType, ResponseType>::HandlerFunc handler, grpc::CompletionQueue* cq,
                      size_t init_size = 200) {
        // 预创建一批实例放入池
        for (size_t i = 0; i < init_size; ++i) {
            pool_.push(new AsyncCallData<RequestType, ResponseType>(service, request_func, handler, cq));
        }
    }

    // 从池里取出实例（如果空了就新创建）
    AsyncCallData<RequestType, ResponseType>* Take() {
        std::lock_guard<std::mutex> lock(mutex_);
        if (pool_.empty()) {
            return new AsyncCallData<RequestType, ResponseType>(service_, request_func_, handler_, cq_);
        }
        auto data = pool_.front();
        pool_.pop();
        return data;
    }

    // 用完放回池
    void Return(AsyncCallData<RequestType, ResponseType>* data) {
        std::lock_guard<std::mutex> lock(mutex_);
        pool_.push(data);
    }

private:
    grpc::Service* service_;
    typename AsyncCallData<RequestType, ResponseType>::RequestFunc request_func_;
    typename AsyncCallData<RequestType, ResponseType>::HandlerFunc handler_;
    grpc::CompletionQueue* cq_;

    std::queue<AsyncCallData<RequestType, ResponseType>*> pool_;
    std::mutex mutex_;
};

然后修改AsyncCallData的Proceed方法，在FINISH状态时把对象放回池而不是销毁，同时重置状态以便复用。

方案优势总结 #

1. 代码简洁性：完全不用写几百个派生类，每个RPC的注册和逻辑实现和同步模型几乎一样
2. 类型安全：模板自动匹配请求/响应类型，避免手动维护标签映射的错误
3. 高性能：模板实例化由编译器完成，没有额外的虚函数开销（仅基类一次虚调用），配合对象池可进一步降低内存分配损耗
4. 扩展性：新增RPC只需要加一行注册代码，无需修改核心逻辑

内容的提问来源于stack exchange，提问作者WhiteSword