Как запустить моно в том же потоке, внутри параллельного потока - PullRequest
Новая
       9

Как запустить моно в том же потоке, внутри параллельного потока

0 голосов
/ 08 января 2020

Я пытаюсь заполнить объекты внутри Flux значениями из Mono. Когда я пытаюсь это сделать, он просто игнорирует мою операцию «set». Я предполагаю, что это потому, что Flux работает параллельно, а Mono - нет. Как я могу решить эту проблему? 

Flux.fromIterable(proxyParserService.getProxyList())
            .parallel()
            .runOn(Schedulers.parallel())
            .filter(proxy -> proxy.getCorrupted() == null || !proxy.getCorrupted())
            .subscribe(proxy -> {
                        try {
                            RestTemplate restTemplate = getProxiedTemplate(proxy.getHost(), proxy.getPort());
                            restTemplate.exchange(URI, HttpMethod.GET, HttpEntity.EMPTY, String.class);
                            geoDataService.getData(proxy.getHost()) // Here comes the Mono object, that contains needed value to set into "proxy"
                                    .subscribe(geoData ->
                                    {
                                        log.info("GEODATA: {} ", geoData);
                                        proxy.setCountryCode(geoData.getCountryCode()); // ignored somehow
                                    });
                            proxy.setCorrupted(false);
                            addresses.add(proxy);
                            log.info("IP {}:{} is OK", proxy.getHost(), proxy.getPort());
                            log.info("Final result: {}", proxy.toString());
                        } catch (ResourceAccessException e) {
                            log.info("IP {}:{} is corrupted!", proxy.getHost(), proxy.getPort());
                            proxy.setCorrupted(true);
                            addresses.add(proxy);
                        }
                    },
                    throwable -> log.error(String.format("Exception caught while trying to fill map: %s", throwable.getCause())));

}

Вот некоторые журналы

Как вы можете видеть, я пытаюсь установить код страны в прокси.

1 Ответ

0 голосов
/ 09 января 2020

решаемая. Добавлен этот моно объект в оператор "flatMap". Пример: 

Flux.fromIterable(proxyParserService.getProxyList())
            .parallel()
            .runOn(Schedulers.parallel())
            .filter(poxy -> !valueExist(addresses.values(), poxy))
            .flatMap(geoDataService::getData) // Now it runs in parallel threads
            .subscribe(proxy -> {
                        try {
                            RestTemplate restTemplate = getProxiedTemplate(proxy.getHost(), proxy.getPort());
                            restTemplate.exchange(URI, HttpMethod.GET, HttpEntity.EMPTY, String.class);
                            proxy.setCorrupted(false);
                            addresses.put(proxy.getCountryCode(), proxy);
                            log.info("IP {}:{} is OK", proxy.getHost(), proxy.getPort());
                            log.info("Final result: {}", proxy.toString());
                        } catch (ResourceAccessException e) {
                            log.info("IP {}:{} is corrupted!", proxy.getHost(), proxy.getPort());
                            proxy.setCorrupted(true);
                            addresses.put(proxy.getCountryCode(), proxy);
                        }
                    },
                    throwable -> log.error(String.format("Exception caught while trying to fill map: %s", throwable.getCause())));
...