Блог

Предположим, что простая реактивная строка потока это:

         Flux.just(1, 2, 3, 4, 5)
                .flatMap(i -> 
                        Mono.just(i)
                                .map( ... map some value)
                                .flatMap( ... async call)
                                .map(... more mappings)
                )
                .doOnNext(log::info)
                .blockLast()
        ;
  

Предположим, что Mono внутри родительского потока .flatMap() представляет собой сложный реактивный поток со сложной логикой. Есть ли способ сохранить этот Mono как переменную, которая может быть передана родительскому потоку? Другими словами, что-то вроде этого:

 final Function<Integer, Publisher<String>> monoPublisher = Mono.something()
                .map( ... map some value)
                .flatMap( ... async call)
                .map(... more mappings);

        Flux.just(1, 2, 3, 4, 5)
                .flatMap(monoPublisher)
                .doOnNext(log::info)
                .blockLast()
        ;
  

Наивный ответ был бы:

 final Function<Integer, Publisher<String>> monoPublisher = i -> Mono.just(i) ...
  

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

Я знаком с Mono.create() , но я не вижу, как заставить его работать здесь, не создавая собственную Publisher реализацию, которая обтекает приемник, возвращаемый Mono.create()

РЕДАКТИРОВАТЬ: Спасибо ESala за предложение. Я просто изменил его пример, чтобы использовать ThreadLocal вместо Queue , что не требуется для этого примера с использованием Flux.just(), но предотвратило бы условия гонки при использовании многопоточного производителя

 final ThreadLocal<Integer> threadLocalStorage = new ThreadLocal<>();

        final Mono<String> mono = Mono.fromSupplier(threadLocalStorage::get)
                .map(Objects::toString);

        Flux.just(1, 2, 3, 4, 5)
                .flatMap(i -> {
                    threadLocalStorage.set(i);
                    return mono;
                })
                .doOnNext(System.out::println)
                .blockLast()
        ;
  

Может быть, что-то подобное сработает, просто замените на LinkedList что-то более причудливое, поддерживающее параллельный доступ:

 Queue<Integer> arguments = new LinkedList<>();

Publisher<String> expensivePublisher =
        Mono.fromSupplier(arguments::remove)
                .map( ... map some value)
                .flatMap( ... async call)
                .map(... more mappings);

Flux.just(1, 2, 3)
        .flatMap(
                i -> {
                    arguments.add(i);
                    return expensivePublisher;
                })
        .blockLast();
  

Но это все равно похоже на взлом. Если во внутреннем действительно есть 100 операторов, тогда я бы предложил попытаться оптимизировать с этой стороны. Улучшения в этой области, вероятно, окажут большее влияние, чем повторное Publisher использование, чтобы избежать его воссоздания.