单线单熔剂

Question

Flux<Foo> flux = dao.getAll();
Flux<Foo> fluxAfterProcessing = flux.flatMap(foo -> monoFilters.map(...));
Mono<Tuple2<MyRequest, MyResponse>> mono = 
                      monoFilters.flatMap(filter -> monoRequest.map(...))
                                 .zipWhen(request -> api.send(request))
                                 .flatMap(tuple -> monoResponseFilters.map(...));
return fluxAfterProcessing.flatMap(foo -> 
                       monoResponseFilters.zipWith(mono).map(...))

Answer 1

让我们假设这个任务如下所示：

• 从数据库获取一些值
• 当所有值到达时，将它们包装在请求中并发送出去
• 用响应压缩结果

然后这就引出了类似这样的东西：

Flux<Foo> foos = dao.getAll();
Mono<List<Foo>> everything = foos.collectList();

Mono<MyRequest> request = everything
    // collect the data into another Mono, then into request
    .map(list -> list.stream().map(Foo::getData).collect(toList()))
    .map(data -> new MyRequest(data));

return request.zipWhen(request -> api.send(request));

Flux<Data> data = dao.getAll().map(Foo::getData);
Mono<MyRequest> request = data.collectList().map(MyRequest::new);