求基于C++的gRPC通用代理服务器实现方案及示例代码

需求概述 #

• 实现proto无关的gRPC通用代理服务器，无需知晓客户端与后端服务的proto定义即可转发任意gRPC请求并返回响应
• 支持所有gRPC服务类型：
  • 一元RPC
  • 客户端流式RPC
  • 服务器流式RPC
  • 双向流式RPC
• 工作在gRPC层，可提取请求中的服务名、方法名等元数据
• 适用于需通过第三方库创建gRPC通道对接服务的场景（TCP/HTTP2代理无法满足）

方案1：基于grpc::AsyncGenericService与grpc::GenericStub的实现（推荐） #

gRPC C++原生提供了通用服务和存根类，专门用于处理未知proto的RPC请求，完全满足proto无关的需求。以下是核心骨架代码：

服务器端（通用代理服务） #

#include <grpcpp/grpcpp.h>
#include <memory>
#include <thread>
#include <iostream>

class GenericProxyServer {
public:
    GenericProxyServer(const std::string& target_addr) 
        : target_channel_(grpc::CreateChannel(target_addr, grpc::InsecureChannelCredentials())),
          stub_(grpc::GenericStub::NewStub(target_channel_)) {}

    void Run(int port) {
        std::string server_addr = "0.0.0.0:" + std::to_string(port);
        grpc::ServerBuilder builder;
        
        // 注册通用异步服务
        grpc::AsyncGenericService generic_service;
        builder.RegisterAsyncGenericService(&generic_service);
        builder.AddListeningPort(server_addr, grpc::InsecureServerCredentials());
        
        cq_ = builder.AddCompletionQueue();
        server_ = builder.BuildAndStart();
        std::cout << "Proxy server listening on " << server_addr << std::endl;

        // 启动请求处理循环
        HandleRpcs(&generic_service);
    }

private:
    // 单个请求的处理上下文
    class CallData {
    public:
        CallData(grpc::AsyncGenericService* service, grpc::ServerCompletionQueue* cq, grpc::GenericStub* stub)
            : service_(service), cq_(cq), stub_(stub), responder_(&ctx_), status_(CREATE) {
            Proceed();
        }

        void Proceed() {
            if (status_ == CREATE) {
                status_ = PROCESS;
                // 接收任意gRPC请求，触发回调
                service_->RequestCall(&ctx_, &request_buf_, &responder_, cq_, cq_, this);
            } else if (status_ == PROCESS) {
                // 提取服务名、方法名（格式为 /service_name/method_name）
                const std::string& full_method = ctx_.method();
                std::cout << "Forwarding request: " << full_method << std::endl;

                // 构建后端调用上下文，转发客户端元数据
                grpc::ClientContext client_ctx;
                for (const auto& meta : ctx_.client_metadata()) {
                    client_ctx.AddMetadata(meta.first, meta.second);
                }

                // 发起对后端的通用调用
                auto rpc_call = stub_->PrepareCall(&client_ctx, full_method, cq_);
                rpc_call->StartCall();
                rpc_call->Write(request_buf_, this);
                rpc_call->WritesDone(this);
                rpc_call->Finish(&response_buf_, &rpc_status_, this);

                status_ = FINISH;
            } else {
                // 将后端响应返回给客户端
                responder_.Finish(response_buf_, rpc_status_, this);
                delete this;
            }
        }

    private:
        grpc::AsyncGenericService* service_;
        grpc::ServerCompletionQueue* cq_;
        grpc::GenericStub* stub_;
        grpc::ServerContext ctx_;
        grpc::GenericServerAsyncReaderWriter responder_;
        grpc::ByteBuffer request_buf_;
        grpc::ByteBuffer response_buf_;
        grpc::Status rpc_status_;

        enum CallStatus { CREATE, PROCESS, FINISH };
        CallStatus status_;
    };

    void HandleRpcs(grpc::AsyncGenericService* service) {
        // 初始化第一个请求处理实例
        new CallData(service, cq_.get(), stub_.get());
        void* tag;
        bool ok;
        // 持续处理队列中的请求
        while (true) {
            GPR_ASSERT(cq_->Next(&tag, &ok));
            GPR_ASSERT(ok);
            static_cast<CallData*>(tag)->Proceed();
        }
    }

    std::shared_ptr<grpc::Channel> target_channel_;
    std::unique_ptr<grpc::GenericStub> stub_;
    std::unique_ptr<grpc::Server> server_;
    std::unique_ptr<grpc::ServerCompletionQueue> cq_;
};

int main(int argc, char** argv) {
    if (argc != 3) {
        std::cerr << "Usage: proxy_server <listen_port> <backend_addr>" << std::endl;
        return 1;
    }
    int port = std::stoi(argv[1]);
    std::string backend_addr = argv[2];

    GenericProxyServer server(backend_addr);
    server.Run(port);

    return 0;
}

关键说明 #

• 该实现通过grpc::AsyncGenericService接收任意gRPC请求，无需绑定具体proto服务定义
• 从grpc::ServerContext::method()提取完整方法路径，可解析出服务名和方法名
• 使用grpc::GenericStub向后端发起通用调用，自动适配所有四种RPC类型（流式/非流式请求均通过grpc::ByteBuffer传输）
• 异步模型支持高并发，可直接扩展元数据修改、请求拦截等逻辑

方案2：基于google.protobuf.Any的通用Proto实现 #

若方案1受限于环境无法使用，可采用通用双向流式Proto定义，该定义可兼容所有四种RPC类型：

通用Proto定义 #

syntax = "proto3";
package genericproxy;
import "google/protobuf/any.proto";

message ProxyGenericRequest {
  google.protobuf.Any request = 1;
}

message ProxyGenericResponse {
  google.protobuf.Any response = 1;
}

service ProxyGenericService {
  rpc ProxyGenericMethod(stream ProxyGenericRequest) returns (stream ProxyGenericResponse);
}

兼容性证明 #

1. 一元RPC：客户端发送1个ProxyGenericRequest（包裹实际请求消息），服务端返回1个ProxyGenericResponse（包裹实际响应），模拟一元调用流程
2. 客户端流式RPC：客户端连续发送多个ProxyGenericRequest，服务端返回1个ProxyGenericResponse
3. 服务器流式RPC：客户端发送1个ProxyGenericRequest，服务端连续返回多个ProxyGenericResponse
4. 双向流式RPC：客户端与服务端自由收发多个ProxyGenericRequest和ProxyGenericResponse，完全匹配双向流交互模型

注意：该方案需要客户端和后端服务都适配此通用Proto，无法做到完全透明的代理，仅作为方案1的备选。

已知实现/骨架参考 #

• gRPC官方提供了基础的异步通用服务示例，可在此基础上扩展转发逻辑
• 云原生代理项目如Envoy的gRPC代理模块采用类似的通用处理逻辑，其C++核心代码可作为实现参考

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