C++远程函数调用实战：基于gRPC的高性能分布式系统设计

1次阅读

没有评论

共计 3619 个字符，预计需要花费 10 分钟才能阅读完成。

现代分布式系统中，远程过程调用（RPC）主要有三种实现方式：

传统 REST API：基于 HTTP/1.1 文本传输，JSON 序列化开销大，难以实现双向流式通信
二进制协议（如 Thrift）：性能较好但生态工具链较弱，跨语言支持有限
gRPC：Google 开源的 HTTP/ 2 多语言 RPC 框架，提供四类通信模式（单一请求 / 响应、服务端流、客户端流、双向流）

实测数据表明（测试环境：16 核 32GB 云服务器，payload 1KB）：

gRPC 延迟比 REST 低 3 - 5 倍（平均 0.8ms vs 4.2ms）
连接复用使 QPS 提升 6 倍（12,000 vs 2,000）
Protobuf 二进制编码体积比 JSON 小 40%

创建 calculator.proto 定义服务契约：

syntax = "proto3";

package rpc_example;

// 建议每个服务单独定义消息类型
message AddRequest {
  int32 operand1 = 1;  // 字段编号必须从 1 开始
  int32 operand2 = 2;
}

message AddResponse {
  int32 result = 1;
  string checksum = 2; // 演示复合返回值
}

service Calculator {
  // 简单 RPC
  rpc Add (AddRequest) returns (AddResponse);

  // 服务端流式示例
  rpc PrimeFactors (AddRequest) returns (stream AddResponse);
}

关键规范：

使用 proto3 语法以获得更好的默认值处理
字段编号预留扩展空间（如 1 -15 给常用字段）
为流式方法添加 stream 修饰符

class CalculatorImpl final : public Calculator::Service {
  Status Add(ServerContext* context, const AddRequest* request,
             AddResponse* reply) override {reply->set_result(request->operand1() + request->operand2());
    reply->set_checksum(MD5(request)); // 假设的校验函数
    return Status::OK;
  }
};

void RunServer() {
  ServerBuilder builder;
  builder.AddListeningPort("0.0.0.0:50051", grpc::InsecureServerCredentials());
  builder.RegisterService(new CalculatorImpl());

  // 配置工作线程池
  builder.SetSyncServerOption(ServerBuilder::MIN_POLLERS, 4);
  builder.SetSyncServerOption(ServerBuilder::MAX_POLLERS, 16);

  std::unique_ptr<Server> server(builder.BuildAndStart());
  server->Wait();}

class AsyncPrimeFactors final : public AsyncService {
  void RequestPrimeFactors(ServerContext* context,
                          AddRequest* request,
                          ServerAsyncWriter<AddResponse>* writer,
                          CompletionQueue* cq,
                          void* tag) {int n = request->operand1();
    for (int i = 2; i <= n; ++i) {while (n % i == 0) {
        AddResponse response;
        response.set_result(i);
        writer->Write(response, tag); // 流式写入
        n /= i;
      }
    }
    writer->Finish(Status::OK, tag);
  }
};

class GrpcClientPool {
public:
  explicit GrpcClientPool(int pool_size) {for (int i = 0; i < pool_size; ++i) {
      channels_.push_back(grpc::CreateChannel("localhost:50051", grpc::InsecureChannelCredentials()));
      stubs_.push_back(Calculator::NewStub(channels_.back()));
    }
  }

  // 使用 Round Robin 负载均衡
  Calculator::Stub& GetStub() {return *stubs_[index_++ % stubs_.size()];
  }

private:
  std::vector<shared_ptr<Channel>> channels_;
  std::vector<std::unique_ptr<Calculator::Stub>> stubs_;
  std::atomic<size_t> index_{0};
};

CompletionQueue cq;
AddRequest request;
request.set_operand1(100);
request.set_operand2(200);

ClientContext context;
context.set_deadline(std::chrono::system_clock::now() + 
                    std::chrono::milliseconds(500)); // 500ms 超时

AddResponse response;
Status status;
stub->async()->Add(&context, &request, &response,
                  [&status](bool ok) { status = ok ? Status::OK 
                                                 : Status::CANCELLED; });

cq.Next(&tag, &ok); // 非阻塞等待

// 使用 gRPC 内置的重试策略
auto args = grpc::ChannelArguments();
args.SetServiceConfigJSON(R"(
{
  "methodConfig": [{"name": [{"service": "rpc_example.Calculator"}],
    "retryPolicy": {
      "maxAttempts": 3,
      "initialBackoff": "0.1s",
      "maxBackoff": "1s",
      "backoffMultiplier": 2,
      "retryableStatusCodes": ["UNAVAILABLE"]
    }
  }]
}
)");