使用项目反应器处理多个外部服务

Question

我是响应式编程的新手，并尝试使用项目反应器模拟以下用例，但我发现将响应从一个服务调用传递到另一个依赖服务有点困难。任何建议或参考将不胜感激。

响应 getDetails(请求 inputRequest){

    //Call two external services parallel based on the incoming request 
        Response1 = callExternalService1(inputRequest)
        Response2 = callExternalService2(inputRequest)


    //Call one external service based on the Response1 and Response2
        Response3 = callExternalService3(Response1,Response2);


    //Call three external service parallel based on the Response1, Response2 and Response3
        Response4 = callExternalService4(Response1,Response2,Response3);
        Response5 = callExternalService5(Response1,Response2,Response3);
        Response6 = callExternalService6(Response1,Response2,Response3);


    //Last service call
       Response finalResponse= callLastExternalService(Response4,Response5,Response6);
       return finalResponse;

}

我尝试了以下示例，它适用于一个服务调用，但无法将响应传递给其他相关服务调用。

更新答案：

Mono<Response> getDetails(Request inputRequest){
    
    return Mono.just(inputRequest.getId())
                    .flatMap(id->{
                        DbResponse res = getDBCall(id).block();
                        if(res == null){
                            return Mono.error(new DBException(....));
                        }
                        return Mono.zip(callExternalService1(res),callExternalService2(inputRequest));
                    }).flatMap(response->{
                        Response extser1 = response.getT1();
                        Response extser2 = response.getT2();
                        //any exceptions?
                        return Mono.zip(Mono.just(extser1),Mono.just(extser2),callExternalService3();
                    }).flatMap(response->callExternalService4(response.getT1(),response.getT2(),response.getT3())
                    });
}

private Mono<DbResponse> getDBCall(String id) {
        return Mono.fromCallable(()->dbservice.get(id))
                .subscribeOn(Schedulers.boundedElastic());
}

问题：

1. 如何在不使用块的情况下将 Mono 转换为 DbResponse 操作？
2. 如果任何外部服务失败，如何构建 平面图中的失败响应并返回？

Answer 1

如果您有 n 个电话，并且您想步入锁步状态（也就是说，如果您有所有电话的响应，则继续前进），请使用 zip。例如：

Mono.zip(call1, call2)
    .flatMap(tuple2 -> {
         ResponseEntity<?> r1 = tuple2.getT1();    //response from call1
         ResponseEntity<?> r2 = tuple2.getT2();    //response from call2
         return Mono.zip(Mono.just(r1), Mono.just(r2), call3);
     })
    .flatMap(tuple3 -> {
         //at this point, you have r1, r2, r3. tuple3.getT1() response from call 1
         return Mono.zip(call4, call5, call6);   //tuple3.getT2() response from call 2, tuple3.getT3() response from call3
     })
    .flatMap(tuple3 -> callLastService);

注意：如果更多的是伪代码，则不会立即编译

您可以扩展上述内容以回答您自己的问题。请注意，由于call1 和call2 是独立的，您可以使用subscribeOn(Schedulers.boundedElastic()) 并行运行它们

编辑：回答两个后续问题：

1. 无需使用block() 订阅，因为flatMap 急切地订阅您的内部流。您可以执行以下操作：

    Mono.just(inputRequest.getId())
        .flatMap(a -> getDBCall(a).switchIfEmpty(Mono.defer(() -> Mono.error(..))))
   

注意：如果可调用返回空，Mono.callable(..) 返回一个空流。这就是为什么switchIfEmpty

1. 您可以使用onErrorResume 等运算符来提供后备流。见：The difference between onErrorResume and doOnError

Answer 2

如果您的服务返回 Mono of Response（否则您必须对其进行转换），您可以使用 zip 进行并行调用：

    Mono.zip( callExternalService1( inputRequest ),
              callExternalService2( inputRequest ) )
        .flatMap( resp1AndResp2 -> this.callExternalService3( resp1AndResp2.getT1(),
                                                              resp1AndResp2.getT2() )
                                       .flatMap( response3 -> Mono.zip( callExternalService4( resp1AndResp2.getT1(),
                                                                                              resp1AndResp2.getT2(),
                                                                                              response3 ),
                                                                        callExternalService5( resp1AndResp2.getT1(),
                                                                                              resp1AndResp2.getT2(),
                                                                                              response3 ),
                                                                        callExternalService6( resp1AndResp2.getT1(),
                                                                                              resp1AndResp2.getT2(),
                                                                                              response3 ) )
                                                                  .flatMap( resp4AndResp5AndResp6 -> callLastExternalService( resp4AndResp5AndResp6.getT1(),
                                                                                                                              resp4AndResp5AndResp6.getT2(),
                                                                                                                              resp4AndResp5AndResp6.getT3() ) ) ) );