Джерси webservice масштабируемый подход для загрузки файла и ответа клиенту

Мне нужно создать веб-сервис с Джерси, который загружает большой файл из другой службы и возвращается клиенту. Я хотел бы, чтобы майка прочитала несколько байтов в буфер и напишет эти байты на клиентский сокет.

Я бы хотел, чтобы он использовал неблокирующий ввод-вывод, поэтому я не занимаюсь нитью. (Этого не может быть достигнуто)

    @GET
    @Path("mypath")
    public void getFile(final @Suspended AsyncResponse res) {
        Client client = ClientBuilder.newClient();
        WebTarget t = client.target("http://webserviceURL");
        t.request()
            .header("some header", "value for header")
                .async().get(new InvocationCallback<byte[]>(){

            public void completed(byte[] response) {
                res.resume(response);
            }

            public void failed(Throwable throwable) {
                res.resume(throwable.getMessage());
                throwable.printStackTrace();
                //reply with error
            }

        });
    }

До сих пор у меня этот код, и я считаю, что Джерси будет загружать полный файл, а затем писать его клиенту, который я не хочу делать. любые мысли?

Ответ 1

Асинхронный запрос на стороне клиента не собирается делать многое для вашего варианта использования. Это более важно для случаев "пожара и забывания". Что вы можете сделать, это просто получить InputStream от клиента Response и смешать с серверной стороной StreamingResource для потоковой передачи результатов. Сервер начнет отправлять данные по мере поступления с другого удаленного ресурса.

Ниже приведен пример. Конечная точка "/file" - это фиктивный удаленный ресурс, который обслуживает файл. Конечная точка "/client" потребляет его.

@Path("stream")
@Produces(MediaType.APPLICATION_OCTET_STREAM)
public class ClientStreamingResource {

    private static final String INFILE = "Some File";

    @GET
    @Path("file")
    public Response fileEndpoint() {
        final File file = new File(INFILE);
        final StreamingOutput output = new StreamingOutput() {
            @Override
            public void write(OutputStream out) {

                try (FileInputStream in = new FileInputStream(file)) {
                    byte[] buf = new byte[512];
                    int len;
                    while ((len = in.read(buf)) != -1) {
                        out.write(buf, 0, len);
                        out.flush();
                        System.out.println("---- wrote 512 bytes file ----");
                    }
                } catch (IOException ex) {
                    throw new InternalServerErrorException(ex);
                }
            }
        };
        return Response.ok(output)
                .header(HttpHeaders.CONTENT_LENGTH, file.length())
                .build();
    }

    @GET
    @Path("client")
    public void clientEndpoint(@Suspended final AsyncResponse asyncResponse) {
        final Client client = ClientBuilder.newClient();
        final WebTarget target = client.target("http://localhost:8080/stream/file");
        final Response clientResponse = target.request().get();

        final StreamingOutput output = new StreamingOutput() {
            @Override
            public void write(OutputStream out) {
                try (final InputStream entityStream = clientResponse.readEntity(InputStream.class)) {
                    byte[] buf = new byte[512];
                    int len;
                    while ((len = entityStream.read(buf)) != -1) {
                        out.write(buf, 0, len);
                        out.flush();
                        System.out.println("---- wrote 512 bytes client ----");
                    }
                } catch (IOException ex) {
                    throw new InternalServerErrorException(ex);
                }
            }
        };
        ResponseBuilder responseBuilder = Response.ok(output);
        if (clientResponse.getHeaderString("Content-Length") != null) {
            responseBuilder.header("Content-Length", clientResponse.getHeaderString("Content-Length"));
        }
        new Thread(() -> {
            asyncResponse.resume(responseBuilder.build());
        }).start();
    }
}

Я использовал cURL, чтобы сделать запрос, и jetty-maven-plugin, чтобы иметь возможность запускать пример из командной строки. Когда вы запустите его и сделаете запрос, вы увидите журнал регистрации

---- wrote 512 bytes file ----
---- wrote 512 bytes file ----
---- wrote 512 bytes client ----
---- wrote 512 bytes file ----
---- wrote 512 bytes client ----
---- wrote 512 bytes file ----
---- wrote 512 bytes client ----
---- wrote 512 bytes file ----
---- wrote 512 bytes client ----
...

в то время как cURL клиент отслеживает результаты

Отличие от этого заключается в том, что ведение журнала "удаленного сервера" происходит в то же время, что и клиентский ресурс. Это показывает, что клиент не ждет получения всего файла. Он начинает отправлять байты, как только он начинает их получать.

Некоторые вещи, которые следует отметить в этом примере:

Я использовал очень маленький размер буфера (512), потому что я тестировал небольшой (1 Мб) файл. Я действительно не хотел ждать большого файла для тестирования. Но я думаю, что большие файлы должны работать одинаково. Конечно, вы захотите увеличить размер буфера до более крупного.
Чтобы использовать меньший размер буфера, вам нужно установить свойство Jersey ServerProperties.OUTBOUND_CONTENT_LENGTH_BUFFER на 0. Причина заключается в том, что Джерси хранит внутренний буфер размером 8192, что приведет к тому, что мои 512-байтовые фрагменты данных не будут скрыты, пока не будут буферизованы 8192 байта. Поэтому я просто отключил его.
При использовании AsyncResponse вы должны использовать другой поток, как и я. Вы можете использовать исполнителей вместо явного создания потоков. Если вы не используете другой поток, вы все равно удерживаете поток из пула потоков контейнера.

UPDATE

Вместо управления своими потоками/исполнителем вы можете аннотировать ресурс клиента с помощью @ManagedAsync и позволить Джерси управлять потоками

@ManagedAsync
@GET
@Path("client")
public void clientEndpoint(@Suspended final AsyncResponse asyncResponse) {
    ...
    asyncResponse.resume(responseBuilder.build());
}