Reactor Mono - 执行并行任务

Question

我是 Reactor 框架的新手，并试图在我们现有的实现之一中使用它。LocationProfileService 和 InventoryService 都返回一个 Mono 并且将并行执行并且彼此没有依赖关系（来自 MainService）。在 LocationProfileService 中 - 发出 4 个查询，最后 2 个查询依赖于第一个查询。有什么更好的方法来写这个？我看到调用按顺序执行，而其中一些应该并行执行。正确的方法是什么？public class LocationProfileService {        static final Cache<String, String> customerIdCache //define Cache        @Override        public Mono<LocationProfileInfo> getProfileInfoByLocationAndCustomer(String customerId, String location) {            //These 2 are not interdependent and can be executed immediately            Mono<String> customerAccountMono = getCustomerArNumber(customerId,location) LocationNumber).subscribeOn(Schedulers.parallel()).switchIfEmpty(Mono.error(new CustomerNotFoundException(location, customerId))).log();            Mono<LocationProfile> locationProfileMono = Mono.fromFuture(//location query).subscribeOn(Schedulers.parallel()).log();    //Should block be called, or is there a better way to do ?            String custAccount = customerAccountMono.block(); // This is needed to execute and the value from this is needed for the next 2 calls            Mono<Customer> customerMono = Mono.fromFuture(//query uses custAccount from earlier step).subscribeOn(Schedulers.parallel()).log();            Mono<Result<LocationPricing>> locationPricingMono = Mono.fromFuture(//query uses custAccount from earlier step).subscribeOn(Schedulers.parallel()).log();            return Mono.zip(locationProfileMono,customerMono,locationPricingMono).flatMap(tuple -> {                LocationProfileInfo locationProfileInfo = new LocationProfileInfo();                //populate values from tuple                return Mono.just(locationProfileInfo);            });        }   

Answer 1

您无需阻止即可通过您的代码非常接近解决方案的参数。我使用您提供的类名编写了代码。只需将所有 Mono.just(....) 替换为对正确服务的调用即可。

    public Mono<LocationProfileInfo> getProfileInfoByLocationAndCustomer(String customerId, String location) {

    Mono<String> customerAccountMono = Mono.just("customerAccount");

    Mono<LocationProfile> locationProfileMono = Mono.just(new LocationProfile());

    return Mono.zip(customerAccountMono, locationProfileMono)

            .flatMap(tuple -> {

                Mono<Customer> customerMono = Mono.just(new Customer(tuple.getT1()));

                Mono<Result<LocationPricing>> result = Mono.just(new Result<LocationPricing>());

                Mono<LocationProfile> locationProfile = Mono.just(tuple.getT2());

                return Mono.zip(customerMono, result, locationProfile);

            })

            .map(LocationProfileInfo::new)

    ;

}

public static class LocationProfileInfo {

    public LocationProfileInfo(Tuple3<Customer, Result<LocationPricing>, LocationProfile> tuple){

        //do wathever

    }

}

public static class LocationProfile {}

private static class Customer {

    public Customer(String cutomerAccount) {

    }

}

private static class Result<T> {}

private static class LocationPricing {}

请记住，第一个拉链不是必需的。我重新编写它以解决您的解决方案。但我会以不同的方式解决问题。会更清楚。

public Mono<LocationProfileInfo> getProfileInfoByLocationAndCustomer(String customerId, String location) {

return Mono.just("customerAccount") //call the service                                                

        .flatMap(customerAccount -> {                                                                 

            //declare the call to get the customer                                                    

            Mono<Customer> customerMono = Mono.just(new Customer(customerAccount));                   

            //declare the call to get the location pricing                                            

            Mono<Result<LocationPricing>> result = Mono.just(new Result<LocationPricing>());          

            //declare the call to get the location profile                                            

            Mono<LocationProfile> locationProfileMono = Mono.just(new LocationProfile());             

            //in the zip call all the services actually are executed                                  

            return Mono.zip(customerMono, result, locationProfileMono);                               

        })                                                                                            

        .map(LocationProfileInfo::new)                                                                

;                                                                                                     

}