通过Reactor功能管道下游的物体

Question

我是一个新的反应器和反应编程，我正试图解决下面的情况。

我从Kafka主题接收到一个对象流，对于flux中的每个记录，我需要调用2个服务并验证对象。

public void consume(Flux<Data> flux)
{
flux.map(data->callRESTService1(data)).map(...<I need the data once again here to call rest service 2>
}

现在我正在使用下面的样式来实现这一点，但是有没有更好的/正确的方法来做到这一点呢？

public void consume(Flux<Data> flux)
{
   flux.subscribe(data->handleData(data));
}


 public void handleData(data)
    {
 Flux.concat(callRestService1(data),callRestService2(data)).reduce(data,reduce());
    }

此外，如果其中一个服务关闭，我需要在侦听器上传播错误，以便消息不被确认，但在另一种情况下，如果验证失败，则需要将消息发布到另一个主题。

Answer 1

这两个路径都需要原始元素，而且每个路径都有不同的处理错误的方法，这一事实很好地表明您可能需要flatmap:

Flux<Data> source; //= ...
return source.flatMap(value -> {
    Mono<IgnoreMe1> service1 = callRestService1(value);
    Mono<IgnoreMe2> service2 = callRestService2(value)
        .onErrorResume(e -> postErrorToTopic(e, value)); //might need some type massaging, eg. if the post to topic method returns a `Mono<Void>`

    //wait for the two to complete, propagate their errors if any, else return original value
    return Mono.when(service1, service2)
       .thenReturn(value);
}