如何在C++ gRPC中使用原始字节负载并规避Protobuf序列化？

如果你想完全绕过Protobuf的序列化逻辑，直接在gRPC双向流中传输原始字节，可以通过以下两种方式实现：

方法一：极简Protobuf消息（低开销、易实现） #

gRPC必须依赖.proto文件定义服务结构，但我们可以定义一个仅包含bytes字段的极简消息，让Protobuf仅作为字节的容器，自己处理序列化/反序列化逻辑。这种方式的开销极低（仅Protobuf对bytes字段的微小头部），且实现简单。

1. 编写.proto文件 #

syntax = "proto3";

package rawgrpc;

// 仅用于承载原始字节的消息
message RawData {
  bytes payload = 1;
}

// 双向流服务定义
service RawStreamService {
  rpc BidirectionalStream(stream RawData) returns (stream RawData);
}

2. 生成C++代码 #

用protoc和gRPC C++插件生成代码：

protoc --grpc_out=. --plugin=protoc-gen-grpc=`which grpc_cpp_plugin` raw.proto
protoc --cpp_out=. raw.proto

3. 服务端实现 #

直接读取RawData的payload字段作为原始字节，处理后再将结果塞回payload发送：

#include "raw.grpc.pb.h"
#include <grpcpp/grpcpp.h>

using grpc::Server;
using grpc::ServerBuilder;
using grpc::ServerContext;
using grpc::ServerReaderWriter;
using grpc::Status;
using rawgrpc::RawData;
using rawgrpc::RawStreamService;

// 自定义原始字节处理逻辑
std::string process_raw_data(const std::string& input) {
  // 替换为你的业务逻辑
  return "processed_" + input;
}

class RawStreamServiceImpl final : public RawStreamService::Service {
  Status BidirectionalStream(ServerContext* context,
                             ServerReaderWriter<RawData, RawData>* stream) override {
    RawData request;
    while (stream->Read(&request)) {
      // 获取客户端发来的原始字节
      const std::string& raw_input = request.payload();
      // 处理原始数据
      std::string raw_output = process_raw_data(raw_input);
      // 构造响应并发送
      RawData response;
      response.set_payload(raw_output);
      stream->Write(response);
    }
    return Status::OK;
  }
};

void RunServer() {
  std::string server_address("0.0.0.0:50051");
  RawStreamServiceImpl service;

  ServerBuilder builder;
  builder.AddListeningPort(server_address, grpc::InsecureServerCredentials());
  builder.RegisterService(&service);
  std::unique_ptr<Server> server(builder.BuildAndStart());
  server->Wait();
}

int main() {
  RunServer();
  return 0;
}

4. 客户端实现 #

将原始字节封装进RawData的payload发送，接收时直接读取payload：

#include "raw.grpc.pb.h"
#include <grpcpp/grpcpp.h>

using grpc::Channel;
using grpc::ClientContext;
using grpc::ClientReaderWriter;
using grpc::Status;
using rawgrpc::RawData;
using rawgrpc::RawStreamService;

class RawStreamClient {
 public:
  RawStreamClient(std::shared_ptr<Channel> channel)
      : stub_(RawStreamService::NewStub(channel)) {}

  void SendAndReceive(const std::string& input) {
    ClientContext context;
    std::unique_ptr<ClientReaderWriter<RawData, RawData>> stream(
        stub_->BidirectionalStream(&context));

    // 发送原始字节
    RawData request;
    request.set_payload(input);
    stream->Write(request);

    // 接收响应
    RawData response;
    while (stream->Read(&response)) {
      std::cout << "Received processed data: " << response.payload() << std::endl;
    }

    stream->WritesDone();
    Status status = stream->Finish();
    if (!status.ok()) {
      std::cerr << "Stream failed: " << status.error_message() << std::endl;
    }
  }

 private:
  std::unique_ptr<RawStreamService::Stub> stub_;
};

int main() {
  RawStreamClient client(grpc::CreateChannel(
      "localhost:50051", grpc::InsecureChannelCredentials()));
  client.SendAndReceive("test_raw_bytes");
  return 0;
}

方法二：Generic gRPC API（完全规避Protobuf开销） #

如果想彻底摆脱Protobuf的任何开销，可以使用gRPC的Generic API，直接发送/接收原始的gRPC消息帧。这种方式实现更复杂，但完全不依赖Protobuf消息定义。

1. 服务端实现（异步） #

#include <grpcpp/grpcpp.h>
#include <grpcpp/generic/generic_stub.h>

using grpc::GenericService;
using grpc::GenericServerContext;
using grpc::GenericServerAsyncReaderWriter;
using grpc::CompletionQueue;
using grpc::ByteBuffer;
using grpc::Status;
using grpc::ServerBuilder;

// 自定义原始字节处理逻辑
std::string process_raw_data(const std::string& input) {
  return "generic_processed_" + input;
}

class RawGenericService : public GenericService {
 public:
  Status HandleCall(GenericServerContext* context,
                    GenericServerAsyncReaderWriter* stream,
                    CompletionQueue* cq) override {
    new CallData(stream, cq);
    return Status::OK;
  }

 private:
  class CallData {
   public:
    CallData(GenericServerAsyncReaderWriter* stream, CompletionQueue* cq)
        : stream_(stream), cq_(cq), state_(READ) {
      StartRead();
    }

    void StartRead() {
      stream_->Read(&request_buf_, this);
    }

    void OnReadDone(bool ok) {
      if (!ok) {
        stream_->Finish(Status::OK, this);
        delete this;
        return;
      }
      // 将ByteBuffer转为原始字符串
      std::string raw_input;
      grpc::ConvertByteBufferToString(&request_buf_, &raw_input);
      // 处理原始数据
      std::string raw_output = process_raw_data(raw_input);
      // 构造响应ByteBuffer
      ByteBuffer response_buf(raw_output.data(), raw_output.size());
      stream_->Write(response_buf, this);
      state_ = WRITE;
    }

    void OnWriteDone(bool ok) {
      if (!ok) {
        stream_->Finish(Status(grpc::StatusCode::INTERNAL, "Write failed"), this);
        delete this;
        return;
      }
      state_ = READ;
      StartRead();
    }

    void OnFinishDone(bool ok) {
      delete this;
    }

    void operator()(bool ok) {
      if (state_ == READ) {
        OnReadDone(ok);
      } else if (state_ == WRITE) {
        OnWriteDone(ok);
      } else if (state_ == FINISH) {
        OnFinishDone(ok);
      }
    }

   private:
    GenericServerAsyncReaderWriter* stream_;
    CompletionQueue* cq_;
    ByteBuffer request_buf_;
    enum State { READ, WRITE, FINISH } state_;
  };
};

void RunServer() {
  std::string server_address("0.0.0.0:50051");
  RawGenericService service;

  ServerBuilder builder;
  builder.AddListeningPort(server_address, grpc::InsecureServerCredentials());
  builder.RegisterGenericService(&service);
  auto cq = builder.AddCompletionQueue();
  auto server = builder.BuildAndStart();

  void* tag;
  bool ok;
  while (cq->Next(&tag, &ok)) {
    static_cast<RawGenericService::CallData*>(tag)->operator()(ok);
  }
}

int main() {
  RunServer();
  return 0;
}

2. 客户端实现 #

#include <grpcpp/grpcpp.h>
#include <grpcpp/generic/generic_stub.h>

using grpc::Channel;
using grpc::ClientContext;
using grpc::GenericStub;
using grpc::GenericClientReaderWriter;
using grpc::ByteBuffer;
using grpc::Status;

class RawGenericClient {
 public:
  RawGenericClient(std::shared_ptr<Channel> channel) : stub_(channel) {}

  void SendAndReceive(const std::string& input) {
    ClientContext context;
    // 指定服务方法的全路径：包名.服务名/方法名
    auto stream = stub_.Call(&context, "/rawgrpc.RawStreamService/BidirectionalStream");

    // 发送原始字节
    ByteBuffer request_buf(input.data(), input.size());
    stream->Write(request_buf);

    // 接收响应
    ByteBuffer response_buf;
    while (stream->Read(&response_buf)) {
      std::string raw_output;
      grpc::ConvertByteBufferToString(&response_buf, &raw_output);
      std::cout << "Received generic processed data: " << raw_output << std::endl;
    }

    stream->WritesDone();
    Status status = stream->Finish();
    if (!status.ok()) {
      std::cerr << "Generic stream failed: " << status.error_message() << std::endl;
    }
  }

 private:
  GenericStub stub_;
};

int main() {
  RawGenericClient client(grpc::CreateChannel(
      "localhost:50051", grpc::InsecureChannelCredentials()));
  client.SendAndReceive("test_generic_raw_bytes");
  return 0;
}

注意事项 #

• 方法一的开销可以忽略不计，适合绝大多数场景，且兼容性更好（可以和其他语言的gRPC客户端/服务端互通）
• 方法二需要手动处理gRPC的消息帧边界，实现复杂度高，仅在对性能有极致要求的场景使用
• 无论哪种方式，都需要确保客户端和服务端的字节流处理逻辑一致（比如是否需要自定义帧分隔符、字节序等）

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