driversti
@driversti
Кратко о себе

Как правильно запустить с задержкой ~2200 объектов, каждый из которых имеет свой эндпоинт для запроса?

Есть онлайн игра, которая имеет рынок. На рынке 9 продуктов с разными характеристиками. В общей сложности 29 уникальных запросов для получения данных по всех продуктах для рынка одной страны. Всего 74 страны. Итого 2146 запросов.
Запросы должны идти с задержкой, хотя бы 200-250мс между каждым. Таким образом это займёт 7-9 минут, чтобы запустить все.
При старте приложения мой CollectorManager собирает все коллекторы (2146 штук) в один сет (чтобы не было дубликатов, в будущем хотелось бы вешать дополнительные Subscriber'ы на конкретные коллекторы), подписывает на них сервис, который будет записывать данные в БД и запускает процесс сбора данных.

ПРОБЛЕМА. Я заметил, что для некоторых конкретных продуктов (страна, тип товара, его качество Country, Industry, Quality) коллекторы прекращают работать. При старте может быть 1-2-3 вызова с ожидаемой задержкой и все. За всю ночь ни одного вызова. Для других рынков стабильно работают. Остановив приложение в деббаге вижу, что все 2146 коллекторов есть в сете, но некоторые тупо не работают (нету логов от них и данных в базе).
Потом я начал запускать в работу каждый коллектор в самом MangerCollector, так как подумал, что randomDelay() в executorService.schedule(new MarketplaceTask(), randomDelay(), MILLISECONDS) не работает должным образом:

CollectorManager.java
@EventListener(ApplicationReadyEvent.class)
public void start() {
  log.info("Let's get party started!");
  Set<MarketplaceCollector> collectorSet = marketplaceCollectorProviders.stream()
    .flatMap(provider -> provider.getCollectors().stream())
    .peek(this::subscribeOfferDBSubscriber)
    .collect(Collectors.toSet());
  collectors.addAll(collectorSet);
  collectors.forEach(this::runWithDelay);
}

private void subscribeOfferDBSubscriber(MarketplaceCollector marketplaceCollector) {
  marketplaceCollector.subscribe(marketplaceOfferDBSubscriber);
}

private void runWithDelay(Collector collector) {
  try {
    long delay = 200;
    Thread.sleep(delay);
    collector.collect();
  } catch (InterruptedException e) {
    e.printStackTrace();
  }
}


MarketplaceCollector.java
@Override
public void collect() {
  executorService.schedule(new MarketplaceTask(), 1, MILLISECONDS);
}


К сожалению, это не принесло должного эффекта. С многопоточностью раньше не работал, поэтому даже не знаю в какую сторону грести, чтобы заставить все это завестись.

UPDATE. Итак, я заметил, что проблема появляется, когда какой-то Collector пытается получить данные от сервера. Я вижу лог перед строкой var responseEntity = restTemplate.exchange(erepublikHost + MARKET_URL, POST, requestEntity, MarketplaceResponse.class);, а дальше ответа нет. То есть я не знаю пошел запрос на сервер или нет и что вернулось. Такое ощущение, что запрос пошел и дальше эта задача посталена на вечную паузу :(
public MarketplaceResponse getOffers(MarketplaceRequest request) {
        var authentication = authenticationService.getAuthentication();
        var requestEntity = new HttpEntity<>(requestString(request, authentication), headersComposer.getHeaders());
        var restTemplate = restTemplateBuilder.build();
        var responseEntity = restTemplate.exchange(erepublikHost + MARKET_URL, POST, requestEntity, MarketplaceResponse.class);
        return responseEntity.getBody();
}


UPDATE2: "Решил" вопрос создав коллекцию RestTemplat'ов. По крайней мере, теперь знаю где искать проблему. Или deadlock ловлю, или есть другая причина, почему restTemplare.exchange() не возвращает результат для некоторых запросов. При чем для других возвращает... Даже если это один и тот же RestTemplate.
  • Вопрос задан
  • 73 просмотра
Пригласить эксперта
Ответы на вопрос 1
@ads83
Метод `peek` - это промежуточная операция. Одна из особенностей в том, что такие операции - ленивые, и у тебя нет гарантии когда они выполнятся и выполнятся ли вообще.

Подход с ExecutorPool более здоровый, но например в MarketplaceCollector я не увидел инициализацию `final ScheduledExecutorService executorService`.
Кроме того, если у тебя консольное приложение, то когда основной поток закончится, JVM пойдет завершать все дочерние и прибьет таску, которая запланирована через 3 минуты. Убедись, что оно живет достаточно долго.

Третий момент может быть в том, что сервис, к которому ты обращаешься, считает такое число запросов с таким интервалом избыточным и временно перестает отвечать. Это тоже требует проверки.
Ответ написан
Ваш ответ на вопрос

Войдите, чтобы написать ответ

Войти через центр авторизации
Похожие вопросы
28 мая 2020, в 01:17
600 руб./в час
27 мая 2020, в 23:51
3000 руб./за проект