Reactor Mono - выполнять параллельные задачи

Question

Я новичок в Reactor Framework и пытаюсь использовать его в одной из наших существующих реализаций.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);
            });


        }

        private Mono<String> getCustomerAccount(String conversationId, String customerId, String location) {
            return CacheMono.lookup((Map)customerIdCache.asMap(),customerId).onCacheMissResume(Mono.fromFuture(//query).subscribeOn(Schedulers.parallel()).map(x -> x.getAccountNumber()));
        }

}


public class InventoryService {

    @Override
    public Mono<InventoryInfo> getInventoryInfo(String inventoryId) {
        Mono<Inventory> inventoryMono = Mono.fromFuture(//inventory query).subscribeOn(Schedulers.parallel()).log();
        Mono<List<InventorySale>> isMono = Mono.fromFuture(//inventory sale query).subscribeOn(Schedulers.parallel()).log();

        return Mono.zip(inventoryMono,isMono).flatMap(tuple -> {
            InventoryInfo inventoryInfo = new InventoryInfo();
            //populate value from tuple

            return Mono.just(inventoryInfo);

        });
    }

}

public class MainService {

        @Autowired
        LocationProfileService locationProfileService;
        @Autowired
        InventoryService inventoryService

        public void mainService(String customerId, String location, String inventoryId) {
            Mono<LocationProfileInfo> locationProfileMono = locationProfileService.getProfileInfoByLocationAndCustomer(....);
            Mono<InventoryInfo> inventoryMono = inventoryService.getInventoryInfo(....);

            //is using block fine or is there a better way to do?
            Mono.zip(locationProfileMono,inventoryMono).subscribeOn(Schedulers.parallel()).block();
        }

}

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)                                                                
;                                                                                                     

}