Многопоточность в Reactor 2.0 — почему я не могу передавать сигналы в несколько потоков

Шаблон в Reactor2.0 заключается в использовании отдельных CachedDispatchers (RingBufferDispatchers) вместо использования WorkQueueDispatcher или ThreadPoolExecutorDispatcher. Это предлагает разделение потока на разные потоки. Это работает только для распараллеливания небольших (обычно неблокирующих) операций в логические потоки.

Для полного неблокирующего выполнения в отдельном пуле потоков см. мою правку внизу.

Вы захотите использовать Stream.groupBy() или Stream.partition(), чтобы разделить поток на несколько потоков, каждый из которых может быть отправлен в отдельные потоки.

1. Stream.groupBy() — разделяет потоки на основе возвращаемого вами ключа.
2. Stream.partition([int]) — сегменты в потоки на основе хэш-кода потоковых объектов (сигналов) и, необязательно, в указанное вами количество сегментов.

После того, как вы разделили/сгруппировали поток на отдельные потоки, вы можете использовать flatMap(), чтобы указать вашим новым потокам отправляться в отдельные потоки.

.flatMap(stream -> stream.dispatchOn(Environment.cachedDispatcher())

Вызов Environment.cachedDispatcher() на самом деле означает получение CachedDispatcher (RingBufferDispatcher) из их пула. Пул по умолчанию имеет размер, эквивалентный количеству процессоров вашего компьютера. CachedDispatcher создаются лениво, поэтому это не идеально делать внутри вашего вызова .dispatchOn(Dispatcher).

Чтобы заранее создать пул (DispatcherSupplier) из CachedDispatchers, вы можете использовать Environment.newCachedDispatchers(int, [groupName]). Вот пример из документации ProjectReactor о том, как все это объединить:

    DispatcherSupplier supplier1 = Environment.newCachedDispatchers(2, "groupByPool");
    DispatcherSupplier supplier2 = Environment.newCachedDispatchers(5, "partitionPool");

    Streams
        .range(1, 10)
        .groupBy(n -> n % 2 == 0) 
        .flatMap(stream -> stream
                .dispatchOn(supplier1.get()) 
                .log("groupBy")
        )
        .partition(5) 
        .flatMap(stream -> stream
                .dispatchOn(supplier2.get()) 
                .log("partition")
        )
        .dispatchOn(Environment.sharedDispatcher()) 
        .log("join")
        .consume();

Документация Reactor: разделение потоков на отдельные потоки

Обратите внимание, что в этом примере они также вызвали .dispatchOn(Environment.sharedDispatcher()) после последнего вызова flatMap(). Это делается для того, чтобы просто объединить поток обратно в один поток Dispatcher. В данном случае Environment.sharedDispatcher(), что является еще одним RingBufferDispatcher.

Я использую эту стратегию, чтобы разделить мой поток на отдельные потоки, чтобы параллельно выполнять блокирующие вызовы на отдельных CachedDispatcher, а затем снова объединять их в один поток на основном Environment.sharedDispatcher(), все неблокирующим/реактивным способом, например так:

    // Spring's RestTemplate for making simple REST (HTTP) calls
    RestTemplate restTemplate = new RestTemplate();
    List<String> urls = Arrays.asList("http://www.google.com", "http://www.imgur.com");
    Streams
        .from(urls)
        .groupBy(s -> s) // unique group (stream) per URL
        .flatMap(stream -> 
            // dispatch on separate threads per stream
            stream.dispatchOn(Environment.cachedDispatcher())
                 // blocking call in separate dispatching thread
                 .map(s -> restTemplate.getForObject(s, String.class)))
        // rejoin into single dispatcher thread
        .dispatchOn(Environment.sharedDispatcher())
        .consume(
             s -> { // consumer
                 System.out.println("---complete in stream---");
             },
             t -> { // error consumer
                 System.out.println("---exception---");
                 System.out.println(t);
             },
             s -> { // complete consumer
                 latch.countDown();
                 System.out.println("---complete---");
             });

EDIT: Для параллельного выполнения блокирующих операций вы все равно можете использовать пул потоков. В Reactor 2.0 вы можете сделать это с помощью диспетчера, поддерживаемого пулом потоков, предпочтительно Executors.newCachedThreadPool(), потому что он будет повторно использовать потоки, ограничивая количество нагрузки на GC.

Самый простой способ сделать это, который я нашел, — это использовать EventBus с ThreadPoolExecutorDispatcher, поддерживаемым Executors.newCachedThreadPool() (который на самом деле просто ThreadPoolExecutor с несколькими специфическими настройками, если вы посмотрите на функцию для создания newCachedThreadPool):

    // dispatcher backed by cached thread pool for limited GC pressure
    Dispatcher dispatcher = new ThreadPoolExecutorDispatcher(1024, 1024, Executors.newCachedThreadPool());

    int n = 1000000;
    EventBus bus = EventBus.create(dispatcher);
    CountDownLatch latch = new CountDownLatch(n);
    bus.on($("topic"), (Event<Integer> ev) -> {
        blockingServiceCall(100); // block for 100ms
        latch.countDown();
        if (ev.getData() % 1000 == 0) {
            System.out.println(Thread.currentThread() + " " + Thread.activeCount());
        }
    });

    long start = System.currentTimeMillis();
    for(int i = 0; i < n; i++) {
        bus.notify("topic", Event.wrap(i));
    }
    latch.await();
    System.out.println("ops/sec: " + (n * 1.0) / ((System.currentTimeMillis() - start) / 1000.0) );

Вам нужно будет отправить обратно в Stream после завершения блокирующего вызова внутри потребителя, если вы хотите вернуться к использованию Streams. Насколько я знаю, вы не можете просто автоматически объединить их обратно в поток, а также не можете использовать ThreadPoolExecutorDispatcher, поддерживаемый кэшированным пулом потоков, напрямую, выполнив Stream.dispatchOn(threadPoolExecutorDispatcher).

Просто просканировав, вы получите правильный ответ как можно скорее. Но сначала проблема не связана с тем фактом, что вы выполняете merge(), а затем выполняете свою логику? Reactive Stream — это однопоточный вызов onXXX для каждой спецификации (также и в RxJava, и в Akka Stream).

Как и в Rx, я бы просто выполнял работу по обработке внутри плоского потока или просто потреблял, если вам не нужно объединяться обратно (что является спорным моментом, если вы думаете об этом).

return objectStream
        .partition(partitions)
        .consume( partitionStream ->
           partitionStream
             .dispatchOn(Environment.cachedDispatcher()) 
             .consume( o -> {
                try {
                    return extension.eval(o, null);
                } catch (UnableToEvaluateException e) {
                    e.printStackTrace();
                    return null;
                }

            }
        ));

Делал и со слиянием, и без. Не имело никакого значения. Кодовая база здесь 2.0. Меня устраивает однопоточность, как только я нажму на кнопку dispatchOn, но мне нужно воспользоваться преимуществами многоядерности (что, в конце концов, является серьезной причиной для использования FRP). Мои тесты не выявили каких-либо потоков, а печать имени потока не показала никакого разветвления на независимые потоки.

Обратите внимание, что здесь я использую Java 7 вместо 8 (и, черт возьми, убил бы я здесь лямда-функции и их синтаксис).