使用project Reacter处理多项外部服务

Question

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

响应getDetails（Request 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());
}

null

Answer 1

如果您有n个呼叫，并且希望在steplock中移动（也就是说，如果您有来自所有呼叫的响应，则向前移动），请使用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())并行运行它们

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

>

不需要使用block()订阅，因为flatmap会急切地订阅您的内部流。您可以执行以下操作：

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

注意:mono.callable(..)如果调用对象返回空流，则返回空流。这就是switchifempty的原因