如何迭代 Flux 中的对象并对其进行操作？

Question

我正在使用项目反应器，我想执行以下操作：

    @Override
    public void run(ApplicationArguments args) {
        Flux.from(KafkaReceiver.create(receiverOptions)
                        .receive()
                        .map(this::getObject)
                        .flatMap(this::iterateElasticWrites)
                        .flatMap(this::writeTheWholeObjectToS3)
        ).subscribe();
    }

    // What I'd like to do - but a non reactive code
    private Publisher<MyObj> iterateElasticWrites(MyObj message) {
        for (MyDoc file: message.getDocs()) {
            writeElasticDoc(file.getText());
        }
        return Mono.just(message);
    }

我正在努力寻找与iterateElasticWrites反应堆项目中的等效项。我想对我的对象 ( MyObj) 执行迭代，并将其每个文档列表的元素反应性地写入 elasticsearch 中。

Answer 1

在 Reactor 中，您始终需要使用不同的运算符构建反应式流程，并且所有反应式/异步代码都应返回Monoor Flux。

看看你的例子，它可能看起来像

private Mono<MyObj> iterateElasticWrites(MyObj message) {
    return Flux.fromIterable(message.getDocs())
            .flatMap(doc -> writeElasticDoc(doc.getText()))
            .thenReturn(message);
}

其中writeElasticDoc可以定义为

private Mono<Void> writeElasticDoc(String text) {
    ...
}