streamalism.org

1. Prehľad

V našom predchádzajúcom úvode do Spring Batch sme predstavili rámec ako nástroj na dávkové spracovanie. Preskúmali sme tiež podrobnosti o konfigurácii a implementáciu vykonávania úloh s jedným vláknom a jedným procesom.

Na implementáciu úlohy s paralelným spracovaním je k dispozícii celý rad možností. Na vyššej úrovni existujú dva režimy paralelného spracovania:

1. Jednoprocesný, viacvláknový
2. Multi-proces

V tomto rýchlom článku si ukážeme rozdelenie na Krok, ktoré je možné implementovať pre jednoprocesové aj viacprocesové úlohy.

2. Rozdelenie disku

Spring Batch s rozdelením nám poskytuje zariadenie na rozdelenie vykonania a Krok:

Prehľad rozdelenia na oddiely

Vyššie uvedený obrázok ukazuje implementáciu a Job s priečkou Krok.

Je tu Krok nazýva sa „Master“, ktorého vykonanie je rozdelené do niekoľkých krokov „Slave“. Títo otroci môžu nastúpiť na miesto pána a výsledok sa stále nezmení. Pán aj otrok sú inštanciami Krok. Otrokmi môžu byť vzdialené služby alebo len lokálne vykonávajúce vlákna.

Ak je to potrebné, môžeme odovzdať údaje od nadriadeného k podriadenému. Metaúdaje (t. J JobRepository), zaisťuje, že každý otrok je vykonaný iba raz v jednom vykonaní Job.

Tu je sekvenčný diagram, ktorý ukazuje, ako to celé funguje:

Krok rozdelenia disku

Ako je znázornené, PartitionStep vedie exekúciu. The PartitionHandler je zodpovedný za rozdelenie diela „Majster“ na „Otroci“. Úplne vpravo Krok je otrokom.

3. Maven POM

Závislosti Maven sú rovnaké, ako sú uvedené v našom predchádzajúcom článku. To znamená Spring Core, Spring Batch a závislosť na databáze (v našom prípade SQLite).

4. Konfigurácia

V našom úvodnom článku sme videli príklad prevodu niektorých finančných údajov z CSV do XML súboru. Pozrime sa na ten istý príklad.

Tu prevedieme finančné informácie z 5 súborov CSV na zodpovedajúce súbory XML pomocou implementácie s viacerými vláknami.

To môžeme dosiahnuť pomocou jediného Job a Krok rozdelenie. Budeme mať päť vlákien, jedno pre každý zo súborov CSV.

Najskôr si vytvorme prácu:

@Bean (name = "partitionerJob") verejná úloha partitionerJob () hodí UnexpectedInputException, MalformedURLException, ParseException {return jobs.get ("partitioningJob") .start (partitionStep ()) .build (); }

Ako vidíme, toto Job začína na Rozdelenie disku. Toto je náš hlavný krok, ktorý bude rozdelený do rôznych krokov otroka:

@Bean public Step partitionStep () hodí UnexpectedInputException, MalformedURLException, ParseException {return steps.get ("partitionStep") .partitioner ("slaveStep", partitioner ()) .step (slaveStep ()) .taskExecutor (taskExecutor ()) .build (); }

Tu vytvoríme PartitioningStep pomocou StepBuilderFactory. Za týmto účelom musíme poskytnúť informácie o Kroky otroka a Rozdeľovač.

The Rozdeľovač je rozhranie, ktoré poskytuje zariadenie na definovanie sady vstupných hodnôt pre každého z podriadených. Inými slovami, ide sem logika rozdelenia úloh na príslušné vlákna.

Vytvorme jeho implementáciu, tzv CustomMultiResourcePartitioner, kde vložíme názvy vstupných a výstupných súborov do ExecutionContext preniesť na každý krok otroka:

public class CustomMultiResourcePartitioner implementuje Partitioner {@Override public Map partition (int gridSize) {Map map = new HashMap (gridSize); int i = 0, k = 1; pre (Zdroj zdroja: zdroje) {ExecutionContext kontext = nový ExecutionContext (); Assert.state (resource.exists (), "Zdroj neexistuje:" + zdroj); context.putString (keyName, resource.getFilename ()); context.putString ("opFileName", "výstup" + k +++ ". xml"); map.put (PARTITION_KEY + i, kontext); i ++; } spiatočná mapa; }}

Pre túto triedu tiež vytvoríme fazuľu, kde dáme zdrojový adresár pre vstupné súbory:

@Bean public CustomMultiResourcePartitioner partitioner () {CustomMultiResourcePartitioner partitioner = nový CustomMultiResourcePartitioner (); Zdroje [] zdroje; try {resources = resourcessePatternResolver .getResources ("súbor: src / main / resources / input / *. csv"); } catch (IOException e) {throw new RuntimeException ("Problémy I / O pri riešení" + "vzoru vstupného súboru.", e); } partitioner.setResources (prostriedky); spätný delič; }

Definujeme otrocký krok, rovnako ako každý iný krok s čitateľom a pisateľom. Čitateľ a pisateľ budú rovnakí, ako sme videli v našom úvodnom príklade, ibaže z parametra StepExecutionContext.

Upozorňujeme, že tieto fazule musia byť stupňovito dimenzované, aby boli schopné prijať stepExecutionContext params, na kazdom kroku. Ak nebudú mať rozsah krokov, ich fazuľa sa vytvorí pôvodne a nebude akceptovať názvy súborov na úrovni kroku:

@StepScope @Bean public FlatFileItemReader itemReader (@Value ("# {stepExecutionContext [názov súboru]}") reťazec názov súboru) vyvolá UnexpectedInputException, ParseException {FlatFileItemReader reader = nový FlatFileItemReader (); DelimitedLineTokenizer tokenizer = nový DelimitedLineTokenizer (); Reťazec [] tokeny = {"používateľské meno", "ID používateľa", "dátum transakcie", "suma"}; tokenizer.setNames (tokeny); reader.setResource (nový ClassPathResource ("vstup /" + názov súboru)); DefaultLineMapper lineMapper = nový DefaultLineMapper (); lineMapper.setLineTokenizer (tokenizer); lineMapper.setFieldSetMapper (nový RecordFieldSetMapper ()); reader.setLinesToSkip (1); reader.setLineMapper (lineMapper); spätná čítačka; } 

@Bean @StepScope public ItemWriter itemWriter (Marshaller marshaller, @Value ("# {stepExecutionContext [opFileName]}") reťazec názov súboru) hodí MalformedURLException {StaxEventItemWriter itemWriter = nový StaxEventItemWriter (); itemWriter.setMarshaller (zriaďovateľ); itemWriter.setRootTagName ("transactionRecord"); itemWriter.setResource (nový ClassPathResource ("xml /" + názov súboru)); vrátiť itemWriter; }

Pri zmienke o čitateľovi a pisateľovi v kroku otroka môžeme argumenty odovzdať ako nulové, pretože tieto názvy súborov sa nebudú používať, pretože názvy súborov dostanú od stepExecutionContext:

@Bean public Step slaveStep () hodí UnexpectedInputException, MalformedURLException, ParseException {return steps.get ("slaveStep"). Chunk (1) .reader (itemReader (null)) .writer (itemWriter (marshaller (), null)) .build (); }

5. Záver

V tomto tutoriáli sme diskutovali o tom, ako implementovať úlohu s paralelným spracovaním pomocou Spring Batch.

Kompletná implementácia pre tento príklad je ako vždy k dispozícii na GitHub.