Project Reactor, использующий приемник Flux за пределами создания лямбды

Question

• Когда мой сервис запускается, я хочу построить простой конвейер.
• Я бы хотел выделить приемник Flux или Процессор для генерации событий с помощью.
• События будут поступать из нескольких потоков и должны обрабатываться в соответствии со спецификацией subscribeOn() конвейера, но, похоже, все работает в потоке main.
• Каков наилучший подход? Я приложил свои попытки ниже.
• (я использую реактор с активной зоной v3.2.8.RELEASE.)

import org.junit.jupiter.api.Test;

import reactor.core.publisher.DirectProcessor;
import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxProcessor;
import reactor.core.publisher.FluxSink;
import reactor.core.scheduler.Schedulers;

/**
 * I want to construct my React pipelines during creation,
 * then emit events over the lifetime of my services.
 */
public class React1Test
{
    /**
     * Attempt 1 - use a DirectProcessor and send items to it.
     * Doesn't work though - seems to always run on the main thread.
     */
    @Test
    public void testReact1() throws InterruptedException
    {
        // Create the flux and sink.
        FluxProcessor<String, String> fluxProcessor = DirectProcessor.<String>create().serialize();
        FluxSink<String> sink = fluxProcessor.sink();

        // Create the pipeline.
        fluxProcessor
            .doOnNext(str -> showDebugMsg(str))   // What thread do ops work on?
            .subscribeOn(Schedulers.elastic())
            .subscribe(str -> showDebugMsg(str)); // What thread does subscribe run on?

        // Give the multi-thread pipeline a second.
        Thread.sleep(1000);

        // Time passes ... things happen ...
        // Pass a few messages to the sink, emulating events.
        sink.next("a");
        sink.next("b");
        sink.next("c");

        // It's multi-thread so wait a sec to receive.
        Thread.sleep(1000);
    }

    // Used down below during Flux.create().
    private FluxSink<String> sink2;

    /**
     * Attempt 2 - use Flux.create() and its FluxSink object.
     * Also seems to always run on the main thread.
     */
    @Test
    public void testReact2() throws InterruptedException
    {
        // Create the flux and sink.
        Flux.<String>create(sink -> sink2 = sink)
            .doOnNext(str -> showDebugMsg(str))   // What thread do ops work on?
            .subscribeOn(Schedulers.elastic())
            .subscribe(str -> showDebugMsg(str)); // What thread does subscribe run on?

        // Give the multi-thread pipeline a second.
        Thread.sleep(1000);

        // Pass a few messages to the sink.
        sink2.next("a");
        sink2.next("b");
        sink2.next("c");

        // It's multi-thread so wait a sec to receive.
        Thread.sleep(1000);
    }

    // Show us what thread we're on.
    private static void showDebugMsg(String msg)
    {
        System.out.println(String.format("%s [%s]", msg, Thread.currentThread().getName()));
    }
}

Вывод всегда:

a [main]
a [main]
b [main]
b [main]
c [main]
c [main]

Но то, что я ожидал, это:

a [elastic-1]
a [elastic-1]
b [elastic-2]
b [elastic-2]
c [elastic-3]
c [elastic-3]

Заранее спасибо.

Answer 1

• Вы можете использовать parallel() и runOn() вместо subscribeOn(), чтобы получить sink.next() для многопоточной работы.
• bsideup также правильно - вы можете использовать publishOn() дляпринудительно запускать нисходящие операторы в одном другом потоке планировщика.

Вот мой обновленный код:

import org.junit.jupiter.api.Test;

import reactor.core.publisher.DirectProcessor;
import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxProcessor;
import reactor.core.publisher.FluxSink;
import reactor.core.scheduler.Schedulers;

/**
 * I want to construct my React pipelines during creation,
 * then emit events over the lifetime of my services.
 */
public class React1Test
{
    /**
     * Version 1 - use a DirectProcessor to dynamically emit items.
     */
    @Test
    public void testReact1() throws InterruptedException
    {
        // Create the flux and sink.
        FluxProcessor<String, String> fluxProcessor = DirectProcessor.<String>create().serialize();
        FluxSink<String> sink = fluxProcessor.sink();

        // Create the pipeline.
        fluxProcessor
            .parallel()
            .runOn(Schedulers.elastic())
            .doOnNext(str -> showDebugMsg(str))   // What thread do ops work on?
            .subscribe(str -> showDebugMsg(str)); // What thread does subscribe run on?

        // Give the multi-thread pipeline a second.
        Thread.sleep(1000);

        // Time passes ... things happen ...
        // Pass a few messages to the sink, emulating events.
        sink.next("a");
        sink.next("b");
        sink.next("c");

        // It's multi-thread so wait a sec to receive.
        Thread.sleep(1000);
    }

    // Used down below during Flux.create().
    private FluxSink<String> sink2;

    /**
     * Version 2 - use Flux.create() and its FluxSink object.
     */
    @Test
    public void testReact2() throws InterruptedException
    {
        // Create the flux and sink.
        Flux.<String>create(sink -> sink2 = sink)
            .parallel()
            .runOn(Schedulers.elastic())
            .doOnNext(str -> showDebugMsg(str))   // What thread do ops work on?
            .subscribe(str -> showDebugMsg(str)); // What thread does subscribe run on?

        // Give the multi-thread pipeline a second.
        Thread.sleep(1000);

        // Pass a few messages to the sink.
        sink2.next("a");
        sink2.next("b");
        sink2.next("c");

        // It's multi-thread so wait a sec to receive.
        Thread.sleep(1000);
    }

    // Show us what thread we're on.
    private static void showDebugMsg(String msg)
    {
        System.out.println(String.format("%s [%s]", msg, Thread.currentThread().getName()));
    }
}

Обе версии выдают желаемый многопоточный вывод:

a [elastic-2]
b [elastic-3]
c [elastic-4]
b [elastic-3]
a [elastic-2]
c [elastic-4]

Answer 2

Вы видите [main], потому что вы звоните onNext из основного потока. subscribeOn вы используете только для подписки (когда срабатывает лямбда create). Если вы используете publishOn вместо subscribeOn.

, вы увидите протокол elastic-*.

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