Как сделать реальный запрос потока? - PullRequest
Новая
       10

Как сделать реальный запрос потока?

3 голосов
/ 13 мая 2019

Я начинаю изучать Webflux из Spring-boot. Я узнал, что для конечной точки RestController вы можете определить тело запроса Flux, где я ожидаю реальный поток потока, то есть части всего запроса идут одна за другой, и эти части могут обрабатываться также одна за другой. , Однако после создания небольшого примера с клиентом и сервером я не смог заставить его работать так, как ожидалось.

Итак, вот фрагмент сервера: 

@PostMapping("/digest")
    public Flux<String> digest(@RequestBody Flux<String> text) {
        continuousMD5.reset();
        return text.log("server.request.").map(piece -> continuousMD5.update(piece)).log("server.response.");
    }

Примечание: каждый фрагмент текста будет отправлен в объект непрерывного MD5, который будет накапливать все фрагменты и вычислять и возвращать промежуточное значение хеш-функции MD5 после каждого накопления. Поток будет зарегистрирован до и после расчета MD5.

А вот фрагмент клиента: 

@PostConstruct
    private void init() {
        webClient = webClientBuilder.baseUrl(reactiveServerUrl).build();
    }

@PostMapping(value = "/send", consumes = MediaType.TEXT_PLAIN_VALUE)
    public Flux<String> send(@RequestBody Flux<String> text) {
        return webClient.post()
            .uri("/digest")
            .accept(MediaType.TEXT_PLAIN)
            .body(text.log("client.request."), String.class)
            .retrieve().bodyToFlux(String.class).log("client.response.");
    }

Примечание: клиент принимает поток потока некоторого текста, регистрирует поток и отправляет его на сервер (как поток потока).

Удивительно, но я заставил его отправить запрос REST и позволить клиенту получить поток потока с помощью следующей командной строки: 

for i in $(seq 1 100); do echo "The message $i"; done | http POST :8080/send  Content-Type:text/plain

и я мог видеть это в журнале клиента: 

2019-05-09 17:02:08.604  INFO 3462 --- [ctor-http-nio-2] client.response.Flux.MonoFlatMapMany.2   : onSubscribe(MonoFlatMapMany.FlatMapManyMain)
2019-05-09 17:02:08.606  INFO 3462 --- [ctor-http-nio-2] client.response.Flux.MonoFlatMapMany.2   : request(1)
2019-05-09 17:02:08.649  INFO 3462 --- [ctor-http-nio-6] client.request.Flux.SwitchIfEmpty.1      : onSubscribe(FluxSwitchIfEmpty.SwitchIfEmptySubscriber)
2019-05-09 17:02:08.650  INFO 3462 --- [ctor-http-nio-6] client.request.Flux.SwitchIfEmpty.1      : request(32)
2019-05-09 17:02:08.674  INFO 3462 --- [ctor-http-nio-2] client.request.Flux.SwitchIfEmpty.1      : onNext(The message 1)
2019-05-09 17:02:08.676  INFO 3462 --- [ctor-http-nio-2] client.request.Flux.SwitchIfEmpty.1      : request(1)
2019-05-09 17:02:08.676  INFO 3462 --- [ctor-http-nio-2] client.request.Flux.SwitchIfEmpty.1      : onNext(The message 2)
...
2019-05-09 17:02:08.710  INFO 3462 --- [ctor-http-nio-2] client.request.Flux.SwitchIfEmpty.1      : onNext(The message 100)
2019-05-09 17:02:08.710  INFO 3462 --- [ctor-http-nio-6] client.request.Flux.SwitchIfEmpty.1      : request(1)
2019-05-09 17:02:08.710  INFO 3462 --- [ctor-http-nio-6] client.request.Flux.SwitchIfEmpty.1      : request(1)
2019-05-09 17:02:08.710  INFO 3462 --- [ctor-http-nio-6] client.request.Flux.SwitchIfEmpty.1      : request(1)
2019-05-09 17:02:08.711  INFO 3462 --- [ctor-http-nio-2] client.request.Flux.SwitchIfEmpty.1      : onComplete()
2019-05-09 17:02:08.711  INFO 3462 --- [ctor-http-nio-6] client.request.Flux.SwitchIfEmpty.1      : request(1)
2019-05-09 17:02:08.711  INFO 3462 --- [ctor-http-nio-6] client.request.Flux.SwitchIfEmpty.1      : request(1)
2019-05-09 17:02:08.860  INFO 3462 --- [ctor-http-nio-6] client.response.Flux.MonoFlatMapMany.2   : onNext(CSubeSX3yIVP2CD6FRlojg==)
2019-05-09 17:02:08.862  INFO 3462 --- [ctor-http-nio-6] client.response.Flux.MonoFlatMapMany.2   : onComplete()
^C2019-05-09 17:02:47.393  INFO 3462 --- [ctor-http-nio-6] client.request.Flux.SwitchIfEmpty.1      : cancel()

что каждый фрагмент текста был распознан как элемент потока потока и был запрошен отдельно.

Но в журнале сервера: 

2019-05-09 17:02:08.811  INFO 3475 --- [ctor-http-nio-2] server.request.Flux.SwitchIfEmpty.1      : onSubscribe(FluxSwitchIfEmpty.SwitchIfEmptySubscriber)
2019-05-09 17:02:08.813  INFO 3475 --- [ctor-http-nio-2] server.response.Flux.Map.2               : onSubscribe(FluxMap.MapSubscriber)
2019-05-09 17:02:08.814  INFO 3475 --- [ctor-http-nio-2] server.response.Flux.Map.2               : request(1)
2019-05-09 17:02:08.814  INFO 3475 --- [ctor-http-nio-2] server.request.Flux.SwitchIfEmpty.1      : request(1)
2019-05-09 17:02:08.838  INFO 3475 --- [ctor-http-nio-2] server.request.Flux.SwitchIfEmpty.1      : onNext(The message 1The message 2The message 3The message 4The message 5The message 6The message 7The message 8The message 9The message 10The message 11The message 12The message 13The message 14The message 15The message 16The message 17The message 18The message 19The message 20The message 21The message 22The message 23The message 24The message 25The message 26The message 27The message 28The message 29The message 30The message 31The message 32The message 33The message 34The message 35The message 36The message 37The message 38The message 39The message 40The message 41The message 42The message 43The message 44The message 45The message 46The message 47The message 48The message 49The message 50The message 51The message 52The message 53The message 54The message 55The message 56The message 57The message 58The message 59The message 60The message 61The message 62The message 63The message 64The message 65The message 66The message 67The message 68The message 69The message 70The message 71The message 72The message 73The message 74The message 75The message 76The message 77The message 78The message 79The message 80The message 81The message 82The message 83The message 84The message 85The message 86The message 87The message 88The message 89The message 90The message 91The message 92The message 93The message 94The message 95The message 96The message 97The message 98The message 99The message 100)
2019-05-09 17:02:08.840  INFO 3475 --- [ctor-http-nio-2] server.response.Flux.Map.2               : onNext(CSubeSX3yIVP2CD6FRlojg==)
2019-05-09 17:02:08.852  INFO 3475 --- [ctor-http-nio-2] server.response.Flux.Map.2               : request(32)
2019-05-09 17:02:08.852  INFO 3475 --- [ctor-http-nio-2] server.request.Flux.SwitchIfEmpty.1      : request(32)
2019-05-09 17:02:08.852  INFO 3475 --- [ctor-http-nio-2] server.request.Flux.SwitchIfEmpty.1      : onComplete()
2019-05-09 17:02:08.852  INFO 3475 --- [ctor-http-nio-2] server.response.Flux.Map.2               : onComplete()
2019-05-09 17:02:47.394  INFO 3475 --- [ctor-http-nio-2] server.response.Flux.Map.2               : cancel()
2019-05-09 17:02:47.394  INFO 3475 --- [ctor-http-nio-2] server.request.Flux.SwitchIfEmpty.1      : cancel()

Я видел, что все фрагменты текста поступили на сервер одновременно и, следовательно, были обработаны как один большой элемент в потоке потока (можно также убедиться, что вместо 100 вычислен только один хэш MD5. * 1024). *

Я ожидал бы, что сервер также получит фрагменты текста от клиента в качестве элементов потока потока, в противном случае для сервера это не будет реактивным, а просто обычным запросом на блокировку.

Может ли кто-нибудь помочь мне понять, как сделать реальный реактивный запрос потока с помощью Webflux? Спасибо!

Обновление

Я использовал аналогичную командную строку, чтобы сделать REST-запрос к серверу, и увидел, что сервер получил фрагменты текста («Сообщение x») в виде потока. Итак, я думаю, что сервер в порядке, теперь проблема может быть в клиенте: как я могу использовать WebClient для выполнения реального потока REST-запроса?

Ответы [ 2 ]

1 голос
/ 13 мая 2019

Если вы хотите добиться эффекта потоковой передачи, вы можете:

0 голосов
/ 14 мая 2019

Попробовав все и прочитав больше документации, я наконец понял, как заставить мой пример работать:

Для клиента мне нужно убедиться, что тело запроса, отправляемое на сервер, также разделено переводом строки: 

@PostMapping(value = "/send", consumes = MediaType.TEXT_PLAIN_VALUE)
    public Flux<String> send(@RequestBody Flux<String> text) {
        return webClient.post()
            .uri("/digest")
            .accept(MediaType.TEXT_PLAIN)
            .body(
                text
                    .onBackpressureBuffer()
                    .log("client.request.")
                    .map(piece -> piece + "\n"),
                String.class)
            .retrieve().bodyToFlux(String.class)
            .onBackpressureBuffer()
            .log("client.response.");
    }

это дает тот же эффект, что и выполнение запроса REST через командную строку, так как for i in $(seq 1 100); do echo "The message $i"; done выводит «Сообщение x» в строках.

Аналогично, для сервера тело ответа также должно быть разделено переводом строки, чтобы клиент мог декодировать тело в поток: 

@PostMapping("/digest")
    public Flux<String> digest(@RequestBody Flux<String> text) {
        continuousMD5.reset();
        return text
            .log("server.request.")
            .map(piece -> continuousMD5.update(piece))
            .map(piece -> piece + "\n")
            .log("server.response.");
    }

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

Однако, несмотря на то, что приведенный выше код «работает», но он не выполняет потоковую передачу, как я вижу в журналах, сервер начал получать тело запроса после того, как клиент отправил все тело запроса, и клиент начал получить тело ответа после того, как сервер отправил все тело ответа. Возможно, как упоминал Илья Зинкович, использование протокола WebSocket может привести к реальному потоковому эффекту, но я пока не пробовал.

...