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(....) 替换为对正确服务的调用即可。&nbsp; &nbsp; public Mono<LocationProfileInfo> getProfileInfoByLocationAndCustomer(String customerId, String location) {&nbsp; &nbsp; Mono<String> customerAccountMono = Mono.just("customerAccount");&nbsp; &nbsp; Mono<LocationProfile> locationProfileMono = Mono.just(new LocationProfile());&nbsp; &nbsp; return Mono.zip(customerAccountMono, locationProfileMono)&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; .flatMap(tuple -> {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; Mono<Customer> customerMono = Mono.just(new Customer(tuple.getT1()));&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; Mono<Result<LocationPricing>> result = Mono.just(new Result<LocationPricing>());&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; Mono<LocationProfile> locationProfile = Mono.just(tuple.getT2());&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; return Mono.zip(customerMono, result, locationProfile);&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; })&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; .map(LocationProfileInfo::new)&nbsp; &nbsp; ;}public static class LocationProfileInfo {&nbsp; &nbsp; public LocationProfileInfo(Tuple3<Customer, Result<LocationPricing>, LocationProfile> tuple){&nbsp; &nbsp; &nbsp; &nbsp; //do wathever&nbsp; &nbsp; }}public static class LocationProfile {}private static class Customer {&nbsp; &nbsp; public Customer(String cutomerAccount) {&nbsp; &nbsp; }}private static class Result<T> {}private static class LocationPricing {}请记住，第一个拉链不是必需的。我重新编写它以解决您的解决方案。但我会以不同的方式解决问题。会更清楚。public Mono<LocationProfileInfo> getProfileInfoByLocationAndCustomer(String customerId, String location) {return Mono.just("customerAccount") //call the service&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;&nbsp; &nbsp; &nbsp; &nbsp; .flatMap(customerAccount -> {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; //declare the call to get the customer&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; Mono<Customer> customerMono = Mono.just(new Customer(customerAccount));&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; //declare the call to get the location pricing&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; Mono<Result<LocationPricing>> result = Mono.just(new Result<LocationPricing>());&nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; //declare the call to get the location profile&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; Mono<LocationProfile> locationProfileMono = Mono.just(new LocationProfile());&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; //in the zip call all the services actually are executed&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; return Mono.zip(customerMono, result, locationProfileMono);&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp; &nbsp; &nbsp; &nbsp; })&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;&nbsp; &nbsp; &nbsp; &nbsp; .map(LocationProfileInfo::new)&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;;&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;}