SpringBoot：使用Spring Batch实现批处理任务

devtools/2024/10/17 20:33:04/

引言

在这里插入图片描述

在企业级应用中，批处理任务是不可或缺的一部分。它们通常用于处理大量数据，如数据迁移、数据清洗、生成报告等。Spring Batch是Spring框架的一部分，专为批处理任务设计，提供了简化的配置和强大的功能。本文将介绍如何使用Spring Batch与SpringBoot结合，构建和管理批处理任务。

项目初始化

首先，我们需要创建一个SpringBoot项目，并添加Spring Batch相关的依赖项。可以通过Spring Initializr快速生成项目。

添加依赖

在pom.xml中添加以下依赖：

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-batch</artifactId>
</dependency>
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-jpa</artifactId>
</dependency>
<dependency><groupId>org.hsqldb</groupId><artifactId>hsqldb</artifactId><scope>runtime</scope>
</dependency>

配置Spring Batch

基本配置

Spring Batch需要一个数据库来存储批处理的元数据。我们可以使用HSQLDB作为内存数据库。配置文件application.properties：

spring.datasource.url=jdbc:hsqldb:mem:testdb
spring.datasource.driverClassName=org.hsqldb.jdbc.JDBCDriver
spring.datasource.username=sa
spring.datasource.password=
spring.batch.initialize-schema=always
创建批处理任务

一个典型的Spring Batch任务包括三个主要部分：ItemReader、ItemProcessor和ItemWriter。

  1. ItemReader：读取数据的接口。
  2. ItemProcessor：处理数据的接口。
  3. ItemWriter：写数据的接口。
创建示例实体类

创建一个示例实体类，用于演示批处理操作：

import javax.persistence.Entity;
import javax.persistence.GeneratedValue;
import javax.persistence.GenerationType;
import javax.persistence.Id;@Entity
public class Person {@Id@GeneratedValue(strategy = GenerationType.IDENTITY)private Long id;private String firstName;private String lastName;// getters and setters
}
创建ItemReader

我们将使用一个简单的FlatFileItemReader从CSV文件中读取数据：

import org.springframework.batch.item.file.FlatFileItemReader;
import org.springframework.batch.item.file.builder.FlatFileItemReaderBuilder;
import org.springframework.batch.item.file.mapping.BeanWrapperFieldSetMapper;
import org.springframework.batch.item.file.mapping.DefaultLineMapper;
import org.springframework.batch.item.file.mapping.DelimitedLineTokenizer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.ClassPathResource;@Configuration
public class BatchConfiguration {@Beanpublic FlatFileItemReader<Person> reader() {return new FlatFileItemReaderBuilder<Person>().name("personItemReader").resource(new ClassPathResource("sample-data.csv")).delimited().names(new String[]{"firstName", "lastName"}).fieldSetMapper(new BeanWrapperFieldSetMapper<Person>() {{setTargetType(Person.class);}}).build();}
}
创建ItemProcessor

创建一个简单的ItemProcessor，将读取的数据进行处理：

import org.springframework.batch.item.ItemProcessor;
import org.springframework.stereotype.Component;@Component
public class PersonItemProcessor implements ItemProcessor<Person, Person> {@Overridepublic Person process(Person person) throws Exception {final String firstName = person.getFirstName().toUpperCase();final String lastName = person.getLastName().toUpperCase();final Person transformedPerson = new Person();transformedPerson.setFirstName(firstName);transformedPerson.setLastName(lastName);return transformedPerson;}
}
创建ItemWriter

我们将使用一个简单的JdbcBatchItemWriter将处理后的数据写入数据库：

import org.springframework.batch.item.database.BeanPropertyItemSqlParameterSourceProvider;
import org.springframework.batch.item.database.JdbcBatchItemWriter;
import org.springframework.batch.item.database.builder.JdbcBatchItemWriterBuilder;
import org.springframework.context.annotation.Bean;
import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;@Configuration
public class BatchConfiguration {@Beanpublic JdbcBatchItemWriter<Person> writer(NamedParameterJdbcTemplate jdbcTemplate) {return new JdbcBatchItemWriterBuilder<Person>().itemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<>()).sql("INSERT INTO person (first_name, last_name) VALUES (:firstName, :lastName)").dataSource(jdbcTemplate.getJdbcTemplate().getDataSource()).build();}
}

配置Job和Step

一个Job由多个Step组成，每个Step包含一个ItemReader、ItemProcessor和ItemWriter。

import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
@EnableBatchProcessing
public class BatchConfiguration {@Autowiredpublic JobBuilderFactory jobBuilderFactory;@Autowiredpublic StepBuilderFactory stepBuilderFactory;@Beanpublic Job importUserJob(JobCompletionNotificationListener listener, Step step1) {return jobBuilderFactory.get("importUserJob").listener(listener).flow(step1).end().build();}@Beanpublic Step step1(JdbcBatchItemWriter<Person> writer) {return stepBuilderFactory.get("step1").<Person, Person>chunk(10).reader(reader()).processor(processor()).writer(writer).build();}
}

监听Job完成事件

创建一个监听器，用于监听Job完成事件：

import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobExecutionListener;
import org.springframework.stereotype.Component;@Component
public class JobCompletionNotificationListener implements JobExecutionListener {@Overridepublic void beforeJob(JobExecution jobExecution) {System.out.println("Job Started");}@Overridepublic void afterJob(JobExecution jobExecution) {System.out.println("Job Ended");}
}

测试与运行

创建一个简单的CommandLineRunner，用于启动批处理任务：

import org.springframework.batch.core.Job;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;@SpringBootApplication
public class BatchApplication implements CommandLineRunner {@Autowiredprivate JobLauncher jobLauncher;@Autowiredprivate Job job;public static void main(String[] args) {SpringApplication.run(BatchApplication.class, args);}@Overridepublic void run(String... args) throws Exception {jobLauncher.run(job, new JobParameters());}
}

在完成配置后，可以运行应用程序，并检查控制台输出和数据库中的数据，确保批处理任务正常运行。

扩展功能

在基本的批处理任务基础上，可以进一步扩展功能，使其更加完善和实用。例如：

  • 多步骤批处理：一个Job可以包含多个Step，每个Step可以有不同的ItemReader、ItemProcessor和ItemWriter。
  • 并行处理：通过配置多个线程或分布式处理，提升批处理任务的性能。
  • 错误处理和重试：配置错误处理和重试机制，提高批处理任务的可靠性。
  • 数据验证：在处理数据前进行数据验证，确保数据的正确性。
多步骤批处理
@Bean
public Job multiStepJob(JobCompletionNotificationListener listener, Step step1, Step step2) {return jobBuilderFactory.get("multiStepJob").listener(listener).start(step1).next(step2).end().build();
}@Bean
public Step step2(JdbcBatchItemWriter<Person> writer) {return stepBuilderFactory.get("step2").<Person, Person>chunk(10).reader(reader()).processor(processor()).writer(writer).build();
}
并行处理

可以通过配置多个线程来实现并行处理：

@Bean
public Step step1(JdbcBatchItemWriter<Person> writer) {return stepBuilderFactory.get("step1").<Person, Person>chunk(10).reader(reader()).processor(processor()).writer(writer).taskExecutor(taskExecutor()).build();
}@Bean
public TaskExecutor taskExecutor() {SimpleAsyncTaskExecutor taskExecutor = new SimpleAsyncTaskExecutor();taskExecutor.setConcurrencyLimit(10);return taskExecutor;
}

结论

通过本文的介绍，我们了解了如何使用Spring Batch与SpringBoot结合，构建和管理批处理任务。从项目初始化、配置Spring Batch、实现ItemReader、ItemProcessor和ItemWriter，到配置Job和Step，Spring Batch提供了一系列强大的工具和框架，帮助开发者高效地实现批处理任务。通过合理利用这些工具和框架

，开发者可以构建出高性能、可靠且易维护的批处理系统。希望这篇文章能够帮助开发者更好地理解和使用Spring Batch，在实际项目中实现批处理任务的目标。


http://www.ppmy.cn/devtools/56489.html

相关文章

【高考志愿】集成电路科学与工程

【高考志愿】集成电路科学与工程

目录 一、专业概述 二、课程设置 三、就业前景 四、适合人群 五、院校推荐 六、集成电路科学与工程专业排名 一、专业概述 集成电路科学与工程&#xff0c;这一新兴且引人注目的交叉学科&#xff0c;正在逐渐崭露头角。它集合了电子工程、计算机科学、材料科学等多个领域的…
阅读更多...
Nginx

Nginx

一、Nginx配置文件 1.1 主配置文件 主配置文件位置&#xff1a;nginx.conf tip&#xff1a;安装方式不同&#xff0c;路径不同 #主配置文件格式main block&#xff1a;主配置段&#xff0c;即全局配置段&#xff0c;对http,mail都有效#配置Nginx服务器的事件模块相关参数 e…
阅读更多...
深入理解外观模式（Facade Pattern）及其实际应用

深入理解外观模式（Facade Pattern）及其实际应用

引言 在软件开发中&#xff0c;复杂的系统往往由多个子系统组成&#xff0c;这些子系统之间的交互可能非常复杂。外观模式&#xff08;Facade Pattern&#xff09;通过为这些子系统提供一个统一的接口&#xff0c;简化了它们的交互。本篇文章将详细介绍外观模式的概念、应用场…
阅读更多...
基于Java游戏售卖网站详细设计和实现(源码+LW+调试文档+讲解等)

基于Java游戏售卖网站详细设计和实现(源码+LW+调试文档+讲解等)

&#x1f497;博主介绍&#xff1a;✌全网粉丝10W,CSDN作者、博客专家、全栈领域优质创作者&#xff0c;博客之星、平台优质作者、专注于Java、小程序技术领域和毕业项目实战✌&#x1f497; &#x1f31f;文末获取源码数据库&#x1f31f; 感兴趣的可以先收藏起来&#xff0c;…
阅读更多...
仙女山感怀

仙女山感怀

仙女山感怀 群山寻仙径&#xff0c;洞府云深处。 ​静心修金身&#xff0c;界域真神铸。 ​天地大道中&#xff0c;微光聚生路。 ​远幽诗书画&#xff0c;近翠色相无。
阅读更多...
Objects and Classes (对象和类)

Objects and Classes (对象和类)

Objects and Classes [对象和类] 1. Procedural and Object-Oriented Programming (过程性编程和面向对象编程)2. Abstraction and Classes (抽象和类)2.1. Classes in C (C 中的类)2.2. Implementing Class Member Functions (实现类成员函数)2.3. Using Classes References O…
阅读更多...
MySQL备份与还原

MySQL备份与还原

随着自动化办公与电子商务的不断发展&#xff0c;企业对于信息系统的依赖性越来越高&#xff0c;而数据库在信息系统中担任着非常重要的角色。尤其一些对数据可靠性要求非常高的行业,如银行证券、电信等&#xff0c;如果发生意外宕机或数据丢失&#xff0c;其损失是非常严重的。…
阅读更多...
风控图算法之社群发现算法（小数据集Python版）

风控图算法之社群发现算法（小数据集Python版）

风控图算法之社群发现算法&#xff08;小数据集Python版&#xff09; 在风险控制领域&#xff0c;图算法扮演着日益重要的角色。&#xff08;这方面的资料有很多&#xff0c;不再赘述&#xff09; 图算法在风控场景的应用 图分析方法在业务风控中的应用 特别是社群发现算法&a…
阅读更多...
最新文章

Copyright @ 2022~2023