В этой статье основное внимание уделяется объяснению принципов! Если вы хотите увидеть размер партии Обработка данных Практический бой, пожалуйста.сосредоточиться ниже (пожалуйста, добавьте больше позже) на）：

Пример применения: очистка и преобразование данных

Очистка и преобразование данных с помощью Spring Batch

Пример приложения: импорт и экспорт данных

Импортируйте и экспортируйте данные с помощью Spring Batch.

Пример приложения: пакетная обработка запланированных задач

Используйте Spring Batch для реализации запланированных задач

Представляем Spring Batch

Spring Batch — это платформа пакетной обработки с открытым исходным кодом на основе Java, предназначенная для обработки крупномасштабных, повторяющихся и высоконадежных задач. Он обеспечивает простой и мощный способ выполнения заданий пакетной обработки, таких как импорт/экспорт данных, создание отчетов, пакетная обработка и т. д.

Что такое весенняя партия?

Spring Batch предназначен для упрощения разработки и управления пакетными заданиями. Он предоставляет расширяемую модель для определения и выполнения пакетных заданий, разделяя задание на несколько шагов (шагов), каждый из которых состоит из одного или нескольких блоков задач (чанков). Используя Spring Batch, можно легко обрабатывать большие объемы данных и сложную бизнес-логику.

Особенности и преимущества Spring Batch

1. Масштабируемость и возможность повторного использования：Spring Пакетная обработка имеет модульную конструкцию и обеспечивает множество возможностей Масштабируемости. и возможность повторного использование. Вы можете настроить процесс задания в соответствии с конкретными потребностями, добавлять или удалять этапы и гибко адаптироваться к различным сценариям пакетной обработки.
2. управление транзакциями：Spring Пакетная обработка обеспечивает мощное управление Механизм транзакций, обеспечивающий согласованность и целостность пакетных заданий. Границы транзакций можно настроить таким образом, чтобы каждый шаг или блок задач выполнялся в отдельной транзакции, что обеспечивает надежность работы.
3. Мониторинг и обработка ошибок：Spring Batch обеспечивает комплексный мониторинг и обработка механизм ошибок. Вы можете использовать прослушиватели и функции обратного вызова для мониторинга выполнения заданий, обработки ошибок и исключений, а также для записи и составления отчетов о состоянии и показателях задания.
4. параллельная обработка：Spring Batchподдерживатьпараллельная вы можете разделить задание на несколько независимых потоков или процессов для выполнения, повышая скорость обработки и эффективность задания.

Начало работы с Spring Batch

1. Установите и настройте Spring Batch.

Сначала убедитесь, что ваша среда разработки Java установлена ​​и настроена. Затем вы можете использовать инструмент сборки, такой как Maven или Gradle, чтобы добавить зависимости Spring Batch в ваш проект. Подробную информацию об установке и настройке см. в официальной документации Spring Batch.

2. Создайте первое пакетное задание.

В Spring Batch пакетное задание состоит из одного или нескольких шагов, а каждый шаг состоит из одного или нескольких блоков задач. Вот простой пример, показывающий, как создать простое пакетное задание:

@Configuration
@EnableBatchProcessing
public class BatchConfiguration {

    @Autowired
    private JobBuilderFactory jobBuilderFactory;

    @Autowired
    private StepBuilderFactory stepBuilderFactory;

    @Bean
    public Step step1() {
        return stepBuilderFactory.get("step1")
                .tasklet((contribution, chunkContext) -> {
                    System.out.println("Hello, Spring Batch!");
                    return RepeatStatus.FINISHED;
                })
                .build();
    }

    @Bean
    public Job job(Step step1) {
        return jobBuilderFactory.get("job")
                .start(step1)
                .build();
    }
}

анализировать

Первое использование@Configurationи@EnableBatchProcessingАннотации помечают классы какSpring Класс пакетной конфигурации. Затем,использоватьJobBuilderFactoryиStepBuilderFactoryСоздать работуистроитель шагов。существоватьstep1в методе,Определяет простой блок задач,напечатать «Здравствуйте! Spring Batch!"и вернутьсяRepeatStatus.FINISHED。наконец,существоватьjobв методе,использоватьjobBuilderFactoryСоздать работу,и будетstep1В качестве стартового шага к работе。

3. Понимание блоков заданий, шагов и задач.

• Работа：Задание — это независимая пакетная задача.,Состоит из одного или нескольких этапов. Он описывает ход и последовательность всего пакетного процесса.,И может иметь свои параметры и конфигурацию.
• Шаг (шаговый блок)：Шаги — это часть работы,Используется для выполнения определенных задач. Задание может содержать один или несколько шагов.,Каждый шаг может определять свои собственные задачи и логику обработки.
• Блок задач (Чанк)：Блок задач — это наименьшая единица выполнения шагов.,Используется для обработки определенного объема данных. Блоки задач делят данные на части для обработки.,Можно определить логику для чтения данных, обработки данных и записи данных.

Создание спроса: Предположим, у нас есть требование прочитать информацию о студентах из файла CSV, преобразовать и проверить оценки каждого студента и записать обработанную информацию о студентах в таблицу базы данных.

Обработка данных

Чтение и запись данных：Spring BatchПредоставляет разнообразныечитатьиписатьданныеспособ。МожетиспользоватьItemReaderчитатьданные,Например изданные Библиотека、очередь файлов или сообщенийсерединачитатьданные。ЗатемиспользоватьItemWriterбудет обработаноданные Написать цель,нравитьсяданные Библиотекаповерхность、очередь файлов или сообщений。 Во-первых, нам нужно определить модель данных для представления информации об учащихся, например

public class Student {
    private String name;
    private int score;

    // Getters and setters
    // ...
}

Далее мы можем использовать Spring BatchпредоставилFlatFileItemReaderПриходитьчитатьCSVв файледанные：

@Bean
public FlatFileItemReader<Student> studentItemReader() {
    FlatFileItemReader<Student> reader = new FlatFileItemReader<>();
    reader.setResource(new ClassPathResource("students.csv"));
    reader.setLineMapper(new DefaultLineMapper<Student>() {
        {
            setLineTokenizer(new DelimitedLineTokenizer() {
                {
                    setNames(new String[] { "name", "score" });
                }
            });
            setFieldSetMapper(new BeanWrapperFieldSetMapper<Student>() {
                {
                    setTargetType(Student.class);
                }
            });
        }
    });
    return reader;
}

Поддерживаемые форматы данных и источники данных

• Spring Batch поддерживает различные форматы данных и источники данных. Адаптеры, устройства чтения и записи могут использоваться для обработки различных форматов данных.,Такие как CSV, XML, JSON и т. д. в то же время,Различные источники данных могут обрабатываться с помощью пользовательских устройств чтения и записи данных.,Например, библиотека реляционных данных, библиотека данных NoSQL и т. д.

Преобразование и проверка данных

Spring Пакетные предложения и проверка данныхмеханизм。МожетиспользоватьItemProcessorверночитатьизданные Сделать преобразование、фильтрипроверять。ItemProcessorПользовательскую бизнес-логику можно применять для обработки каждогоданныеэлемент。

Мы настроилиFlatFileItemReader,Установите расположение и сопоставление строк для файлов CSV.,Указывает разделители полей и сопоставления атрибутов поля и модели.

Следующий,Мы можем определитьItemProcessorПриходитьверночитатьиз学生信息Преобразуйте и проверьте：

@Bean
public ItemProcessor<Student, Student> studentItemProcessor() {
    return new ItemProcessor<Student, Student>() {
        @Override
        public Student process(Student student) throws Exception {
            // Преобразуйте и проверьте
            if (student.getScore() < 0) {
                // Если проверка не удалась, выдается исключение.
                throw new IllegalArgumentException("Invalid score for student: " + student.getName());
            }
            // Операции преобразования, такие как преобразование дробей в проценты.
            int percentage = student.getScore() * 10;
            student.setScore(percentage);
            return student;
        }
    };
}

В приведенном выше коде,Мы определилиItemProcessor,Преобразование контрольной суммы информации об учениках. Если балл ученика меньше 0,в противном случае выдает исключение;,Преобразуйте дроби в проценты.

Наконец, мы можем использовать Spring BatchпредоставилJdbcBatchItemWriterбудет обработано学生信息писатьданные Библиотека：

@Bean
public JdbcBatchItemWriter<Student> studentItemWriter(DataSource dataSource) {
    JdbcBatchItemWriter<Student> writer = new JdbcBatchItemWriter<>();
    writer.setItemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<>());
    writer.setSql("INSERT INTO students (name, score) VALUES (:name, :score)");
    writer.setDataSource(dataSource);
    return writer;
}

Планирование и контроль работы

Конфигурация планировщика заданий：Spring Пакетная служба предоставляет планировщик заданий для настройки и управления выполнением пакетных заданий. Задания можно планировать с помощью среды планирования Spring (например, Quartz) или инструмента планирования операционной системы (например, cron). Настраивая планировщик заданий, вы можете установить время запуска задания, частоту и другие параметры планирования.

В приведенном выше коде,Мы настроилиJdbcBatchItemWriter,Настройка операторов SQL и источников данных,Пакетно вставляйте обработанную информацию о студентах в таблицу базы данных данных.

Наконец, нам нужно настроить шаг задания для сборки процесса чтения, обработки и записи данных:

@Bean
public Step processStudentStep(ItemReader<Student> reader, ItemProcessor<Student, Student> processor, ItemWriter<Student> writer) {
    return stepBuilderFactory.get("processStudentStep")
        .<Student, Student>chunk(10)
        .reader(reader)
        .processor(processor)
        .writer(writer)
        .build();
}

В приведенном выше коде,насиспользоватьstepBuilderFactoryсоздал шаг,И указаны считыватели, процессоры и записи данных.

Мониторинг и управление выполнением работ：Spring Пакетная обработка обеспечивает богатые функции мониторинга и управления. Можно использовать пружину Интерфейс пакетного управления и API для мониторинга состояния выполнения заданий, показателей прогресса и производительности. Вы также можете использовать механизмы ведения журнала, уведомлений и сигналов тревоги для своевременного получения информации о состоянии выполнения задания и исключениях.

Наконец, мы можем настроить задание для планирования этого шага:

@Bean
public Job processStudentJob(JobBuilderFactory jobBuilderFactory, Step processStudentStep) {
    return jobBuilderFactory.get("processStudentJob")
        .flow(processStudentStep)
        .end()
        .build();
}

насиспользоватьjobBuilderFactoryСоздал работу,и указанные шаги для выполнения.

В приведенном выше примере мы продемонстрировали Spring Batchсередина Чтение и запись данныхспособ,использовать ПонятноFlatFileItemReaderчитатьCSVдокумент,использовать ПонятноJdbcBatchItemWriterбудет обработано学生信息писатьданные Библиотека。в то же время,насиспользовать ПонятноItemProcessorверночитатьиз学生信息Преобразуйте и пожалуйста. Этот пример также показывает, как Spring Поддержка пакетной обработки различных источников данных и форматов данных, а также способы настройки и объединения этапов задания для выполнения всей задачи пакетной обработки.

Обработка ошибок и механизм повтора

Spring Пакетные предложения Обработка ошибок и механизм повтора для обеспечения стабильности и надежности пакетных заданий. Политики можно настроить для обработки ошибок и исключений во время чтения, обработки и записи. Количество повторных попыток, интервалы повторных попыток и стратегии обработки ошибок можно настроить для адаптации к различным сценариям возникновения ошибок и потребностям. первый,Мы можем установить стратегию обработки ошибок в конфигурации шага. Например,нас МожетиспользоватьSkipPolicyпропустить определенные исключения,илииспользоватьRetryPolicyпопробовать еще раз。

@Bean
public Step processStudentStep(ItemReader<Student> reader, ItemProcessor<Student, Student> processor, ItemWriter<Student> writer) {
    return stepBuilderFactory.get("processStudentStep")
        .<Student, Student>chunk(10)
        .reader(reader)
        .processor(processor)
        .writer(writer)
        .faultTolerant()
        .skip(Exception.class)
        .skipLimit(10)
        .retry(Exception.class)
        .retryLimit(3)
        .build();
}

насиспользоватьfaultTolerant()Способ включения политики обработки ошибок。Затем,использоватьskip(Exception.class)Укажите, чтобы пропустить определенные исключения,использоватьskipLimit(10)设置跳过из最大Второсортный Число10Второсортный。в то же время,использоватьretry(Exception.class)Укажите повторную попытку определенных исключений,использоватьretryLimit(3)Установите максимальное количество повторов3Второсортный。

По умолчанию, если во время чтения, обработки или записи возникает исключение, Spring Batch помечает элемент как элемент ошибки и пытается пропустить или повторить попытку, пока не будет достигнуто максимальное количество пропусков или повторов.

также,Вы также можете настроить обработчики ошибок для каждого шага.,Настройте логику обработки элементов ошибок. Например,МожетиспользоватьSkipListenerПриходить处理跳过изэлемент,использоватьRetryListenerПриходить处理重试изэлемент。

@Bean
public SkipListener<Student, Student> studentSkipListener() {
    return new SkipListener<Student, Student>() {
        @Override
        public void onSkipInRead(Throwable throwable) {
            // Обработка исключений, возникающих во время чтения
        }

        @Override
        public void onSkipInWrite(Student student, Throwable throwable) {
            // Обработка исключений, возникающих во время записи
        }

        @Override
        public void onSkipInProcess(Student student, Throwable throwable) {
            // Обработка исключений, возникающих во время обработки.
        }
    };
}

@Bean
public RetryListener studentRetryListener() {
    return new RetryListener() {
        @Override
        public <T, E extends Throwable> boolean open(RetryContext retryContext, RetryCallback<T, E> retryCallback) {
            // Логика выполнения перед повторной попыткой
            return true;
        }

        @Override
        public <T, E extends Throwable> void onError(RetryContext retryContext, RetryCallback<T, E> retryCallback, Throwable throwable) {
            // Обработка исключений, возникающих во время повторных попыток
        }

        @Override
        public <T, E extends Throwable> void close(RetryContext retryContext, RetryCallback<T, E> retryCallback, Throwable throwable) {
            // Логика, которая будет выполнена после повторной попытки
        }
    };
}

@Bean
public Step processStudentStep(ItemReader<Student> reader, ItemProcessor<Student, Student> processor, ItemWriter<Student> writer,
                               SkipListener<Student, Student> skipListener, RetryListener retryListener) {
    return stepBuilderFactory.get("processStudentStep")
        .<Student, Student>chunk(10)
        .reader(reader)
        .processor(processor)
        .writer(writer)
        .faultTolerant()
        .skip(Exception.class)
        .skipLimit(10)
        .retry(Exception.class)
        .retryLimit(3)
        .listener(skipListener)
        .listener(retryListener)
        .build();
}

Рекомендации по пакетной обработке

• Контроль объема данных：существоватьпакетное заданиесередина,Следует обратить внимание на контроль количества данных.,Чтобы избежать переполнения памяти или низкой скорости обработки. Объем данных можно контролировать с помощью обработки фрагментов и постраничного чтения.
• управление транзакциями：существоватьпакетное заданиесередина,Для операций, требующих согласованности и целостности данных.,отвечатьиспользовать适当изуправление механизм транзакций. Границы транзакций можно настроить так, чтобы каждый шаг или блок задач выполнялся в рамках независимой транзакции.
• Обработка и протоколирование ошибок：Разумный Обработка ошибок и исключений Является важной частью заданий пакетной обработки.。отвечатьиспользовать适当из错误处理策略、регистрацияи Механизм сигнализации,Чтобы своевременно обнаружить и устранить проблемы.
• Настройка производительности：существоватьпакетное заданиесередина,отвечатьсосредоточиться на Настройка вопрос производительности. Может пройти разумную параллельную обработка, правильно настроенные пулы потоков и соответствующее Чтение. и запись стратегии обработки данных для повышения скорости и эффективности обработки заданий.
• Контролируйте и управляйте：Для длительных пакетных заданий,отвечать设置适当из Контролируйте и управляйте механизмом. Вы можете использовать инструменты мониторинга, системы оповещений и инструменты автоматического управления задачами для мониторинга выполнения заданий и показателей производительности.

Продлить весенний пакет

Пользовательские считыватели, писатели и процессоры

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

public class MyItemReader implements ItemReader<String> {
    private List<String> data = Arrays.asList("item1", "item2", "item3");
    private Iterator<String> iterator = data.iterator();

    @Override
    public String read() throws Exception {
        if (iterator.hasNext()) {
            return iterator.next();
        } else {
            return null;
        }
    }
}

Пользовательский писатель:

public class MyItemWriter implements ItemWriter<String> {
    @Override
    public void write(List<? extends String> items) throws Exception {
        for (String item : items) {
            // Пользовательская логика записи
        }
    }
}

Пользовательский процессор:

public class MyItemProcessor implements ItemProcessor<String, String> {
    @Override
    public String process(String item) throws Exception {
        // Пользовательская логика обработки
        return item.toUpperCase();
    }
}

Параллельная обработка пакетных заданий:

Spring Batch поддерживает разделение заданий пакетной обработки на несколько независимых шагов и достижение параллельной обработки посредством многопоточности или распределенной обработки.

Многопоточная обработка. Многопоточная обработка может быть достигнута путем настройки TaskExecutor. Используя TaskExecutor, каждый шаг может выполняться в отдельном потоке, что обеспечивает параллельную обработку.

@Bean
public TaskExecutor taskExecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(5);
    executor.setMaxPoolSize(10);
    executor.setQueueCapacity(25);
    return executor;
}

@Bean
public Step myStep(ItemReader<String> reader, ItemProcessor<String, String> processor, ItemWriter<String> writer) {
    return stepBuilderFactory.get("myStep")
            .<String, String>chunk(10)
            .reader(reader)
            .processor(processor)
            .writer(writer)
            .taskExecutor(taskExecutor())
            .build();
}

В приведенном выше коде,мы проходимtaskExecutor()Метод определяет исполнителя задачи пула потоков.,и будетнастроить его на шагсерединаизtaskExecutor()в методе。

Распределенная обработка. Если требуется более высокий параллелизм и масштабируемость, рассмотрите возможность использования распределенной обработки. Spring Batch обеспечивает интеграцию с такими проектами, как Spring Integration и Spring Cloud Task, для обеспечения распределенного развертывания и обработки.

Интеграция с другими проектами Spring

1. Интеграция с Spring Integration:

Сначала вам необходимо настроить канал сообщений и адаптер Spring Integration в пакетном задании Spring. Вы можете использовать каналы сообщений для отправки и получения входных и выходных данных задания, а также адаптеры для взаимодействия с внешними системами.

@Configuration
@EnableBatchProcessing
@EnableIntegration
public class BatchConfiguration {

    @Autowired
    private JobBuilderFactory jobBuilderFactory;

    @Autowired
    private StepBuilderFactory stepBuilderFactory;

    @Autowired
    private MyItemReader reader;

    @Autowired
    private MyItemProcessor processor;

    @Autowired
    private MyItemWriter writer;

    @Bean
    public IntegrationFlow myJobFlow() {
        return IntegrationFlows.from("jobInputChannel")
                .handle(jobLaunchingGateway())
                .get();
    }

    @Bean
    public MessageChannel jobInputChannel() {
        return new DirectChannel();
    }

    @Bean
    public MessageChannel jobOutputChannel() {
        return new DirectChannel();
    }

    @Bean
    public MessageChannel stepInputChannel() {
        return new DirectChannel();
    }

    @Bean
    public MessageChannel stepOutputChannel() {
        return new DirectChannel();
    }

    @Bean
    public JobLaunchingGateway jobLaunchingGateway() {
        SimpleJobLauncher jobLauncher = new SimpleJobLauncher();
        jobLauncher.setJobRepository(jobRepository());
        return new JobLaunchingGateway(jobLauncher);
    }

    @Bean
    public JobRepository jobRepository() {
        // Настроить хранилище заданий
    }

    @Bean
    public Job myJob() {
        return jobBuilderFactory.get("myJob")
                .start(step1())
                .build();
    }

    @Bean
    public Step step1() {
        return stepBuilderFactory.get("step1")
                .<String, String>chunk(10)
                .reader(reader)
                .processor(processor)
                .writer(writer)
                .inputChannel(stepInputChannel())
                .outputChannel(stepOutputChannel())
                .build();
    }
}

В приведенном выше коде мы настраиваем Spring Batch作业из消息通道иадаптер。myJobFlow()метод определяет процесс интеграции,Он назван отjobInputChannelКанал сообщений получает запросы на работу,и пройтиjobLaunchingGateway()Способ начать работу。jobLaunchingGateway()метод создаетJobLaunchingGatewayПример,Используется для запуска задания.

Интеграция с Spring Cloud Задача:

Во-первых, вам необходимо настроить средство запуска задач и прослушиватель задач Spring Cloud Task в пакетном задании Spring. Средство запуска задач используется для запуска распределенных задач и управления ими, а прослушиватель задач — для выполнения некоторых операций во время выполнения задач.

@Configuration
@EnableBatchProcessing
@EnableTask
public class BatchConfiguration {

    @Autowired
    private JobBuilderFactory jobBuilderFactory;

    @Autowired
    private StepBuilderFactory stepBuilderFactory;

    @Autowired
    private MyItemReader reader;

    @Autowired
    private MyItemProcessor processor;

    @Autowired
    private MyItemWriter writer;

    @Bean
    public TaskConfigurer taskConfigurer() {
        return new DefaultTaskConfigurer();
    }

    @Bean
    public TaskExecutor taskExecutor() {
        return new SimpleAsyncTaskExecutor();
    }

    @Bean
    public Job myJob() {
        return jobBuilderFactory.get("myJob")
                .start(step1())
                .build();
    }

    @Bean
    public Step step1() {
        return stepBuilderFactory.get("step1")
                .<String, String>chunk(10)
                .reader(reader)
                .processor(processor)
                .writer(writer)
                .taskExecutor(taskExecutor())
                .build();
    }

    @Bean
    public TaskListener myTaskListener() {
        return new MyTaskListener();
    }

    @Bean
    public TaskExecutionListener myTaskExecutionListener() {
        return new MyTaskExecutionListener();
    }
}