如何在C++服务器中实现长时间gRPC异步流数据更新

// 服务端代码
class MyServiceImplementation final : public MyService::Service {
public:
  grpc::Status GetDataUpdates(grpc::ServerContext* context, const DataRequest* request, grpc::ServerWriter<DataUpdate>* writer) override {
    // 在这里实现数据更新的逻辑
    while (true) {
      // 生成数据更新
      DataUpdate update;
      // 设置数据更新的内容
      update.set_data("New data update");

      // 发送数据更新给客户端
      writer->Write(update);

      // 模拟等待一段时间
      std::this_thread::sleep_for(std::chrono::seconds(1));
    }

    return grpc::Status::OK;
  }
};

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

  grpc::ServerBuilder builder;
  builder.AddListeningPort(server_address, grpc::InsecureServerCredentials());
  builder.RegisterService(&service);

  std::unique_ptr<grpc::Server> server(builder.BuildAndStart());
  std::cout << "Server listening on " << server_address << std::endl;

  server->Wait();
}

// 客户端代码
void GetDataUpdates() {
  std::shared_ptr<grpc::Channel> channel = grpc::CreateChannel("localhost:50051", grpc::InsecureChannelCredentials());
  std::unique_ptr<MyService::Stub> stub = MyService::NewStub(channel);

  grpc::ClientContext context;
  DataRequest request;
  // 设置请求参数
  request.set_param("Request parameter");

  std::unique_ptr<grpc::ClientReader<DataUpdate>> reader = stub->GetDataUpdates(&context, request);

  DataUpdate update;
  while (reader->Read(&update)) {
    // 处理接收到的数据更新
    std::cout << "Received data update: " << update.data() << std::endl;
  }

  grpc::Status status = reader->Finish();
  if (status.ok()) {
    std::cout << "Data updates streaming finished successfully." << std::endl;
  } else {
    std::cout << "Data updates streaming failed with error: " << status.error_message() << std::endl;
  }
}

int main() {
  std::thread server_thread(RunServer);

  // 等待一段时间，确保服务端已经启动
  std::this_thread::sleep_for(std::chrono::seconds(1));

  GetDataUpdates();

  server_thread.join();

  return 0;
}