streamalism.org

1. Úvod

V tomto článku sa zameriame na praktické úvodné slovo do Spring Batch zamerané na kód. Spring Batch je rámec spracovania určený pre robustné vykonávanie úloh.

Je to aktuálna verzia 3.0, ktorá podporuje Spring 4 a Java 8. Je tiež prispôsobená JSR-352, čo je nová špecifikácia Java pre dávkové spracovanie.

Tu uvádzam niekoľko zaujímavých a praktických prípadov použitia rámca.

2. Základy pracovného postupu

Jarná dávka sleduje tradičnú dávkovú architektúru, kde úložisko úloh vykonáva prácu s plánovaním a interakciou s úlohou.

Úloha môže mať viac ako jeden krok - a každý krok zvyčajne sleduje postupnosť načítania údajov, ich spracovania a zápisu.

Rámec tu samozrejme urobí väčšinu ťažkého zdvíhania - najmä pokiaľ ide o prácu na pretrvávaní nízkych úrovní pri riešení úloh - pomocou sqlite pre archív úloh.

2.1. Náš príklad použitia

Jednoduchý prípad použitia, ktorý tu budeme riešiť, je - budeme migrovať niektoré údaje o finančných transakciách z CSV do XML.

Vstupný súbor má veľmi jednoduchú štruktúru - obsahuje transakciu na riadok, ktorú tvoria: používateľské meno, ID používateľa, dátum transakcie a suma:

používateľské meno, identifikátor používateľa, dátum transakcie, suma transakcie devendra, 1234, 31. 10. 2015, 10 000 Ján, 2134, 3. 12. 2015, 12321 Robin, 2134, 2. 2. 2015, 23411

3. Maven POM

Závislosti požadované pre tento projekt sú jarné jadro, jarná dávka a sqlite jdbc konektor:

   org.xerial sqlite-jdbc 3.15.1 org.springframework spring-oxm 5.2.0.RELEASE org.springframework spring-jdbc 5.2.0.RELEASE org.springframework.batch spring-batch-core 4.2.0.RELEASE 

4. Konfigurácia jarnej dávky

Prvá vec, ktorú urobíme, je nakonfigurovať Spring Batch s XML:

K dispozícii je samozrejme aj konfigurácia Java:

@Configuration @EnableBatchProcessing verejná trieda SpringConfig {@Value ("org / springframework / batch / core / schema-drop-sqlite.sql") private Resource dropReopsitoryTables; @Value ("org / springframework / batch / core / schema-sqlite.sql") private Resource dataReopsitorySchema; @Bean public DataSource dataSource () {DriverManagerDataSource dataSource = nový DriverManagerDataSource (); dataSource.setDriverClassName ("org.sqlite.JDBC"); dataSource.setUrl ("jdbc: sqlite: repository.sqlite"); vrátiť dátový zdroj; } @Bean public DataSourceInitializer dataSourceInitializer (DataSource dataSource) hodí MalformedURLException {ResourceDatabasePopulator databasePopulator = nový ResourceDatabasePopulator (); databasePopulator.addScript (dropReopsitoryTables); databasePopulator.addScript (dataReopsitorySchema); databasePopulator.setIgnoreFailedDrops (true); DataSourceInitializer initializer = nový DataSourceInitializer (); initializer.setDataSource (dataSource); initializer.setDatabasePopulator (databasePopulator); návratový inicializátor; } private JobRepository getJobRepository () vyvolá výnimku {JobRepositoryFactoryBean factory = nový JobRepositoryFactoryBean (); factory.setDataSource (dataSource ()); factory.setTransactionManager (getTransactionManager ()); factory.afterPropertiesSet (); návrat (JobRepository) factory.getObject (); } private PlatformTransactionManager getTransactionManager () {return new ResourcelessTransactionManager (); } public JobLauncher getJobLauncher () hodí výnimku {SimpleJobLauncher jobLauncher = nový SimpleJobLauncher (); jobLauncher.setJobRepository (getJobRepository ()); jobLauncher.afterPropertiesSet (); vrátiť jobLauncher; }}

5. Konfigurácia jarnej dávkovej úlohy

Poďme teraz napísať náš popis práce pre prácu s CSV na XML:

                           com.baeldung.spring_batch_intro.model.Transakcia 

A samozrejme podobná konfigurácia úloh založená na prostredí Java:

verejná trieda SpringBatchConfig {@Autowired súkromné ​​úlohy JobBuilderFactory; @Autowired súkromné ​​kroky StepBuilderFactory; @Value ("input / record.csv") private Resource inputCsv; @Value ("súbor: xml / output.xml") súkromný zdroj outputXml; @Bean public ItemReader itemReader () hodí 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 (inputCsv); DefaultLineMapper lineMapper = nový DefaultLineMapper (); lineMapper.setLineTokenizer (tokenizer); lineMapper.setFieldSetMapper (nový RecordFieldSetMapper ()); reader.setLineMapper (lineMapper); spätná čítačka; } @Bean public ItemProcessor itemProcessor () {vrátiť nový CustomItemProcessor (); } @Bean public ItemWriter itemWriter (Marshaller marshaller) hodí MalformedURLException {StaxEventItemWriter itemWriter = nový StaxEventItemWriter (); itemWriter.setMarshaller (zriaďovateľ); itemWriter.setRootTagName ("transactionRecord"); itemWriter.setResource (outputXml); vrátiť itemWriter; } @Bean public Marshaller marshaller () {Jaxb2Marshaller marshaller = nový Jaxb2Marshaller (); marshaller.setClassesToBeBound (nová trieda [] {Transaction.class}); návratový radič; } @Bean chránený Krok krok 1 (čítačka ItemReader, procesor ItemProcessor, zapisovač ItemWriter) {návrat steps.get ("krok1"). chunk (10) .reader (reader) .processor (processor) .writer (writer) .build (); } @Bean (name = "firstBatchJob") verejná úloha Job (@Qualifier ("step1") Step step1) {return jobs.get ("firstBatchJob"). Start (step1) .build (); }}

Dobre, takže teraz, keď máme celú konfiguráciu, rozoberme ju a začneme o nej diskutovať.

5.1. Čítanie údajov a vytváranie objektov pomocou ItemReader

Najprv sme nakonfigurovali cvsFileItemReader ktorý načíta údaje z záznam.csv a previesť ho na Transakcia objekt:

@SuppressWarnings ("obmedzenie") @XmlRootElement (name = "transactionRecord") verejná trieda Transakcia {súkromné ​​reťazec meno používateľa; private int userId; private LocalDateTime transactionDate; súkromná dvojnásobná suma; / * vyhľadávače a nastavovatelia atribútov * / @Override public String toString () {return "Transaction [username =" + username + ", userId =" + userId + ", transactionDate =" + transactionDate + ", amount =" + suma + "]"; }}

Použije na to vlastný mapovač:

verejná trieda RecordFieldSetMapper implementuje FieldSetMapper {verejná transakcia mapFieldSet (FieldSet fieldSet) hodí BindException {DateTimeFormatter formatter = DateTimeFormatter.ofPattern ("d / M / rrr"); Transakčná transakcia = nová transakcia (); transaction.setUsername (fieldSet.readString ("používateľské meno")); transaction.setUserId (fieldSet.readInt (1)); transaction.setAmount (fieldSet.readDouble (3)); Reťazec dateString = fieldSet.readString (2); transaction.setTransactionDate (LocalDate.parse (dateString, formatter) .atStartOfDay ()); spätná transakcia; }}

5.2. Spracovanie údajov pomocou ItemProcessor

Vytvorili sme si vlastný procesor položiek, CustomItemProcessor. Toto nespracováva nič, čo súvisí s objektom transakcie - všetko, čo robí, je odovzdanie pôvodného objektu prichádzajúceho od čitateľa po zapisovateľa:

verejná trieda CustomItemProcessor implementuje ItemProcessor {public Transaction process (Transaction item) {return item; }}

5.3. Písanie predmetov na FS pomocou ItemWriter

Nakoniec to ideme uložiť transakcia do xml súboru umiestneného na xml / output.xml:

5.4. Konfigurácia dávkovej úlohy

Všetko, čo musíme urobiť, je spojiť bodky s úlohou - pomocou znaku dávka: práca syntax.

Všimnite si interval spáchania - to je počet transakcií, ktoré sa majú uchovať v pamäti pred pridelením dávky do itemWriter; transakcie bude držať v pamäti až do tohto bodu (alebo kým sa nestretne koniec vstupných údajov):

5.5. Spustenie dávkovej úlohy

To je všetko - poďme teraz všetko nastaviť a spustiť:

public class App {public static void main (String [] args) {// Spring Java config AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext (); context.register (SpringConfig.class); context.register (SpringBatchConfig.class); context.refresh (); JobLauncher jobLauncher = (JobLauncher) context.getBean ("jobLauncher"); Job job = (Job) context.getBean ("firstBatchJob"); System.out.println ("Spustenie dávkovej úlohy"); try {JobExecution execution = jobLauncher.run (job, new JobParameters ()); System.out.println ("Stav úlohy:" + execution.getStatus ()); System.out.println ("Úloha dokončená"); } catch (Výnimka e) {e.printStackTrace (); System.out.println ("Úloha zlyhala"); }}}

6. Záver

Tento tutoriál vám poskytne základnú predstavu o ako pracovať s Spring Batch a ako ho použiť v jednoduchom prípade použitia.

Ukazuje, ako môžete ľahko vytvoriť svoj kanál dávkového spracovania a ako môžete prispôsobiť rôzne fázy čítania, spracovania a zápisu.

The úplná implementácia tohto tutoriálu nájdete v projekte github - toto je projekt založený na Eclipse, takže by malo byť ľahké ho importovať a spustiť tak, ako je.