Использование Flux.buffer реактора для пакетной работы работает только для одного элемента - PullRequest
Новая
       14

Использование Flux.buffer реактора для пакетной работы работает только для одного элемента

12 голосов
/ 15 марта 2019

Я пытаюсь использовать Flux.buffer() для пакетной загрузки из базы данных.

Вариант использования: загрузка записей из БД может быть "пакетной", и я хотел бы представитьНебольшой буфер для группирования загружается там, где это возможно.

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

Я пробовал несколько разных подходов (разные типы процессоров, по-разному создавая отфильтрованный Mono).

Ниже я дошел до того места - в основном, спотыкаясь.

В настоящее время возвращается один результат, но последующие вызовы сбрасываются (хотя я не уверен, где). 

class BatchLoadingRepository {
    // I've tried all manner of different processors here.  I'm unsure if
    // TopicProcessor is the correct one to use.
    private val bufferPublisher = TopicProcessor.create<String>()
    private val resultsStream = bufferPublisher
            .bufferTimeout(50, Duration.ofMillis(50))
            // I'm unsure if concatMapIterable is the correct operator here, 
            // but it seems to work.
            // I'm really trying to turn the List<MyEntity> 
            // into a stream of MyEntity, published on the Flux<>
            .concatMapIterable { requestedIds ->
                // this is a Spring Data repository.  It returns List<MyEntity>
                repository.findAllById(requestedIds)
            }

    // Multiple callers will invoke this method, and then subscribe to receive
    // their entity back.
    fun findByIdAsync(id: String): Mono<MyEntity> {

        // Is there a potential race condition here, caused by a result
        // on the resultsStream, before I've subscribed?
        return Mono.create<MyEntity> { sink ->
            bufferPublisher.sink().next(id)
            resultsStream.filter { it.id == id }
                    .subscribe { next ->
                        sink.success(next)
                    }
        }
    }
}

1 Ответ

0 голосов
/ 20 марта 2019

Привет, я тестировал ваш код и думаю, что лучший способ - это использовать EmitterProcessor shared. Я сделал тест с emitterProcessor, и он работает 

 Flux<String> fluxi;
EmitterProcessor emitterProcessor;

@Override
public void run(String... args) throws Exception {
    emitterProcessor = EmitterProcessor.create();

    fluxi = emitterProcessor.share().bufferTimeout(500, Duration.ofMillis(500))
            .concatMapIterable(o -> o);
;

    Flux.range(0,1000)
            .flatMap(integer -> findByIdAsync(integer.toString()))
            .map(s -> {
                System.out.println(s);
                return s;
            }).subscribe();

}

private Mono<String> findByIdAsync(String id) {
    return Mono.create(monoSink -> {
        fluxi.filter(s -> s == id).subscribe(value -> monoSink.success(value));
        emitterProcessor.onNext(id);
    });
}
...