Лечим Java Reactor при помощи Kotlin Coroutines

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

Очень помагает ReactorDebugAgent.init(); Все становится красивым и понятным.

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

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

Вот такой подход он неверный, тк он не реактивный.

transactionRepository.findByUniqueKey(transactionKey)
    .map(Optional::of)
    .defaultIfEmpty(Optional.empty())

зачем вы возвращаете монаду Optional<Something> чтобы завернуть ее потом в Mono<Optional<Something>> — если у вас уже реактивный стек так почему не так

public Mono<Something> findByUniqueKey(...) {...}

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

Опять же — бейте на слои, классы и функции чтобы все было плоским, как то так .flatMap(this::createRequest) а уже вложеная логика идет в createRequest и в результате получается понятно и плоско.

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

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

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

Ничуть не сложнее чем Optional.empty + заставляет вас убирать null тем самым код становится менее бажный.

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

Достаточно один раз написать свой паблишер, который добавить в reactor.core.publisher.Hooks#onEachOperator(Function) и он будет пробрасывать в контекст все что вам нужно и не захламлять бизнес код работой с MDC.

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

Опять же, проблема в ваших простынях. Такой код и на Java Streams будет тяжело понимать и дебажить.

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

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

Мне нужно выйти из метода без ошибки, если вернулся пустой Mono.
Буду благодарен, если расскажете как это сделать без Optional

Вместо такого кода

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();
  }
  ...
}

я бы сделал отдельный обьект для хранения информации:

@Value
@Builder
public class TransactionOperation {
    Account from;
    Account to;
    Transaction  transaction;
}

и тогда код стал бы

Mono.just(TransactionOperation.builder())
                .flatMap(builder -> Flux.merge(
                        transactionRepository.findByUniqueKey(transactionKey)
                                .doOnNext(builder::transaction),
                        accountRepository.findById(fromAccountId)
                                .doOnNext(builder::from),
                        accountRepository.findById(toAccountId)
                                .doOnNext(builder::to)
                ).last().map(ignored -> builder.build()))
                .filter(this::validate)
                .map(ops -> {...});
...
    private boolean validate(@NonNull TransactionOperation ops) {
        if (null == ops.getFrom()) {
            throw new AccountNotFound();
        }
        if (null == ops.getTo()) {
            throw new AccountNotFound();
        }
        if (null != ops.getTransaction()) {
            log.warn("retry of transaction " + transactionKey);
            return false;
        }
        return true;
    }

Буквально на этой неделе нашёл у себя багу, что не все ошибки из метода логировались, как >раз потому что не сразу видно на что навешен обработчик.

я обычно ставлю в самом конце цепочти doOnError(err -> log.error("Unexpected exception", err))

Спасибо за статью со Spring и Coroutines!

GlobalScope.async

Вот так не надо, используйте structured concurrency и код станет короче (не нужно будет писать GlobalScope), контекст не нужно будет явно передавать.

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

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

Если используете logback, обратите внимание на вот эту статью — https://habr.com/ru/post/524130/. Корутины подчас генерят исключения с циклами, что может быть ВНЕЗАПНЫМ сюрпризом.

Чуть больше информации (специфично про корутины) — здесь