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

Question

[Yurii Nekrasov]

Есть онлайн игра, которая имеет рынок. На рынке 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.

Answer 1

[ads83]

Метод `peek` - это промежуточная операция. Одна из особенностей в том, что такие операции - ленивые, и у тебя нет гарантии когда они выполнятся и выполнятся ли вообще.

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

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

Comments

[Yurii Nekrasov]

Спасибо за ответы.
По поводу peek: работает, проверено. Проблема выше, не запускаются таски, если их число переваливает за несколько сотен. Пока не знаю каков предел. При 100-200 тасках все работает как надо.
Инициализация ScheduledExecutorService в конструкторе, это Spring Boot приложение.
При большом количестве запросов, внешний сервер отвечает 200, to many requests, что ожидаемо и я логгирую такие ситуации и пропускаю.