Asynchronous

На текущей работе пишем на Reactor. Технология классная, но как всегда есть много НО. Некоторые вещи раздражают, код сложнее писать и читать, с ThreadLocal совсем беда. Решил посмотреть какие проблемы уйдут, если перейти на Kotlin Coroutines, а какие проблемы, наоборот, добавятся.

Карточка пациента

Для статьи написал маленький проект, воспроизведя проблемы, с которыми столкнулся на работе. Основной код здесь. Алгоритм специально не стал разбивать на отдельные методы, так лучше видно проблемы.

В двух словах об алгоритме:

Переводим деньги с одного счёта на другой, записывая транзакции о факте перевода.

Перевод идемпотентен, так что если транзакция уже есть в БД, то отвечаем клиенту, что всё хорошо. При вставке транзакции может вылететь DataIntegrityViolationException, это тоже значит, что транзакция уже есть.

Чтобы не уйти в минус есть проверка в коде + Optimistic lock, который не позволяет конкурентно обновлять счета. Чтобы он работал нужен retry и дополнительная обработка ошибок.

Жалобы пациента

1. Stacktrace не показывает каким образом мы попали в проблемное место.

2. Код явно сложнее, чем был бы на блокирующих технологиях.

3. Многоступенчатая вложенность кода из-за flatMap.

4. Неудобная обработка ошибок и их выброс.

5. Сложная обработка поведения для Mono.empty().

6. Сложности с логированием, если надо в лог добавить что-то глобальное, например traceId. (в статье не описываю, но те же проблемы с другими ThreadLocal переменными, например SpringSecurity)

7. Неудобно дебажить.

8. Неявное api для параллелизации.

Ход лечения

Написал отдельный PR перехода с Java на Kotlin.

Интеграция почти везде гладкая.

Понадобилось добавить com.fasterxml.jackson.module:jackson-module-kotlin чтобы заработала сериализация data классов и org.jetbrains.kotlin.plugin.spring чтобы не прописывать везде open модификаторы.

В контроллере достаточно было написать suspend fun transfer(@RequestBody request: TransferRequest) вместо public Mono<Void> transfer(@RequestBody TransferRequest request)

В репозитории написал suspend fun save(account: Account): Account вместо Mono<Account> save(Account account); Единственное, репозитории не определяются, если в них только suspend функции, надо, чтобы хоть один метод работал с Reactor типами.

Тесты обернул в runBlocking { }, чтобы можно было вызывать suspend функции.

Для реализации Retry использовал библиотеку kotlin-retry. Единственное, в ней не было функции фильтрации по классу ошибки, но это было легко добавить (завёл PR).

Ну и, естественно, переписал алгоритм. Все детали опишу ниже по-отдельности.

Было:

public Mono<Void> transfer(String transactionKey, long fromAccountId,                           long toAccountId, BigDecimal amount) {  return transactionRepository.findByUniqueKey(transactionKey)    .map(Optional::of)    .defaultIfEmpty(Optional.empty())    .flatMap(withMDC(foundTransaction -> {      if (foundTransaction.isPresent()) {        log.warn("retry of transaction " + transactionKey);        return Mono.empty();      }      return accountRepository.findById(fromAccountId)        .switchIfEmpty(Mono.error(new AccountNotFound()))        .flatMap(fromAccount -> accountRepository.findById(toAccountId)          .switchIfEmpty(Mono.error(new AccountNotFound()))          .flatMap(toAccount -> {            var transactionToInsert = Transaction.builder()              .amount(amount)              .fromAccountId(fromAccountId)              .toAccountId(toAccountId)              .uniqueKey(transactionKey)              .build();            var amountAfter = fromAccount.getAmount().subtract(amount);            if (amountAfter.compareTo(BigDecimal.ZERO) < 0) {              return Mono.error(new NotEnoghtMoney());            }            return transactionalOperator.transactional(              transactionRepository.save(transactionToInsert)                .onErrorResume(error -> {                  //transaction was inserted on parallel transaction,                  //we may return success response                  if (error instanceof DataIntegrityViolationException             && error.getMessage().contains("TRANSACTION_UNIQUE_KEY")) {                    return Mono.empty();                  } else {                    return Mono.error(error);                  }                })                .then(accountRepository.transferAmount(                  fromAccount.getId(), fromAccount.getVersion(),                   amount.negate()                ))                .then(accountRepository.transferAmount(                  toAccount.getId(), toAccount.getVersion(), amount                ))            );          }));    }))    .retryWhen(Retry.backoff(3, Duration.ofMillis(1))      .filter(OptimisticLockException.class::isInstance)      .onRetryExhaustedThrow((__, retrySignal) -> retrySignal.failure())    )    .onErrorMap(      OptimisticLockException.class,      e -> new ResponseStatusException(        BANDWIDTH_LIMIT_EXCEEDED,        "limit of OptimisticLockException exceeded", e      )    )    .onErrorResume(withMDC(e -> {      log.error("error on transfer", e);      return Mono.error(e);    }));}

Стало:

suspend fun transfer(transactionKey: String, fromAccountId: Long,                     toAccountId: Long, amount: BigDecimal) {  try {    try {      retry(limitAttempts(3) + filter { it is OptimisticLockException }) {        val foundTransaction = transactionRepository          .findByUniqueKey(transactionKey)        if (foundTransaction != null) {          logger.warn("retry of transaction $transactionKey")          return@retry        }        val fromAccount = accountRepository.findById(fromAccountId)          ?: throw AccountNotFound()        val toAccount = accountRepository.findById(toAccountId)          ?: throw AccountNotFound()        if (fromAccount.amount - amount < BigDecimal.ZERO) {          throw NotEnoghtMoney()        }        val transactionToInsert = Transaction(          amount = amount,          fromAccountId = fromAccountId,          toAccountId = toAccountId,          uniqueKey = transactionKey        )        transactionalOperator.executeAndAwait {          try {            transactionRepository.save(transactionToInsert)          } catch (e: DataIntegrityViolationException) {            if (e.message?.contains("TRANSACTION_UNIQUE_KEY") != true) {              throw e;            }          }          accountRepository.transferAmount(            fromAccount.id!!, fromAccount.version, amount.negate()          )          accountRepository.transferAmount(            toAccount.id!!, toAccount.version, amount          )        }      }    } catch (e: OptimisticLockException) {      throw ResponseStatusException(        BANDWIDTH_LIMIT_EXCEEDED,         "limit of OptimisticLockException exceeded", e      )    }  } catch (e: Exception) {    logger.error(e) { "error on transfer" }    throw e;  }}

Stacktraces

Пожалуй, это самое главное.

Было:

o.s.w.s.ResponseStatusException: 509 BANDWIDTH_LIMIT_EXCEEDED "limit of OptimisticLockException exceeded"; nested exception is c.g.c.v.r.OptimisticLockExceptionat c.g.c.v.r.services.Ledger.lambda$transfer$5(Ledger.java:75)...Caused by: c.g.c.v.r.OptimisticLockException: nullat c.g.c.v.r.repos.AccountRepositoryImpl.lambda$transferAmount$0(AccountRepositoryImpl.java:27)at r.c.p.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:125)  ...

Стало:

error on transfer o.s.w.s.ResponseStatusException: 509 BANDWIDTH_LIMIT_EXCEEDED "limit of OptimisticLockException exceeded"; nested exception is c.g.c.v.r.OptimisticLockExceptionat c.g.c.v.r.services.Ledger.transfer$suspendImpl(Ledger.kt:70)at c.g.c.v.r.services.Ledger$transfer$1.invokeSuspend(Ledger.kt)...Caused by: c.g.c.v.r.OptimisticLockException: nullat c.g.c.v.r.repos.AccountRepositoryImpl.transferAmount(AccountRepositoryImpl.kt:24)...at c.g.c.v.r.services.Ledger$transfer$3$1.invokeSuspend(Ledger.kt:65)at c.g.c.v.r.services.Ledger$transfer$3$1.invoke(Ledger.kt)at o.s.t.r.TransactionalOperatorExtensionsKt$executeAndAwait$2$1.invokeSuspend(TransactionalOperatorExtensions.kt:30)(Coroutine boundary)at o.s.t.r.TransactionalOperatorExtensionsKt.executeAndAwait(TransactionalOperatorExtensions.kt:31)at c.g.c.v.r.services.Ledger$transfer$3.invokeSuspend(Ledger.kt:56)at com.github.michaelbull.retry.RetryKt$retry$3.invokeSuspend(Retry.kt:38)at c.g.c.v.r.services.Ledger.transfer$suspendImpl(Ledger.kt:35)at c.g.c.v.r.controllers.LedgerController$transfer$2$1.invokeSuspend(LedgerController.kt:20)at c.g.c.v.r.controllers.LedgerController$transfer$2.invokeSuspend(LedgerController.kt:19)at kotlin.reflect.full.KCallables.callSuspend(KCallables.kt:55)at o.s.c.CoroutinesUtils$invokeSuspendingFunction$mono$1.invokeSuspend(CoroutinesUtils.kt:64)(Coroutine creation stacktrace)at k.c.i.IntrinsicsKt__IntrinsicsJvmKt.createCoroutineUnintercepted(IntrinsicsJvm.kt:122)at k.c.i.CancellableKt.startCoroutineCancellable(Cancellable.kt:30)...Caused by: c.g.c.v.r.OptimisticLockException: nullat c.g.c.v.r.repos.AccountRepositoryImpl.transferAmount(AccountRepositoryImpl.kt:24)...at c.g.c.v.r.services.Ledger$transfer$3$1.invokeSuspend(Ledger.kt:65)at c.g.c.v.r.services.Ledger$transfer$3$1.invoke(Ledger.kt)at o.s.t.r.TransactionalOperatorExtensionsKt$executeAndAwait$2$1.invokeSuspend(TransactionalOperatorExtensions.kt:30)...

Скучные части стектрейсов я вырезал, пакеты сократил (забочусь о читателе, и без того длинно).

В Java очень куцая информация. Да, ошибка есть. Даже видно на какой строчке она вылетела. Только непонятно а как мы в эту строчку кода попали. В Kotlin версии виден весь трейс от контроллера.

Вот представьте себе, что вы видите ошибку в логе где-то на обращении в регулярно вызываемый метод. А кто его вызывал? Придётся по логам рядом искать. Это хорошо, если логи объединены через что-нибудь вроде traceId (thread name нам не поможет) и вообще логи есть.

Сложность кода

Kotlin версия выглядит проще, точно так же, как выглядел бы код на блокирующих технологиях. По крайне мере пока мы не вводим параллельные операции (тут ещё вопрос какой вариант сложнее писать: на блокирующих технологиях или с корутинами).

Многоступенчатая вложенность кода

Никаких flatMap. Добавились вложения из-за явных try catch, но схожая логика вся объявлена на одном уровне.

Было:

return accountRepository.findById(fromAccountId)  .switchIfEmpty(Mono.error(new AccountNotFound()))  .flatMap(fromAccount -> accountRepository.findById(toAccountId)    .switchIfEmpty(Mono.error(new AccountNotFound()))    .flatMap(toAccount -> {      ...    })

Стало:

val fromAccount = accountRepository.findById(fromAccountId)  ?: throw AccountNotFound()val toAccount = accountRepository.findById(toAccountId)  ?: throw AccountNotFound()...

Обработка ошибок и их выброс

Обработка ошибок теперь через обычный try catch, легко понять какой кусок кода мы обернули.

Было:

return transactionRepository.findByUniqueKey(transactionKey)  ...  .onErrorMap(    OptimisticLockException.class,    e -> new ResponseStatusException(      BANDWIDTH_LIMIT_EXCEEDED,       "limit of OptimisticLockException exceeded", e    )  )

Стало:

try {  val foundTransaction = transactionRepository    .findByUniqueKey(transactionKey)  ...} catch (e: OptimisticLockException) {  throw ResponseStatusException(    BANDWIDTH_LIMIT_EXCEEDED,     "limit of OptimisticLockException exceeded", e  )}

Ошибки выбрасывать можно просто через throw, а не возвращая объект ошибки. В Reactor меня особенно раздражают конструкции вида:

.flatMap(foo -> {  if (foo.isEmpty()) {     return Mono.error(new IllegalStateException());  } else {    return Mono.just(foo);  }})

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

Mono.empty()

Это заслуживает отдельного обсуждения. В реактор нельзя передавать null в качестве результата. При этом нельзя написать C5C.

Ide никак тебе не подсказывает, может ли этот конкретный mono быть пустым и обработал ли ты это. В рабочем проекте для меня именно это было источником тупых ошибок. То забудешь обработать, то после рефакторинга как-то не так работает.

В Kotlin будет not null тип, если ты точно знаешь, что результат будет. Или это будет nullable тип и компилятор обяжет тебя что-то с этим сделать.

Конкретно на нашем примере:

Было:

return transactionRepository.findByUniqueKey(transactionKey)  .map(Optional::of)  .defaultIfEmpty(Optional.empty())  .flatMap(foundTransaction -> {    if (foundTransaction.isPresent()) {      log.warn("retry of transaction " + transactionKey);      return Mono.empty();    }...

Стало:

val foundTransaction = transactionRepository  .findByUniqueKey(transactionKey)if (foundTransaction != null) {  logger.warn("retry of transaction $transactionKey")  return@retry}...

Может, как-то можно эту логику написать адекватнее на Reactor, но то что я нагуглил выглядит ещё хуже.

Логирование и контекст

Допустим, мы хотим всегда логировать traceId во время обработки запроса. ThreadLocal больше не работает, в том числе и MDC (контекст для логирования). Что делать?

Есть контекст. И в Reactor и в Coroutines контекст immutable, так что новое значение в MDC подбрасывать будет не так просто (нужно пересоздавать контекст).

Чтобы работало в Java надо написать фильтр, который сохранит traceId в контекст:

@Componentpublic class TraceIdFilter implements WebFilter {  @Override  public Mono<Void> filter(    ServerWebExchange exchange, WebFilterChain chain  ) {    var traceId = Optional.ofNullable(      exchange.getRequest().getHeaders().get("X-B3-TRACEID")    )      .orElse(Collections.emptyList())      .stream().findAny().orElse(UUID.randomUUID().toString());    return chain.filter(exchange)      .contextWrite(context ->        LoggerHelper.addEntryToMDCContext(context, "traceId", traceId)      );  }}

И каждый раз, когда мы хотим что-то залогировать, надо переносить traceId из контекста в MDC:

public static <T, R> Function<T, Mono<R>> withMDC(  Function<T, Mono<R>> block) {  return value -> Mono.deferContextual(context -> {    Optional<Map<String, String>> mdcContext = context      .getOrEmpty(MDC_ID_KEY);    if (mdcContext.isPresent()) {      try {        MDC.setContextMap(mdcContext.get());        return block.apply(value);      } finally {        MDC.clear();      }    } else {      return block.apply(value);    }  });}

Да, это опять Mono. Т.е. мы можем логировать только тогда, когда код позволяет вернуть Mono. Например вот так:

.onErrorResume(withMDC(e -> {  log.error("error on transfer", e);  return Mono.error(e);}))

В Kotlin проще. Нужно написать фильтр, чтобы сохранить traceId сразу в MDC:

@Componentclass TraceIdFilter : WebFilter {  override fun filter(    exchange: ServerWebExchange, chain: WebFilterChain  ): Mono<Void> {    val traceId = exchange.request.headers["X-B3-TRACEID"]?.first()     MDC.put("traceId", traceId ?: UUID.randomUUID().toString())    return chain.filter(exchange)  }}

И при создании корутины вызывать withContext(MDCContext()) { }

Каждый раз на время работы корутины, будет восстанавливаться MDC и в логах будет traceId. Во время записи в лог ни о чем задумываться не нужно.

Тут есть одно НО, об этом позже.

Дебаг

В Java Reactor проблемы с дебагом примерно те же, что и со стектрейсом: нужно думать о том, кто в каком порядке кого вызвал, ставить breakpoints в лямбдах и т.д..

С корутинами не замечаешь разницы с обычным кодом: работает stepOver, переменные подсвечиваются, видно стектрейс и по нему можно перемещаться (в нашем примере вплоть до контроллера).

Всё идеально, кроме возможности запускать suspend функции во время дебага. На это уже есть issue. Правда, надо сказать, что и в Java Reactor особо не получается в evaluate сделать то, что хочется.

Параллелизация

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

Было:

return Mono.zip(  transactionRepository.findByUniqueKey(transactionKey)    .map(Optional::of)    .defaultIfEmpty(Optional.empty()),  accountRepository.findById(fromAccountId)    .switchIfEmpty(Mono.error(new AccountNotFound())),  accountRepository.findById(toAccountId)    .switchIfEmpty(Mono.error(new AccountNotFound())),).flatMap(withMDC(fetched -> {  var foundTransaction = fetched.getT1();  var fromAccount = fetched.getT2();  var toAccount = fetched.getT3();  if (foundTransaction.isPresent()) {    log.warn("retry of transaction " + transactionKey);    return Mono.empty();  }  ...}

Стало:

val foundTransactionAsync = GlobalScope.async(coroutineContext) {  logger.info("async fetch of transaction $transactionKey")  transactionRepository.findByUniqueKey(transactionKey)}val fromAccountAsync = GlobalScope.async(coroutineContext) {   accountRepository.findById(fromAccountId) }val toAccountAsync = GlobalScope.async(coroutineContext) {   accountRepository.findById(toAccountId) }if (foundTransactionAsync.await() != null) {  logger.warn("retry of transaction $transactionKey")  return@retry}val fromAccount = fromAccountAsync.await() ?: throw AccountNotFound()val toAccount = toAccountAsync.await() ?: throw AccountNotFound()

В Kotlin версии есть явное указание вот это выполни асинхронно, вместо выполни всё это в параллель в Reactor.

Что самое важное, код ведёт себя по-разному. В случае с Reactor мы создаем три параллельных запроса и продолжаем работу только после того, как все три завершатся. С корутинами мы запускаем все три запроса и ждать чего-то начинаем только при вызове foundTransactionAsync.await(). Таким образом, если transactionRepository.findByUniqueKey() выполнится быстрее, то мы завершим обработку, без ожидания accountRepository.findById() (эти операции отменятся).

В более сложных алгоритмах разница будет ощутимей. Не уверен, что вообще в Reactor получится написать такую версию алгоритма:

val foundTransactionAsync = GlobalScope.async(coroutineContext) {  logger.info("async fetch of transaction $transactionKey")  transactionRepository.findByUniqueKey(transactionKey)}val fromAccountAsync = GlobalScope.async(coroutineContext) {  accountRepository.findById(fromAccountId)}val toAccountAsync = GlobalScope.async(coroutineContext) {  accountRepository.findById(toAccountId)}if (foundTransactionAsync.await() != null) {  logger.warn("retry of transaction $transactionKey")  return@retry}val transactionToInsert = Transaction(  amount = amount,  fromAccountId = fromAccountId,  toAccountId = toAccountId,  uniqueKey = transactionKey)transactionalOperator.executeAndAwait {  try {    transactionRepository.save(transactionToInsert)  } catch (e: DataIntegrityViolationException) {    if (e.message?.contains("TRANSACTION_UNIQUE_KEY") != true) {      throw e;    }  }  val fromAccount = fromAccountAsync.await() ?: throw AccountNotFound()  val toAccount = toAccountAsync.await() ?: throw AccountNotFound()  if (fromAccount.amount - amount < BigDecimal.ZERO) {    throw NotEnoghtMoney()  }  accountRepository.transferAmount(    fromAccount.id!!, fromAccount.version, amount.negate()  )  accountRepository.transferAmount(    toAccount.id!!, toAccount.version, amount  )}

Здесь я ожидаю асинхронный запрос уже внутри открытой БД транзакции. Т.е. мы при работе с одним коннектом, ждём результата выполнения на другом коннекте. Так делать не стоит, скорее для примера, хоть оно и работает (пока коннектов хватает).

Побочные эффекты

Конечно, есть проблемы, куда без них.

Надо явно указывать context и scope

Чтобы программа работала как ожидается, надо:

1. Каждому запросу назначить scope. Таким образом все порождаемые при обработке запроса корутины будут отменены все вместе, например, в случае ошибки.

2. В каждом запросе проставить context. Зачем нам нужен контекст я рассказывал в разделе про логирование.

Spring не берет на себя заботу об этом вопросе, приходится в контроллере указывать явно:

@PutMapping("/transfer")suspend fun transfer(@RequestBody request: TransferRequest) {  coroutineScope {    withContext(MDCContext()) {      ledger.transfer(request.transactionKey, request.fromAccountId,                       request.toAccountId, request.amount)    }  }}

Естественно, можно объединить в одну функцию и потом через regexp проверить, что во всех методах есть нужный вызов. Но лучше была бы какая-то автоматизация для этого.

Передача context

В случае, если мы порождаем дополнительные корутины, необходимо явным образом передать текущий контекст. Вот пример:

val foundTransactionAsync = GlobalScope.async(coroutineContext) {  logger.info("async fetch of transaction $transactionKey")  transactionRepository.findByUniqueKey(transactionKey)}

По-умолчанию в async() указывать контекст не требуется (он будет пуст). Видимо, чтобы было меньше неявных вещей. Но как итог, в нашем случае, если забыть передать контекст, в логе не окажется traceId. Помните я писал, что не надо задумываться во время логирования? Вместо этого надо задумываться при создании корутины (что, конечно, лучше).

AOP и suspend

Автоматизацию, которую я упоминал в первом пункте, написать самому сложно. Потому что пока нельзя нормально написать aspect для suspend функции.

Я в итоге сумел написать такой аспект. Но для объяснения того, как это работает понадобится отдельная статья.

Надеюсь, появится более адекватный способ писать аспекты (попробую этому посодействовать).

Оценка лечения

Все проблемы исчезли. Добавилась пара новых, но оно терпимо.

Надо сказать, что корутины быстро развиваются и я ожидаю только улучшения работы с ними.

Видно, что команда JetBrains внимательно отнеслась к проблемам разработчиков. Насколько я знаю, где-то год назад всё ещё были проблемы с дебагом и стактрейсом, к примеру.

Самое главное, с корутинами не надо в голове держать все особенности работы Reactor и его могучий API. Ты просто пишешь код.

В преддверии старта курса Разработчик на Spring Framework подготовили традиционный перевод полезного материала.

Также абсолютно бесплатно предлагаем посмотреть запись демо-урока на тему Введение в облака, создание кластера в Mongo DB Atlas.

WebClient это неблокирующий, реактивный клиент для выполнения HTTP-запросов.

Время RestTemplate подошло к концу

Возможно, вы слышали, что время RestTemplate на исходе. Теперь это указано и в официальной документации:

NOTE: As of 5.0 this class is in maintenance mode, with only minor requests for changes and bugs to be accepted going forward. Please, consider using the org.springframework.web.reactive.client.WebClient which has a more modern API and supports sync, async, and streaming scenarios.

ПРИМЕЧАНИЕ: Начиная с версии 5.0, этот класс законсервирован и в дальнейшем будут приниматься только минорные запросы на изменения и на исправления багов. Пожалуйста, подумайте об использовании org.springframework.web.reactive.client.WebClient, который имеет более современный API и поддерживает синхронную, асинхронную и потоковую передачи.

Конечно, мы понимаем, что RestTemplate не исчезнет мгновенно, однако новой функциональности в нем уже не будет. По этой причине в Интернете можно видеть десятки вопросов о том, что такое WebClient и как его использовать. В этой статье вы найдете советы по использованию новой библиотеки.

Отличия между WebClient и RestTemplate

Если в двух словах, то основное различие между этими технологиями заключается в том, что RestTemplate работает синхронно (блокируя), а WebClient работает асинхронно (не блокируя).

RestTemplate это синхронный клиент для выполнения HTTP-запросов, он предоставляет простой API с шаблонным методом поверх базовых HTTP-библиотек, таких как HttpURLConnection (JDK), HttpComponents (Apache) и другими.

Spring WebClient это асинхронный, реактивный клиент для выполнения HTTP-запросов, часть Spring WebFlux.

Вам, вероятно, интересно, как можно заменить синхронного клиента на асинхронный. У WebClient есть решение этой задачи. Мы рассмотрим несколько примеров использования WebClient.

А сейчас настало время попрощаться с RestTemplate , сказать ему спасибо и продолжить изучение WebClient.

Начало работы с WebClient

Предварительные условия

• Spring Boot 2

• JDK 11

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

Давайте создадим базовый проект с зависимостями, используя Spring Initializr.

Теперь взглянем на зависимости нашего проекта. Самая важная для нас зависимость spring-boot-starter-webflux.

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-webflux</artifactId> </dependency>

Spring WebFlux является частью Spring 5 и обеспечивает поддержку реактивного программирования для веб-приложений.

Пришло время настроить WebClient.

Настройка WebClient

Есть несколько способов настройки WebClient. Первый и самый простой создать его с настройками по умолчанию.

WebClient client = WebClient.create();

Можно также указать базовый URL:

WebClient client = WebClient.create("http://base-url.com");

Третий и самый продвинутый вариант создать WebClient с помощью билдера. Мы будем использовать конфигурацию, которая включает базовый URL и таймауты.

@Configurationpublic class WebClientConfiguration {    private static final String BASE_URL = "https://jsonplaceholder.typicode.com";    public static final int TIMEOUT = 1000;    @Bean    public WebClient webClientWithTimeout() {        final var tcpClient = TcpClient                .create()                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, TIMEOUT)                .doOnConnected(connection -> {                    connection.addHandlerLast(new ReadTimeoutHandler(TIMEOUT, TimeUnit.MILLISECONDS));                    connection.addHandlerLast(new WriteTimeoutHandler(TIMEOUT, TimeUnit.MILLISECONDS));                });        return WebClient.builder()                .baseUrl(BASE_URL)                .clientConnector(new ReactorClientHttpConnector(HttpClient.from(tcpClient)))                .build();    }}

Параметры, поддерживаемые WebClient.Builder можно посмотреть здесь.

Подготовка запроса с помощью Spring WebClient

WebClient поддерживает методы: get(), post(), put(), patch(), delete(), options() и head().

Также можно указать следующие параметры:

• Переменные пути (path variables) и параметры запроса с помощью метода uri().

• Заголовки запроса с помощью метода headers().

• Куки с помощью метода cookies().

После указания параметров можно выполнить запрос с помощью retrieve() или exchange(). Далее мы преобразуем результат в Mono с помощью bodyToMono() или во Flux с помощью bodyToFlux().

Асинхронный запрос

Давайте создадим сервис, который использует бин WebClient и создает асинхронный запрос.

@Service@AllArgsConstructorpublic class UserService {    private final WebClient webClient;    public Mono<User> getUserByIdAsync(final String id) {        return webClient                .get()                .uri(String.join("", "/users/", id))                .retrieve()                .bodyToMono(User.class);    }}

Как вы видите, мы не сразу получаем модель User. Вместо User мы получаем Mono-обертку, с которой выполняем различные действия. Давайте подпишемся неблокирующим способом, используя subscribe().

userService  .getUserByIdAsync("1")  .subscribe(user -> log.info("Get user async: {}", user));

Выполнение продолжается сразу без блокировки на методе subscribe(), даже если для получения значения будет требоваться некоторое время.

Синхронный запрос

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

public User getUserByIdSync(final String id) {    return webClient            .get()            .uri(String.join("", "/users/", id))            .retrieve()            .bodyToMono(User.class)            .block();}

Здесь поток блокируется, пока запрос не выполнится. В этом случае мы получаем запрашиваемую модель сразу же после завершения выполнения метода.

Повторные попытки

Мы все знаем, что сетевой вызов не всегда может быть успешным. Но мы можем перестраховаться и в некоторых случаях выполнить его повторно. Для этого используется метод retryWhen(), который принимает в качестве аргумента класс response.util.retry.Retry.

public User getUserWithRetry(final String id) {    return webClient            .get()            .uri(String.join("", "/users/", id))            .retrieve()            .bodyToMono(User.class)            .retryWhen(Retry.fixedDelay(3, Duration.ofMillis(100)))            .block();}

С помощью билдера можно настроить параметры и различные стратегии повтора (например, экспоненциальную). Если вам нужно повторить успешную попытку, то используйте repeatWhen() или repeatWhenEmpty() вместо retryWhen().

Обработка ошибок

В случае ошибки, когда повторная попытка не помогает, мы все еще можем контролировать ситуацию с помощью резервного варианта. Доступны следующие методы:

• doOnError() срабатывает, когда Mono завершается с ошибкой.

• onErrorResume() при возникновении ошибки подписывается на резервного издателя, используя функцию для выбора действия в зависимости от ошибки.

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

public User getUserWithFallback(final String id) {    return webClient            .get()            .uri(String.join("", "/broken-url/", id))            .retrieve()            .bodyToMono(User.class)            .doOnError(error -> log.error("An error has occurred {}", error.getMessage()))            .onErrorResume(error -> Mono.just(new User()))            .block();}

В некоторых ситуациях может быть полезно отреагировать на конкретный код ошибки. Для этого можно использовать метод onStatus().

public User getUserWithErrorHandling(final String id) {  return webClient          .get()          .uri(String.join("", "/broken-url/", id))          .retrieve()              .onStatus(HttpStatus::is4xxClientError,                      error -> Mono.error(new RuntimeException("API not found")))              .onStatus(HttpStatus::is5xxServerError,                      error -> Mono.error(new RuntimeException("Server is not responding")))          .bodyToMono(User.class)          .block();}

Клиентские фильтры

Для перехвата и изменения запроса можно настроить фильтры через билдер WebClient .

WebClient.builder()  .baseUrl(BASE_URL)  .filter((request, next) -> next          .exchange(ClientRequest.from(request)                  .header("foo", "bar")                  .build()))  .clientConnector(new ReactorClientHttpConnector(HttpClient.from(tcpClient)))  .build();

Ниже приведен пример фильтра для базовой аутентификации с помощью статического фабричного метода.

WebClient.builder()  .baseUrl(BASE_URL)  .filter(basicAuthentication("user", "password")) // org.springframework.web.reactive.function.client.basicAuthentication()  .clientConnector(new ReactorClientHttpConnector(HttpClient.from(tcpClient)))  .build();

Заключение

В этой статье мы узнали, как настроить WebClient и выполнять синхронные и асинхронные HTTP-запросы. Все фрагменты кода, упомянутые в статье, можно найти в GitHub-репозитории. Документацию по Spring WebClient вы можете найти здесь.

Подводя итог, мы видим, что WebClient прост в использовании и включает в себя все необходимые функции, необходимые в современном программировании.

Удачи с новым Spring WebClient!

Узнать подробнее о курсе Разработчик на Spring Framework.

Посмотреть запись демо-урока на тему Введение в облака, создание кластера в Mongo DB Atlas.