2021-11 spring-26-spring-batch added

2026-05-30 10:50:42 +00:00 · 2022-03-17 02:47:40 +04:00
parent e70ac395c2
commit a02bfa2629
17 changed files with 706 additions and 0 deletions
@@ -0,0 +1,9 @@
+.idea/
+*.iml
+
+target/
+
+output.csv
+*.log
+output*.dat
+test-output.dat
@@ -0,0 +1,16 @@
+Ivan,23
+John,24
+Ivan,23
+John,24
+Ivan,23
+Mary,24
+Ivan,23
+John,24
+Sunny,23
+John,24
+Ivan,23
+John,24
+Ivan,23
+John,24
+Ivan,23
+John,24
@@ -0,0 +1,106 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <groupId>ru.otus.example</groupId>
+    <artifactId>spring-batch-demo</artifactId>
+    <version>1.0-SNAPSHOT</version>
+
+    <parent>
+        <groupId>org.springframework.boot</groupId>
+        <artifactId>spring-boot-starter-parent</artifactId>
+        <version>2.6.4</version>
+    </parent>
+
+    <properties>
+        <java.version>11</java.version>
+        <maven.compiler.sourcre>11</maven.compiler.sourcre>
+        <maven.compiler.target>11</maven.compiler.target>
+        <mongock.version>4.3.8</mongock.version>
+    </properties>
+
+
+    <dependencies>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-batch</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>com.h2database</groupId>
+            <artifactId>h2</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-data-mongodb</artifactId>
+        </dependency>
+
+
+        <dependency>
+            <groupId>de.flapdoodle.embed</groupId>
+            <artifactId>de.flapdoodle.embed.mongo</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>com.github.cloudyrock.mongock</groupId>
+            <artifactId>mongock-spring-v5</artifactId>
+            <version>${mongock.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>com.github.cloudyrock.mongock</groupId>
+            <artifactId>mongodb-springdata-v3-driver</artifactId>
+            <version>${mongock.version}</version>
+        </dependency>
+
+<!--
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-web</artifactId>
+        </dependency>
+-->
+
+        <dependency>
+            <groupId>org.projectlombok</groupId>
+            <artifactId>lombok</artifactId>
+            <optional>true</optional>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.shell</groupId>
+            <artifactId>spring-shell-starter</artifactId>
+            <version>2.0.1.RELEASE</version>
+        </dependency>
+
+        <!--Тестирование-->
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.batch</groupId>
+            <artifactId>spring-batch-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <finalName>spring-batch-demo</finalName>
+        <plugins>
+            <plugin>
+                <groupId>org.springframework.boot</groupId>
+                <artifactId>spring-boot-maven-plugin</artifactId>
+            </plugin>
+        </plugins>
+    </build>
+</project>
@@ -0,0 +1,19 @@
+package ru.otus.example.springbatch;
+
+import com.github.cloudyrock.spring.v5.EnableMongock;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.context.annotation.PropertySource;
+
+import java.io.IOException;
+
+@EnableMongock
+@SpringBootApplication
+public class Main {
+
+    public static void main(String[] args) {
+        SpringApplication.run(Main.class, args);
+    }
+}
+
+
@@ -0,0 +1,45 @@
+package ru.otus.example.springbatch.chandgelogs;
+
+import com.github.cloudyrock.mongock.ChangeLog;
+import com.github.cloudyrock.mongock.ChangeSet;
+import com.github.cloudyrock.mongock.driver.mongodb.springdata.v3.decorator.impl.MongockTemplate;
+import com.mongodb.client.MongoDatabase;
+import ru.otus.example.springbatch.model.Person;
+
+@ChangeLog(order = "001")
+public class InitMongoDBDataChangeLog {
+
+    @ChangeSet(order = "000", id = "dropDB", author = "stvort", runAlways = true)
+    public void dropDB(MongoDatabase database){
+        database.drop();
+    }
+
+    @ChangeSet(order = "001", id = "initPersons", author = "stvort", runAlways = true)
+    public void initPersons(MongockTemplate template){
+        template.save(new Person("Джон", 21));
+        template.save(new Person("Игорь", 32));
+        template.save(new Person("Дмитрий", 52));
+        template.save(new Person("Михаил", 22));
+        template.save(new Person("Герман", 33));
+        template.save(new Person("Джон", 21));
+        template.save(new Person("Игорь", 32));
+        template.save(new Person("Дмитрий", 52));
+        template.save(new Person("Михаил", 22));
+        template.save(new Person("Герман", 33));
+        template.save(new Person("Джон", 21));
+        template.save(new Person("Игорь", 32));
+        template.save(new Person("Дмитрий", 52));
+        template.save(new Person("Михаил", 22));
+        template.save(new Person("Герман", 33));
+        template.save(new Person("Джон", 21));
+        template.save(new Person("Игорь", 32));
+        template.save(new Person("Дмитрий", 52));
+        template.save(new Person("Михаил", 22));
+        template.save(new Person("Герман", 33));
+        template.save(new Person("Джон", 21));
+        template.save(new Person("Игорь", 32));
+        template.save(new Person("Дмитрий", 52));
+        template.save(new Person("Михаил", 22));
+        template.save(new Person("Герман", 33));
+    }
+}
@@ -0,0 +1,14 @@
+package ru.otus.example.springbatch.config;
+
+import lombok.Data;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.stereotype.Component;
+
+@Data
+@Component
+@ConfigurationProperties("app")
+public class AppProps {
+    private String inputFile;
+    private String outputFile;
+
+}
@@ -0,0 +1,19 @@
+package ru.otus.example.springbatch.config;
+
+import org.springframework.batch.core.configuration.JobRegistry;
+import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
+import org.springframework.batch.core.configuration.support.JobRegistryBeanPostProcessor;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+@EnableBatchProcessing
+@Configuration
+public class BatchConfig {
+    @Bean
+    public JobRegistryBeanPostProcessor postProcessor(JobRegistry jobRegistry) {
+        var processor = new JobRegistryBeanPostProcessor();
+        processor.setJobRegistry(jobRegistry);
+        return processor;
+    }
+
+}
@@ -0,0 +1,193 @@
+package ru.otus.example.springbatch.config;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.batch.core.*;
+import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
+import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
+import org.springframework.batch.core.configuration.annotation.StepScope;
+import org.springframework.batch.core.launch.support.RunIdIncrementer;
+import org.springframework.batch.core.scope.context.ChunkContext;
+import org.springframework.batch.core.step.tasklet.MethodInvokingTaskletAdapter;
+import org.springframework.batch.item.ItemProcessor;
+import org.springframework.batch.item.ItemReader;
+import org.springframework.batch.item.file.FlatFileItemReader;
+import org.springframework.batch.item.file.FlatFileItemWriter;
+import org.springframework.batch.item.file.builder.FlatFileItemReaderBuilder;
+import org.springframework.batch.item.file.builder.FlatFileItemWriterBuilder;
+import org.springframework.batch.item.file.transform.DelimitedLineAggregator;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.core.io.FileSystemResource;
+import org.springframework.lang.NonNull;
+import ru.otus.example.springbatch.model.Person;
+import ru.otus.example.springbatch.service.CleanUpService;
+import ru.otus.example.springbatch.service.HappyBirthdayService;
+
+import java.util.List;
+
+
+@Configuration
+public class JobConfig {
+    private static final int CHUNK_SIZE = 5;
+    private final Logger logger = LoggerFactory.getLogger("Batch");
+
+    public static final String OUTPUT_FILE_NAME = "outputFileName";
+    public static final String INPUT_FILE_NAME = "inputFileName";
+    public static final String IMPORT_USER_JOB_NAME = "importUserJob";
+
+    @Autowired
+    private JobBuilderFactory jobBuilderFactory;
+
+    @Autowired
+    private StepBuilderFactory stepBuilderFactory;
+
+    @Autowired
+    private CleanUpService cleanUpService;
+
+    @StepScope
+    @Bean
+    public FlatFileItemReader<Person> reader(@Value("#{jobParameters['" + INPUT_FILE_NAME + "']}") String inputFileName) {
+        return new FlatFileItemReaderBuilder<Person>()
+                .name("personItemReader")
+                .resource(new FileSystemResource(inputFileName))
+
+                // Работа через lineMapper
+                .lineMapper((s, i) -> {
+                    String[] fieldsValues = s.split(",");
+                    return new Person(fieldsValues[0], Integer.parseInt(fieldsValues[1]));
+                })
+/*
+
+                // Работа через fieldSetMapper
+                .delimited()
+                .names("name", "age")
+                .fieldSetMapper(new BeanWrapperFieldSetMapper<>() {{
+                    setTargetType(Person.class);
+                }})
+*/
+                .build();
+    }
+
+    @StepScope
+    @Bean
+    public ItemProcessor<Person, Person> processor(HappyBirthdayService happyBirthdayService) {
+        return happyBirthdayService::doHappyBirthday;
+    }
+
+    @StepScope
+    @Bean
+    public FlatFileItemWriter<Person> writer(@Value("#{jobParameters['" + OUTPUT_FILE_NAME + "']}") String outputFileName) {
+        return new FlatFileItemWriterBuilder<Person>()
+                .name("personItemWriter")
+                .resource(new FileSystemResource(outputFileName))
+                .lineAggregator(new DelimitedLineAggregator<>())
+                .build();
+    }
+
+
+    @Bean
+    public MethodInvokingTaskletAdapter cleanUpTasklet() {
+        MethodInvokingTaskletAdapter adapter = new MethodInvokingTaskletAdapter();
+
+        adapter.setTargetObject(cleanUpService);
+        adapter.setTargetMethod("cleanUp");
+
+        return adapter;
+    }
+
+
+    @Bean
+    public Job importUserJob(Step transformPersonsStep, Step cleanUpStep) {
+        return jobBuilderFactory.get(IMPORT_USER_JOB_NAME)
+                .incrementer(new RunIdIncrementer())
+                .flow(transformPersonsStep)
+                .next(cleanUpStep)
+                .end()
+                .listener(new JobExecutionListener() {
+                    @Override
+                    public void beforeJob(@NonNull JobExecution jobExecution) {
+                        logger.info("Начало job");
+                    }
+
+                    @Override
+                    public void afterJob(@NonNull JobExecution jobExecution) {
+                        logger.info("Конец job");
+                    }
+                })
+                .build();
+    }
+
+    @Bean
+    public Step transformPersonsStep(ItemReader<Person> reader, FlatFileItemWriter<Person> writer,
+                                     ItemProcessor<Person, Person> itemProcessor) {
+        return stepBuilderFactory.get("step1")
+                .<Person, Person>chunk(CHUNK_SIZE)
+                .reader(reader)
+                .processor(itemProcessor)
+                .writer(writer)
+                .listener(new ItemReadListener<>() {
+                    public void beforeRead() {
+                        logger.info("Начало чтения");
+                    }
+
+                    public void afterRead(@NonNull Person o) {
+                        logger.info("Конец чтения");
+                    }
+
+                    public void onReadError(@NonNull Exception e) {
+                        logger.info("Ошибка чтения");
+                    }
+                })
+                .listener(new ItemWriteListener<>() {
+                    public void beforeWrite(@NonNull List list) {
+                        logger.info("Начало записи");
+                    }
+
+                    public void afterWrite(@NonNull List list) {
+                        logger.info("Конец записи");
+                    }
+
+                    public void onWriteError(@NonNull Exception e, @NonNull List list) {
+                        logger.info("Ошибка записи");
+                    }
+                })
+                .listener(new ItemProcessListener<>() {
+                    public void beforeProcess(Person o) {
+                        logger.info("Начало обработки");
+                    }
+
+                    public void afterProcess(@NonNull Person o, Person o2) {
+                        logger.info("Конец обработки");
+                    }
+
+                    public void onProcessError(@NonNull Person o, @NonNull Exception e) {
+                        logger.info("Ошибка обработки");
+                    }
+                })
+                .listener(new ChunkListener() {
+                    public void beforeChunk(@NonNull ChunkContext chunkContext) {
+                        logger.info("Начало пачки");
+                    }
+
+                    public void afterChunk(@NonNull ChunkContext chunkContext) {
+                        logger.info("Конец пачки");
+                    }
+
+                    public void afterChunkError(@NonNull ChunkContext chunkContext) {
+                        logger.info("Ошибка пачки");
+                    }
+                })
+//                .taskExecutor(new SimpleAsyncTaskExecutor())
+                .build();
+    }
+
+    @Bean
+    public Step cleanUpStep() {
+        return this.stepBuilderFactory.get("cleanUpStep")
+                .tasklet(cleanUpTasklet())
+                .build();
+    }
+}
@@ -0,0 +1,13 @@
+package ru.otus.example.springbatch.model;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@NoArgsConstructor
+@AllArgsConstructor
+@Data
+public class Person {
+    private String name;
+    private int age;
+}
@@ -0,0 +1,12 @@
+package ru.otus.example.springbatch.service;
+
+import org.springframework.stereotype.Service;
+
+@Service
+public class CleanUpService {
+    public void cleanUp() throws Exception {
+        System.out.println("Выполняю завершающие мероприятия...");
+        Thread.sleep(1000);
+        System.out.println("Завершающие мероприятия закончены");
+    }
+}
@@ -0,0 +1,13 @@
+package ru.otus.example.springbatch.service;
+
+import org.springframework.stereotype.Service;
+import ru.otus.example.springbatch.model.Person;
+
+@Service
+public class HappyBirthdayService {
+
+    public Person doHappyBirthday(Person person){
+        person.setAge(person.getAge() + 1);
+        return person;
+    }
+}
@@ -0,0 +1,60 @@
+package ru.otus.example.springbatch.shell;
+
+import lombok.RequiredArgsConstructor;
+import lombok.SneakyThrows;
+import org.springframework.batch.core.Job;
+import org.springframework.batch.core.JobExecution;
+import org.springframework.batch.core.JobParametersBuilder;
+import org.springframework.batch.core.JobParametersInvalidException;
+import org.springframework.batch.core.explore.JobExplorer;
+import org.springframework.batch.core.launch.JobInstanceAlreadyExistsException;
+import org.springframework.batch.core.launch.JobLauncher;
+import org.springframework.batch.core.launch.JobOperator;
+import org.springframework.batch.core.launch.NoSuchJobException;
+import org.springframework.batch.core.repository.JobExecutionAlreadyRunningException;
+import org.springframework.batch.core.repository.JobInstanceAlreadyCompleteException;
+import org.springframework.batch.core.repository.JobRestartException;
+import org.springframework.shell.standard.ShellComponent;
+import org.springframework.shell.standard.ShellMethod;
+import ru.otus.example.springbatch.config.AppProps;
+
+
+import static ru.otus.example.springbatch.config.JobConfig.*;
+
+@RequiredArgsConstructor
+@ShellComponent
+public class BatchCommands {
+
+    private final AppProps appProps;
+    private final Job importUserJob;
+
+    private final JobLauncher jobLauncher;
+    private final JobOperator jobOperator;
+    private final JobExplorer jobExplorer;
+
+    //http://localhost:8080/h2-console/
+
+    @ShellMethod(value = "startMigrationJobWithJobLauncher", key = "sm-jl")
+    public void startMigrationJobWithJobLauncher() throws Exception {
+        JobExecution execution = jobLauncher.run(importUserJob, new JobParametersBuilder()
+                .addString(INPUT_FILE_NAME, appProps.getInputFile())
+                .addString(OUTPUT_FILE_NAME, appProps.getOutputFile())
+                .toJobParameters());
+        System.out.println(execution);
+    }
+
+    @ShellMethod(value = "startMigrationJobWithJobOperator", key = "sm-jo")
+    public void startMigrationJobWithJobOperator() throws Exception {
+        Long executionId = jobOperator.start(IMPORT_USER_JOB_NAME,
+                INPUT_FILE_NAME + "=" + appProps.getInputFile() + "\n" +
+                        OUTPUT_FILE_NAME + "=" + appProps.getOutputFile()
+        );
+        System.out.println(jobOperator.getSummary(executionId));
+    }
+
+    @ShellMethod(value = "showInfo", key = "i")
+    public void showInfo() {
+        System.out.println(jobExplorer.getJobNames());
+        System.out.println(jobExplorer.getLastJobInstance(IMPORT_USER_JOB_NAME));
+    }
+}
@@ -0,0 +1,47 @@
+spring:
+  main:
+    allow-circular-references: true
+
+  batch:
+    job:
+      enabled: false
+  shell:
+    interactive:
+      enabled: true
+
+  mongodb:
+    embedded:
+      version: 3.4.0
+
+  data:
+    mongodb:
+      host: localhost
+      port: 0
+      database: SpringBatchExampleDB
+
+  datasource:
+    url: jdbc:h2:mem:testdb
+    driverClassName: org.h2.Driver
+    username: sa
+    password:
+
+  h2:
+    console:
+      enabled: true
+      path: /h2-console
+
+mongock:
+  runner-type: "InitializingBean"
+  change-logs-scan-package:
+    - ru.otus.example.springbatch.chandgelogs
+  mongo-db:
+    write-concern:
+      journal: false
+    read-concern: local
+
+app:
+  ages-count-to-add: 1
+  input-file: entries.csv
+  output-file: output.dat
+
+#debug: true
@@ -0,0 +1,73 @@
+package ru.otus.example.springbatch.config;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.batch.core.Job;
+import org.springframework.batch.core.JobExecution;
+import org.springframework.batch.core.JobParameters;
+import org.springframework.batch.core.JobParametersBuilder;
+import org.springframework.batch.test.AssertFile;
+import org.springframework.batch.test.JobLauncherTestUtils;
+import org.springframework.batch.test.JobRepositoryTestUtils;
+import org.springframework.batch.test.context.SpringBatchTest;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.core.io.FileSystemResource;
+
+
+import java.net.URLDecoder;
+import java.nio.charset.StandardCharsets;
+import java.util.Objects;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static ru.otus.example.springbatch.config.JobConfig.*;
+
+@SpringBootTest
+@SpringBatchTest
+class ImportUserJobTest {
+
+    private static final String TEST_INPUT_FILE_NAME = "test-entries.csv";
+    private static final String EXPECTED_OUTPUT_FILE_NAME = "expected-test-output.dat";
+    private static final String TEST_OUTPUT_FILE_NAME = "test-output.dat";
+
+    @Autowired
+    private JobLauncherTestUtils jobLauncherTestUtils;
+
+    @Autowired
+    private JobRepositoryTestUtils jobRepositoryTestUtils;
+
+    @BeforeEach
+    void clearMetaData() {
+        jobRepositoryTestUtils.removeJobExecutions();
+    }
+
+    @Test
+    void testJob() throws Exception {
+        var classLoader = ImportUserJobTest.class.getClassLoader();
+        var testInputFileName = URLDecoder.decode(
+                Objects.requireNonNull(classLoader.getResource(TEST_INPUT_FILE_NAME)).getFile(),
+                StandardCharsets.UTF_8
+        );
+        var expectedResultFileName = URLDecoder.decode(
+                Objects.requireNonNull(classLoader.getResource(EXPECTED_OUTPUT_FILE_NAME)).getFile(),
+                StandardCharsets.UTF_8
+        );
+
+        FileSystemResource expectedResult = new FileSystemResource(expectedResultFileName);
+        FileSystemResource actualResult = new FileSystemResource(TEST_OUTPUT_FILE_NAME);
+
+        Job job = jobLauncherTestUtils.getJob();
+        assertThat(job).isNotNull()
+                .extracting(Job::getName)
+                .isEqualTo(IMPORT_USER_JOB_NAME);
+
+        JobParameters parameters = new JobParametersBuilder()
+                .addString(INPUT_FILE_NAME, testInputFileName)
+                .addString(OUTPUT_FILE_NAME, TEST_OUTPUT_FILE_NAME)
+                .toJobParameters();
+        JobExecution jobExecution = jobLauncherTestUtils.launchJob(parameters);
+
+        assertThat(jobExecution.getExitStatus().getExitCode()).isEqualTo("COMPLETED");
+        AssertFile.assertFileEquals(expectedResult, actualResult);
+    }
+}
@@ -0,0 +1,35 @@
+spring:
+  main:
+    allow-circular-references: true
+
+  batch:
+    job:
+      enabled: false
+  shell:
+    interactive:
+      enabled: false
+
+  datasource:
+    url: jdbc:h2:mem:testdb
+    driverClassName: org.h2.Driver
+    username: sa
+    password:
+
+  mongodb:
+    embedded:
+      version: 3.4.0
+
+  data:
+    mongodb:
+      host: localhost
+      port: 0
+      database: SpringBatchTestExampleDB
+
+mongock:
+  runner-type: "InitializingBean"
+  change-logs-scan-package:
+    - ru.otus.example.springbatch.chandgelogs
+  mongo-db:
+    write-concern:
+      journal: false
+    read-concern: local
@@ -0,0 +1,16 @@
+Person(name=Ivan, age=24)
+Person(name=John, age=25)
+Person(name=Ivan, age=24)
+Person(name=John, age=25)
+Person(name=Ivan, age=24)
+Person(name=Mary, age=25)
+Person(name=Ivan, age=24)
+Person(name=John, age=25)
+Person(name=Sunny, age=24)
+Person(name=John, age=25)
+Person(name=Ivan, age=24)
+Person(name=John, age=25)
+Person(name=Ivan, age=24)
+Person(name=John, age=25)
+Person(name=Ivan, age=24)
+Person(name=John, age=25)
@@ -0,0 +1,16 @@
+Ivan,23
+John,24
+Ivan,23
+John,24
+Ivan,23
+Mary,24
+Ivan,23
+John,24
+Sunny,23
+John,24
+Ivan,23
+John,24
+Ivan,23
+John,24
+Ivan,23
+John,24