GRPC для загрузки больших файлов ⇐ JAVA

[Anonymous]

В настоящее время я обновляю озеро данных проекта, в котором микросервисы обмениваются данными по протоколу HTTP/1.1 на связь gRPC. Для этого проекта я использую Quarkus, Java (Gradle).
Раньше соединение между сервисами осуществлялось через quarkus.rest-client. Теперь я изменил его на quarkus.grpc.clients
При загрузке небольших файлов (от 4 КБ до 1–2 МБ) gRPC работает значительно лучше, чем оригинал, но когда он появляется для файлов гораздо большего размера, скажем, 20 МБ, мой gRPC работает ужасно. Я выполнил фрагментацию, но кадры работают очень плохо по сравнению с HTTP, поскольку я установил больший максимальный размер сообщения с помощью quarkus.grpc.server.max-inbound-message-size
Здесь у меня есть 2 службы, называемые приборной панелью и ядром. Я загружаю текстовый файл размером 20 МБ через Curl на панель управления службами. Затем панель управления и ядро будут взаимодействовать друг с другом с помощью gRPC, так что ядро ​​будет писать в dfs. Он работает, но все еще очень медленно работает с большими файлами.
Клиент gRPC (

Код: Выделить всё

dashboard

сервис):

Код: Выделить всё

        ObjectModel obj = new ObjectModel();
var objJson = "";
obj.cid = "null";
try {
byte[] bytes = new byte[1024*64];
int len;
while ((len = input.is.read(bytes)) != -1) {
ByteString byteString = ByteString.copyFrom(bytes, 0, len);
objJson = coreChunkGrpcClient.uploadCoreChunk(Multi.createFrom().item(UploadCoreChunkRequest.newBuilder()
.setBearer("Bearer " + jwt.getRawToken())
.setOutput(io.quarkus.example.ObjectFormModelChunk.newBuilder()
.setMetadata(output.metadata)
.setIs(byteString)
.setFileID(obj.cid)
.build())
.build())).onItem().transform(UploadCoreResponse::getToken).await().indefinitely();
obj.cid = objJson;
}

gRPC-сервер (

Код: Выделить всё

core

сервис)

Код: Выделить всё

    @POST
@Path("/file")
@Transactional
@Consumes(MediaType.MULTIPART_FORM_DATA)
@Produces(MediaType.APPLICATION_JSON)
@Operation(summary = "Create a new binary object")
@Override
public StreamObserver uploadCoreChunk(StreamObserver responseObserver) {
return new StreamObserver() {
@Override
public void onNext(UploadCoreChunkRequest request) {
var bearer = request.getBearer();
var inputs = request.getOutput();
String cid = inputs.getFileID();

try {
var result = insertFileChunk(bearer, inputs, cid);
cid = result.getResp().toString();
responseObserver.onNext(UploadCoreResponse.newBuilder().setCode(result.getCode()).setMessage(result.getMsg()).setToken(result.getResp().toString()).build());
} catch (IOException e) {
log.error("error parsing multipart form data {}", e.getMessage());
responseObserver.onNext(UploadCoreResponse.newBuilder().setCode(500).setMessage("Internal Error").build());
}
}

@Override
public void onError(Throwable t) {
...
}

@Override
public void onCompleted() {
...
}
};
}

Записать в фс:

Код: Выделить всё

    public LakeHttpResponse insertFileChunk(
@HeaderParam("Authorization") String bearer,
@RequestBody(description = "Multipart form data. metadata: extra json info " +
"{name:'original filename', gid: 'object group id', length: 'binary length').  file: binary data to save")
ObjectFormModelChunk input, String cid) throws IOException {

LakeObjectMetadata meta;
ByteString is = input.getIs();

// check chunk and save to backend
if (cid == null || cid.isEmpty() || cid.isBlank() || cid.equals("null")) {
cid = fs.create(meta.getName(), meta.getLength(), is);
return new LakeHttpResponse(200, "OK", cid);
}
else {
if (!fs.insertChunk(cid, is)) {
return new LakeHttpResponse(500, "Internal error", null);
}
}
LakeObject object = new LakeObject();
object.setCid(cid);
Long now = new Date().getTime();
object.setCreateTime(now);
object.setAccessTime(now);
object.setParentId(0L);
log.info("POST: object metadata saved cid={}", cid);
return new LakeHttpResponse(200, "OK", cid);
}

Ожидаемый результат: сокращение времени процесса.

Подробнее здесь: https://stackoverflow.com/questions/793 ... -uploading