java使用反应堆的通量。缓冲区到批处理工作仅适用于单个项目

Question

我正在尝试使用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)
                    }
        }
    }
}

Answer 1

嗨，我在测试你的代码，我认为最好的方法是使用EmitterProcessor共享。我用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);
    });
}