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

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

RPC 核心概念与方案对比

远程过程调用 (RPC) 允许程序像调用本地函数一样执行远程服务，关键在于隐藏网络通信细节。主流方案中：

• REST：基于 HTTP/JSON，开发简单但性能较差（文本序列化开销大）
• gRPC：Google 开源的高性能框架，默认使用 Protocol Buffers 二进制编码，支持多语言和双向流
• Thrift：Facebook 的跨语言 RPC，但社区活跃度不如 gRPC

实测对比（单次调用延迟）：

1. gRPC+Protobuf：~150μs
2. Thrift：~220μs
3. REST/JSON：~1.2ms

Protocol Buffers 接口定义

.proto文件是 gRPC 的接口契约，示例定义计算服务：

syntax = "proto3";

service Calculator {rpc Add (AddRequest) returns (AddResponse);
  rpc BatchAdd (stream AddRequest) returns (AddResponse);
}

message AddRequest {
  int32 a = 1;
  int32 b = 2;
}

message AddResponse {int32 result = 1;}

关键规范：

• 字段编号而非名称决定二进制编码位置
• stream关键字声明流式接口
• 使用 protoc 编译器生成 C ++ 桩代码

C++ 服务端实现

基于 CompletionQueue 的异步服务端核心代码：

class AsyncCalculatorService final {
public:
  ~AsyncCalculatorService() {server_->Shutdown();
    cq_->Shutdown();}

  void Run() {
    ServerBuilder builder;
    builder.AddListeningPort(server_address_, grpc::InsecureServerCredentials());
    builder.RegisterService(&service_);
    cq_ = builder.AddCompletionQueue();
    server_ = builder.BuildAndStart();

    // 处理 RPC 请求的线程池
    std::vector<std::thread> workers;
    for (int i = 0; i < thread_pool_size_; ++i) {workers.emplace_back(&AsyncCalculatorService::HandleRpcs, this);
    }
    ...
  }

private:
  void HandleRpcs() {new CallData(&service_, cq_.get());
    void* tag;
    bool ok;
    while (cq_->Next(&tag, &ok)) {static_cast<CallData*>(tag)->Proceed(ok);
    }
  }
};

客户端优化技巧

连接池实现

class GrpcChannelPool {
public:
  std::shared_ptr<grpc::Channel> GetChannel() {std::lock_guard<std::mutex> lock(mutex_);
    if (channels_.empty()) {return CreateChannel();
    }
    auto channel = channels_.back();
    channels_.pop_back();
    return channel;
  }

  void ReleaseChannel(std::shared_ptr<grpc::Channel> channel) {std::lock_guard<std::mutex> lock(mutex_);
    channels_.push_back(channel);
  }
};

异步调用示例

void AsyncClient::BatchAdd() {
  CompletionQueue cq;
  ClientContext context;
  AddResponse response;

  auto stream = stub_->AsyncBatchAdd(&context, &response, &cq, nullptr);

  // 发送 10 个异步请求
  for (int i = 0; i < 10; ++i) {
    AddRequest request;
    request.set_a(i);
    request.set_b(i*2);
    stream->Write(request, nullptr);
  }
  stream->WritesDone(nullptr);

  // 等待响应
  Status status;
  stream->Finish(&status, nullptr);
  ...
}

生产环境实践

熔断机制实现

使用滑动窗口统计错误率：

class CircuitBreaker {
public:
  bool AllowRequest() {auto now = std::chrono::steady_clock::now();
    if (state_ == State::OPEN && 
        now > open_until_) {state_ = State::HALF_OPEN;}
    return state_ != State::OPEN;
  }

  void RecordFailure() {
    failures_++;
    if (failures_ > threshold_ && state_ != State::OPEN) {
      state_ = State::OPEN;
      open_until_ = std::chrono::steady_clock::now() + timeout_;}
  }
};

5 个常见陷阱及解决方案

1. 内存泄漏：异步调用未正确处理 tag 对象

2. 方案：使用 std::unique_ptr 管理生命周期

3. 线程阻塞：同步调用卡住主线程

4. 方案：统一使用 CompletionQueue 异步模式

5. 连接抖动：网络不稳定导致超时

6. 方案：实现指数退避重试机制

7. 版本冲突：proto 文件未同步更新

8. 方案：使用 CI 自动生成桩代码

9. 性能瓶颈：大量小包传输

10. 方案：启用 gRPC 消息压缩

扩展练习：带超时的批量调用

尝试实现以下功能：

1. 客户端批量发送 100 个加法请求
2. 设置 500ms 总超时
3. 单个请求超时 150ms 时自动重试
4. 使用 std::async 并行处理响应

核心提示：

context.set_deadline(std::chrono::system_clock::now() + std::chrono::milliseconds(500));

通过本文的实践，可构建延迟低于 200μs 的 C ++ 分布式服务。建议结合业务场景调整线程模型和连接池参数，并持续监控 gRPC 内置的指标（如grpc.server.calls）。