Конфигурация задания весеннего пакета с зависимыми шагами, которые должны выполнять сложный процесс.

Question

У меня есть требование, подобное приведенному ниже. Я новичок в весенней серии, так что кто-то может мне помочь в создании работы для этого случая.

1. Прочитайте список стран, в кусках (скажем, мне нужно прочитать 5 стран одновременно)
2. Для каждой страны получить список государств
3. Для каждого государства получить список городов
4. a. Для каждого города получить список людей в кусках (скажем, мне нужно читать 1000 записей за раз)
   b. Обработка информации о людях
   c. Написать информацию в БД

Все вышеперечисленные шаги зависят сказать, что мне нужно передать идентификатор страны, начиная с шага 1 к шагу 2, stateId от шага 2 к шагу 3 и CityId из шага 3 этап 4.

Внутреннее выполнение этапов 1 и 4 может быть параллельным. Даже это можно сделать на разных jvm

Answer 1

Я думаю, у вас есть несколько вариантов. Во-первых, я бы предложил посмотреть на org.springframework.batch.item.support.CompositeItemProcessor и связать эти процессы за один шаг (http://docs.spring.io/spring-batch/trunk/reference/html/readersAndWriters.html#chainingItemProcessors).

Другим вариантом было бы использовать прослушиватель рекламных акций для обмена данными между этапами (org.springframework.batch.core.listener.ExecutionContextPromotionListener). Это позволит вам создавать разные этапы и контролировать интервал фиксации на каждом шаге. Вот пример базового XML-конфигурации для продвижения слушателя:

<bean id="promotionListener" class="org.springframework.batch.core.listener.ExecutionContextPromotionListener"> 
    <property name="keys"> 
     <util:list> 
      <value>countries</value> 
      <value>states</value> 
      <value>cities</value> 
     </util:list> 
    </property> 
</bean> 

Вы можете добавить параметры ExecutionContext с помощью пользовательского Tasklet путем реализации интерфейса и переопределение метода execute. В приведенном выше примере три списка добавляются в прослушиватель рекламных роликов, которые затем могут быть разделены между шагами по цепочке.

Вот краткий пример того, как переопределить execute метод:

@Override 
public RepeatStatus execute(StepContribution stepContribution, ChunkContext chunkContext) throws Exception { 

    ExecutionContext executionContext = chunkContext.getStepContext().getStepExecution().getExecutionContext(); 

    executionContext.put("countries", /* country list */; 
    executionContext.put("states", /* state list */); 
    executionContext.put("cities", /* city list */); 

    return RepeatStatus.FINISHED; 
} 

Вы, вероятно, хотите, чтобы разорвать этот выход на три тасклетов - по одному для каждого шага в процессе - и добавить один список за тасклет. Я бы также увеличил обработку RepeatStatus в случае ошибки, но это всего лишь быстрый пример, который даст вам представление.

Надеюсь, что это поможет!

Answer 2

Вот предлагаемый подход, так как ваш шаг № 1 - шаг 4 - это в основном чтение из базы данных, хотя эти четыре шага выглядят отдельными шагами, но эти четыре шага - это один шаг. И вы также должны уменьшить удары DB в дополнение к решению проблемы функционально.

Вы можете использовать Spring Batch секционирования подход, при котором вы разделите ваши данные по тройным - (страна, штат, город) то ваш читатель считывает данные блоками по 1000 для каждой тройки в параллельных потоках, т.е. WHERE COUNTRY = ? AND STATE =? AND CITY = ?.

Вы должны реализовать интерфейс Partitioner и создать свои триплеты, выполнив запрос SELECT и сделав GROUP BY COUNTRY,STATE,CITY.

Тогда вы пишете простой читатель, который возвращает JdbcPagingItemReader и использовать COUNTRY,STATE,CITY значения заселенного в Partitioner в WHERE пункте PagingQueryProvider.

Вам нужно будет ввести значения разделов из логики разделителя в считыватель с помощью синтаксиса @Value("#{stepExecutionContext[...]}".

org.springframework.batch.core.partition.support.Partitioner 

Вам не потребуются для выполнения всех разделов в один раз - вы можете контролировать количество параллельных потоков, обеспечивая SimpleAsyncTaskExecutor разметить мастер шаг и установив его concurrencyLimit в соответствующее значение.

Затем вы продолжаете писать свой процессор и писателя. Вам нужно будет отметить шаг & компонентов шага в @StepScope, если вводить значения из логики раздела.

Это Sample of partitiong, но его xml-конфигурация. Я предпочитаю конфигурацию на основе Java.

Надеюсь, это поможет!