Могу ли я запустить 2 моно асинхронных ядра реактора?

Question

в java:

    Mono<response> response = mon.just()
    Mono<object> object = mono.just()
    return response.block()

Ответ и объект не зависят друг от друга.Можно ли одновременно запустить 2 моно одновременно?

Answer 1

Если эти 2 Mono s никак не связаны друг с другом и вы хотите запустить их параллельно, я бы порекомендовал вам подумать о вашем дизайне.

---

Если выхотел бы запустить их параллельно и использовать их результат, когда оба результата доступны:

Mono<Integer> source1 = Mono.just(1).subscribeOn(Schedulers.elastic());
Mono<String> source2 = Mono.just("aaaa").subscribeOn(Schedulers.elastic());

Mono.zip(source1, source2, (integer, string) -> string.concat(integer.toString()))
        .subscribe(x -> System.out.println(x));

Вывод:

"aaaa1"

---

Если тип результата совпадает (но не), то вы можете сделать:

Mono<Integer> source1 = Mono.just(1).subscribeOn(Schedulers.elastic());
Mono<Integer> source2 = Mono.just(2).subscribeOn(Schedulers.elastic());

Flux.merge(source1, source2)
        .map(number -> number * 10)
        .subscribe(x -> System.out.println(x));

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

Вывод:

10
20
or
20
10

---

Дополнительная информация о классе Scheduler, доступных планировщиках в реакторе lib и операции подписки.

Answer 2

Существует более одного способа.Одним из простых решений является использование оператора subscribeOn:

Mono<response> response = Mono.just(...).subscribeOn(Schedulers.elastic());
Mono<object> object = Mono.just(...).subscribeOn(Schedulers.elastic());

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

Если вас интересуют реактивные потоки и многопоточность, я недавно написал статья об этом.