Reactive Core

• Для обробки запитів сервера існує два рівні засобів підтримки.

  • HttpHandler: Базовий контракт для обробки HTTP-запитів з неблокуючим введенням-виведенням та зворотною реакцією за специфікацією Reactive Streams, а також адаптери для Reactor Netty, Undertow, Tomcat, Jetty та будь-якого контейнера Servlet 3.1+. API: Трохи більш високорівневий веб-API загального призначення для обробки запитів, на основі якого будуються конкретні моделі програмування, такі як анотовані контролери та функціональні кінцеві точки.

• Для клієнтської частини базовий контракт ClientHttpConnector для виконання HTTP-запитів з неблокуючим введенням-виведенням та зворотною реакцією за специфікацією Reactive Streams, а також адаптери для Reactor Netty, реактивних Jetty HttpClient та Apache HttpComponents. Більш високорівневий WebClient, що використовується в додатках, базується на цьому базовому контракті.

• Для клієнта та сервера передбачені кодеки для серіалізації та десеріалізації вмісту запитів та відповідей HTTP.

HttpHandler

HttpHandler — це простий контракт з єдиним методом для обробки запиту та відповіді. Він навмисно строгий, а його головна і єдина мета — бути простою абстракцією над різними API HTTP-серверів.

У наступній таблиці описані залежності від сервера (також див. підтримувані версії):

У наведених нижче фрагментах коду показано використання адаптерів HttpHandler з кожним серверним API:

Reactor Netty

HttpHandler handler = ...
ReactorHttpHandlerAdapter adapter = new ReactorHttpHandlerAdapter(handler);
HttpServer.create().host(host).port(port).handle(adapter).bind().block();

val handler: HttpHandler = ...
val adapter = ReactorHttpHandlerAdapter (handler)
HttpServer.create().host(host).port(port).handle(adapter).bind().block()

Undertow

HttpHandler handler = ...
UndertowHttpHandlerAdapter adapter = new UndertowHttpHandlerAdapter(handler);
Undertow server = Undertow.builder().addHttpListener(port, host).setHandler(adapter).build();
server.start();

val handler: HttpHandler = ...
val adapter = UndertowHttpHandlerAdapter(handler)
val server = Undertow.builder ().addHttpListener(port, host).setHandler(adapter).build()
server.start()

Tomcat

HttpHandler handler = ...
Servlet servlet = new TomcatHttpHandlerAdapter(handler);
Tomcat server = new Tomcat();
File base = new File(System.getProperty("java.io.tmpdir"));
Context rootContext = server.addContext("", base.getAbsolutePath());
Tomcat.addServlet(rootContext, "main", servlet);
rootContext.addServletMappingDecoded("/", "main");
server.setHost(host);
server.setPort(port);
server.start();

val handler: HttpHandler = ...
val servlet = TomcatHttpHandlerAdapter(handler)
val server = Tomcat()
val base = File(System.getProperty("java.io.tmpdir"))
val rootContext = server.addContext("", base.absolutePath)
Tomcat.addServlet(rootContext, "main", servlet)
rootContext.addServletMappingDecoded("/", "main")
server.host = host
server.setPort(port)
server.start()

Jetty

HttpHandler handler = ...
Servlet servlet = new JettyHttpHandlerAdapter(handler);
Server server = new Server();
ServletContextHandler contextHandler = new ServletContextHandler(server, "");
contextHandler.addServlet(new ServletHolder(servlet), "/");
contextHandler.start();
ServerConnector connector = new ServerConnector(server);
connector.setHost(host);
connector.setPort(port);
server.addConnector(connector);
server.start();
        

val handler: HttpHandler = ...
val servlet = JettyHttpHandlerAdapter(handler)
val server = Server()
val contextHandler = ServletContextHandler(server, "")
contextHandler.addServlet(ServletHolder(servlet), "/")
contextHandler.start(); val
connector = ServerConnector(server)
connector.host = host connector.port = port
server.addConnector(connector)
server.start()

Контейнер Servlet 3.1+

Щоб розгорнути WAR-файл у будь-якому контейнері Servlet 3.1+, можна розширити та увімкнути AbstractReactiveWebInitializer у WAR-файл. Цей клас обертає HttpHandler у ServletHttpHandlerAdapter і реєструє його як Servlet.

WebHandler API

Пакет org.springframework.web.server заснований на контракті HttpHandler і надає веб-API загального призначення для обробки запитів через ланцюжок з кількох WebExceptionHandler, кількох WebFilter та одного компонента WebHandler. Ланцюжок може бути зібраний разом з WebHttpHandlerBuilder шляхом простого вказання на ApplicationContext з Spring, де компоненти визначаються автоматично, та/або шляхом реєстрації компонентів у конструкторі.

У той час як мета HttpHandler проста — абстрагувати використання різних HTTP-серверів, WebHandler API націлений на надання більш широкого набору функцій, які зазвичай використовуються у вебдодатках, таких як:

• Сесія користувача з атрибутами.

• Атрибути запиту.

• Дозволена Locale або Principal для запиту.

• Доступ до парсованих та кешованих даних форми.

• Абстракції для багатокомпонентних даних.

• і багато іншого...

Спеціалізовані види бінів

У таблиці нижче перераховані компоненти, які WebHttpHandlerBuilder може автоматично виявити в ApplicationContext із Spring, або які можуть бути зареєстровані безпосередньо в ньому:

Дані форми

ServerWebExchange надає наступний метод для доступу до даних форми:

Mono<MultiValueMap<String, String>> getFormData();

suspend fun getFormData(): MultiValueMap<String, String>

The DefaultServerWebExchange використовується для HttpMessageReader для попереднього формату часу (application/x-www-form-urlencoded) до MultiValueMap. За замовчуванням FormHttpMessageReader налаштований під використання біном ServerCodecConfigurer.

Багатокомпонентні дані

ServerWebExchange надає наступний метод для доступу до багатокомпонентних даних:

Mono<MultiValueMap<String, Part>> getMultipartData();

suspend fun getMultipartData(): MultiValueMap<String, Part>

DefaultServerWebExchange використовує налаштований HttpMessageReader<MultiValueMap<String, Part>> для парсингу вмісту multipart/form-data у Mul. За замовчуванням це DefaultPartHttpMessageReader, який не має жодних сторонніх залежностей. Як альтернативу можна використовувати SynchronossPartHttpMessageReader, заснований на бібліотеці Synchronoss NIO Multipart. Обидва конфігуруються за допомогою біна ServerCodecConfigurer.

Для потокового парсингу багатокомпонентних даних можна використовувати Flux<Part>, що повертається з HttpMessageReader<Part>. Наприклад, в анотованому контролері використання @RequestPart має на увазі Map-подібний доступ до окремих компонентів на ім'я і, отже, вимагає повного парсингу багатокомпонентних даних. І навпаки, ти можеш використовувати анотацію @RequestBody для декодування вмісту у Flux<Part> без збору в MultiValueMap.

Заголовки, що пересилаються

Якщо запит проходить через проксі-сервери (наприклад, розподільники навантаження), хост, порт і схема можуть змінюватися. Це ускладнює завдання клієнта щодо створення посилань, що вказують на правильний хост, порт та схему.

7239 визначає HTTP-заголовок Forwarded, який проксі-сервери можуть використовувати для надання інформації про вихідний запит. Існують і інші нестандартні заголовки, включно з X-Forwarded-Host, X-Forwarded-Port, X-Forwarded-Proto, X -Forwarded-Ssl та X-Forwarded-Prefix.

ForwardedHeaderTransformer — це компонент, який змінює хост, порт і схему запиту на основі пересилаються заголовків, а потім видаляє ці заголовки. Якщо ти оголосиш його як бін з ім'ям forwardedHeaderTransformer, він буде виявлений і використаний.

Є деякі застереження щодо безпеки для заголовків, що пересилаються, оскільки програма не може знати, чи були заголовки додані проксі-сервером, як передбачалося, або шкідливим клієнтом. Саме тому проксі-сервер на межі довіри повинен бути налаштований на видалення ненадійного трафіку, що пересилається, що надходить ззовні. Ти також можеш налаштувати ForwardedHeaderTransformer за допомогою removeOnly=true, і в цьому випадку він буде видаляти, але не використовувати заголовки.

Фільтри

В API WebHandler ти можеш використовувати WebFilter для застосування логіки в стилі перехоплення перед та після решти ланцюжка обробки фільтрів та цільового WebHandler. При використанні конфігурації WebFlux зареєструвати WebFilter вкрай просто: оголошуємо його як бін Spring і (опціонально) висловлюємо рівень старшинства, використовуючи анотацію @Order в оголошенні біна або реалізуючи клас Ordered .

CORS

Spring WebFlux забезпечує тонку підтримку конфігурації CORS за допомогою анотацій для контролерів. Однак, якщо ти використовуєш його зі Spring Security, то радимо покладатися на вбудований CorsFilter, який повинен знаходитися по порядку попереду всього ланцюжка фільтрів Spring Security.

Exceptions

В API WebHandler ти можеш використовувати WebExceptionHandler для обробки винятків із ланцюжка екземплярів WebFilter та цільового WebHandler. При використанні конфігурації WebFlux реєструвати обробник WebExceptionHandler так само просто, як оголошувати його як бін Spring і (опціонально) висловлювати старшість за допомогою анотації @Order в оголошенні біну або шляхом реалізації класу Ordered.

У наступній таблиці описані доступні реалізації WebExceptionHandler:

Кодеки

Модулі spring-web та spring-core забезпечують засоби підтримки серіалізації та десеріалізації байтового вмісту в об'єкти вищого рівня та з них за допомогою неблокуючого введення-виведення зі зворотною реакцією зі специфікації Reactive Streams. Нижче наведено ці засоби підтримки:

• Encoder та Decoder — це контракти низького рівня для кодування та декодування вмісту незалежно від HTTP протоколу.

• HttpMessageReader та HttpMessageWriter — це контракти для кодування та декодування вмісту HTTP-повідомлень.

• Encoder можна обернути в EncoderHttpMessageWriter, щоб адаптувати його під використання у вебпрограмі, а Decoder можна обернути в DecoderHttpMessageReader.

• DataBuffer абстрагує різні уявлення байтових буферів (наприклад, ByteBuf, java.nio.ByteBuffer для Netty і т.д.), і саме з ним і працюють усі кодеки.

Модуль spring-core містить реалізації кодера та декодера byte[], ByteBuffer, DataBuffer, Resource та String. Модуль spring-web містить Jackson JSON, Jackson Smile, JAXB2, Protocol Buffers та інші кодувальники та декодувальники поряд з реалізаціями веб-орієнтованих засобів читання та запису HTTP-повідомлень для даних форми, багатокомпонентного вмісту, подій, що надсилаються сервером тощо.

ClientCodecConfigurer та ServerCodecConfigurer зазвичай використовуються, щоб конфігурувати та налаштовувати кодеки для використання в додатку.

Jackson JSON

Формат JSON та двійковий JSON (Smile) підтримуються за наявності бібліотеки Jackson.

Jackson2Decoder працює таким чином:

• Асинхронний, неблокуючий парсер бібліотеки Jackson використовується для об'єднання потоку байтових фрагментів в екземпляри TokenBuffer, кожен з яких являє собою об'єкт JSON.

• Кожен TokenBuffer передається до ObjectMapper бібліотеки Jackson для створення об'єкта вищого рівня.

• При декодуванні в публікатор з одним значенням (наприклад, Mono), існує один TokenBuffer.

• При декодуванні в багатозначний публікатор (наприклад, Flux) кожен TokenBuffer передається до ObjectMapper, як тільки буде отримано достатньо байтів інформації для повністю сформованого об'єкта. Вміст може бути масив JSON або будь-який формат JSON з розмежуванням рядків, такий як NDJSON, JSON Lines або JSON Text Sequences.

Jackson2Encoder працює таким чином:

• Для публікатора з одним значенням (наприклад, Mono) просто серіалізуємо його через ObjectMapper.

• Для багатозначного публікатора з application/json за замовчуванням збираємо значення за допомогою Flux#collectToList(), а потім серіалізуємо отриману колекцію.

• Для багатозначного публікатора з потоковим типом середовища передачі даних, таким як application/x-ndjson або application/stream+x-jackson-smile, кодуємо, записуємо та скидаємо кожне значення окремо, використовуючи формат JSON з розмежуванням рядків. У кодувальнику можуть бути зареєстровані інші типи потокових середовищ передачі.

• У разі SSE (подій, що посилаються сервером) Jackson2Encoder викликається для кожної події, а виведення скидається, щоб забезпечити передачу без затримки.

Дані форми

FormHttpMessageReader та FormHttpMessageWriter підтримують декодування та кодування вмісту application/x-www-form-urlencoded.

На стороні сервера, де доступ до вмісту форми часто повинен здійснюватися з декількох місць, ServerWebExchange передбачає спеціальний метод getFormData(), який парсить вміст через FormHttpMessageReader, а потім кешує результат для здійснення повторного доступу.

Після використання getFormData() вихідний сирий вміст більше не можна прочитати з тіла запиту. З цієї причини очікується, що програми будуть послідовно проходити через ServerWebExchange для отримання доступу до кешованих даних форми замість читання з сирого тіла запиту.

Багатокомпонентність

На стороні сервера, де доступ до вмісту багатокомпонентної форми може знадобитися здійснювати з кількох місць, ServerWebExchange передбачає спеціальний метод getMultipartData(), який парсить вміст через MultipartHttpMessageReader, а потім кешує результат для отримання повторного доступу.

Після використання параметра getMultipartData() вихідний сирий вміст більше не можна прочитати з тіла запиту. З цієї причини додаткам потрібно постійно використовувати параметр getMultipartData() для багаторазового доступу до компонентів у вигляді Map, або покладатися на SynchronossPartHttpMessageReader для одноразового доступу до Flux<Part> .

Обмеження

Реалізації Decoder та HttpMessageReader, які буферизують частину або весь потік вступних даних, можуть бути налаштовані з обмеженням на максимальну кількість байт для буферизації в пам'яті. У деяких випадках буферизація відбувається тому, що вступні дані агрегуються і подаються як єдиний об'єкт — наприклад, метод контролера з анотацією @RequestBody byte[], x-www-form-urlencoded дані тощо. Буферизація також може виникати під час потокової передачі, якщо відбувається поділ вступного потоку — наприклад, текст з роздільниками, потік об'єктів у форматі JSON тощо. Для цих випадків потокової передачі обмеження застосовується до кількості байт, пов'язаних з одним об'єктом у потоці.

Для конфігурування розмірів буферів можна перевірити, чи відкриває цей Decoder або HttpMessageReader властивість maxInMemorySize, і якщо так, то в Javadoc будуть міститися подробиці про значення за замовчуванням. На стороні сервера ServerCodecConfigurer передбачає єдине місце, звідки можна встановити всі кодеки. На стороні клієнта обмеження для всіх кодеків можна змінити в WebClient.Builder.

Для багатокомпонентного парсингу властивість maxInMemorySize обмежує розмір нефайлових компонентів. Для файлових компонентів воно визначає поріг, у якому компонент записується на диск. Для файлових компонентів, записаних на диск, існує додаткова властивість maxDiskUsagePerPart, що обмежує обсяг дискового простору для кожного компонента. Існує також властивість maxParts для обмеження кількості компонентів у багатокомпонентному запиті. Щоб конфігурувати всі три компоненти в WebFlux, потрібно вказати попередньо налаштований екземпляр MultipartHttpMessageReader у ServerCodecConfigurer.

Потокова передача

При потоковій передачі даних у HTTP-відповідь (наприклад, text/event-stream, application/x-ndjson) важливо передавати дані періодично, щоб якомога раніше точно виявити відключений клієнт. Так передавати можна лише коментар, порожню SSE-подію або будь-які інші дані "порожні операції", які ефективно послужать як heartbeat-повідомлення.

DataBuffer

DataBuffer — це подання для байтового буфера в WebFlux. Важливо розуміти, що на деяких серверах, таких як Netty, байтові буфери об'єднані в пул і підраховуються за посиланнями, і повинні бути звільнені при споживанні, щоб уникнути витоку пам'яті

У випадку з додатками на WebFlux зазвичай нема потреби турбуватися про такі проблеми, якщо тільки вони не споживають і не виробляють буфери даних безпосередньо замість використання кодеків для перетворення на об'єкти вищого рівня і назад або якщо вони не створюють власні кодеки.

Журналювання

Журналювання на рівні DEBUG у Spring WebFlux спроєктовано так, щоб бути компактним, простим та зручним для людини. Воно зосереджено на найбільш значущих бітах інформації, які будуть використовуватися знову і знову, на відміну від інших, які використовуються тільки при налагодженні конкретної проблеми.

Журналювання на рівні TRACE загалом відповідає тим же принципам, що і DEBUG (і, наприклад, також не повинно бути перевантаженим), але може використовуватися для налагодження будь-якої проблеми. До того ж, деякі повідомлення журналу можуть демонструвати різний рівень деталізації на рівнях TRACE та DEBUG.

Належне журналювання залежить від досвіду використання журналів. Якщо ти помітиш щось, що не відповідає заявленим цілям, будь ласка, повідом нам про це.

Ідентифікатор журналу

У WebFlux один запит може виконуватись у кількох потоках, а ідентифікатор потоку не має жодної користі для зіставлення повідомлень журналу, які стосуються конкретного запиту. Саме тому повідомлення журналу WebFlux за замовчуванням мають префікс із ідентифікатором конкретного запиту.

На стороні сервера ідентифікатор журналу зберігається в атрибуті ServerWebExchange LOG_ID_ATTRIBUTE), а повністю відформатований префікс на основі цього ідентифікатора доступний із ServerWebExchange#getLogPrefix(). На стороні WebClient ідентифікатор журналу зберігається в атрибуті ClientRequest (LOG_ID_ATTRIBUTE), а повністю відформатований префікс доступний з ClientRequest#logPrefix().

Конфіденційні дані

Журнал на рівнях DEBUG та TRACE може реєструвати конфіденційну інформацію. Саме тому параметри та заголовки форм за замовчуванням маскуються, і вам необхідно явно активувати їхнє повне протоколювання.

У наступному прикладі показано, як це зробити для запитів на стороні сервера:

@Configuration
@EnableWebFlux
class MyConfig implements WebFluxConfigurer {
@Override
public void
configureHttpMessageCodecs(ServerCodecConfigurer configurer ) {
configurer.defaultCodecs().enableLoggingRequestDetails(true);
}
}

@Configuration
@EnableWebFlux
class  MyConfig : WebFluxConfigurer {
 override fun  configureHttpMessageCodecs(configurer: ServerCodecConfigurer) {
configurer.defaultCodecs().enableLoggingRequestDetails(true)
}
}

У наступному прикладі показано, як це зробити для запитів на стороні клієнта:

Consumer<ClientCodecConfigurer> ; consumer = configurer ->
configurer.defaultCodecs().enableLoggingRequestDetails(true);
WebClient webClient = WebClient.builder()
.exchangeStrategies(strategies -> strategies.codecs(consumer))
.build();

val consumer: (ClientCodecConfigurer) -> Unit = { configurer -> configurer.defaultCodecs().enableLoggingRequestDetails(true) }
val webClient = WebClient.builder()
.exchangeStrategies( { strategies -> strategies.codecs(consumer) })
.build()

Appenders

Бібліотеки логування, такі як SLF4J та Log4J 2, надають асинхронні логери, які дозволяють уникнути блокування. Хоча вони мають свої недоліки, такі як потенційний пропуск повідомлень, які не можна поставити в чергу на логування, вони є найкращими доступними варіантами для використання в реактивних, неблокуючих програмах.

Кастомні кодеки

Програми можуть реєструвати кастомні кодеки для підтримки додаткових типів середовища передачі даних або специфічної логіки роботи, які не підтримуються стандартними кодеками.

Деякі параметри конфігурації, виражені розробниками, застосовуються до кодеків за замовчуванням. Користувацьким кодекам, напевно, знадобиться можливість узгоджуватися з цими налаштуваннями, наприклад, примусово обмежувати буферизацію або реєструвати конфіденційні дані.

У наступному прикладі показано, як це зробити для запитів на стороні клієнта:

WebClient webClient = WebClient.builder()
.codecs(configurer -> {
        CustomDecoder decoder = new CustomDecoder();
        configurer.customCodecs().registerWithDefaultConfig(decoder);
})
.build();

val webClient = WebClient.builder()
.codecs({ configurer ->
        val decoder = CustomDecoder()
        configurer.customCodecs().registerWithDefaultConfig(decoder)
})
.build()