使用反应器的 Flux.buffer 进行批处理工作仅适用于单个项目

java kotlin project-reactor reactive-streams

2022-09-01 21:26:03

我正在尝试用于从数据库批处理加载。Flux.buffer()

用例是，从数据库加载记录可能是“突发的”，我想引入一个小缓冲区，以便在可能的情况下将负载组合在一起。

我的概念方法是使用某种形式的处理器，发布到它的接收器，让该缓冲区，然后订阅和过滤我想要的结果。

我尝试了多种不同的方法（不同类型的处理器，以不同的方式创建过滤的单声道）。

以下是我到目前为止所得到的地方 - 主要是通过绊脚石。

目前，这返回单个结果，但后续调用被丢弃（尽管我不确定在哪里）。

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

嗨，我正在测试你的代码，我认为最好的方法是使用发射器处理器共享。我用发射器处理器做了一个测试，它似乎有效。

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);
    });
}

答案 2