【Spring云原生】Spring Batch：海量数据高并发任务处理！数据处理纵享新丝滑！事务管理机制+并行处理+实例应用讲解

2024-03-04 16:36
文章标签 java 数据 云原生 讲解 应用 并发 spring 处理 实例 海量 任务 事务 数据处理 batch 纵享 管理机制 并行处理 新丝滑

本文主要是介绍【Spring云原生】Spring Batch：海量数据高并发任务处理！数据处理纵享新丝滑！事务管理机制+并行处理+实例应用讲解，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

 🎉🎉欢迎光临🎉🎉

🏅我是苏泽，一位对技术充满热情的探索者和分享者。🚀🚀

🌟特别推荐给大家我的最新专栏《Spring 狂野之旅：从入门到入魔》 🚀

本专栏带你从Spring入门到入魔！

这是苏泽的个人主页可以看到我其他的内容哦👇👇

努力的苏泽icon-default.png?t=N7T8http://suzee.blog.csdn.net/

本文重点讲解原理！如要看批量数据处理的实战请关注下文（后续补充敬请关注）：

实例应用：数据清洗和转换

使用Spring Batch清洗和转换数据

实例应用：数据导入和导出

使用Spring Batch导入和导出数据

实例应用：批处理定时任务

使用Spring Batch实现定时任务

目录

实例应用：数据清洗和转换

使用Spring Batch清洗和转换数据

实例应用：数据导入和导出

使用Spring Batch导入和导出数据

实例应用：批处理定时任务

使用Spring Batch实现定时任务

介绍Spring Batch

Spring Batch入门

解析

需求缔造：假设我们有一个需求，需要从一个CSV文件中读取学生信息，对每个学生的成绩进行转换和校验，并将处理后的学生信息写入到一个数据库表中。

数据处理

扩展Spring Batch

自定义读取器、写入器和处理器

 与其他Spring项目的集成

与Spring Integration的集成：

与Spring Cloud Task的集成：

介绍Spring Batch

Spring Batch是一个基于Java的开源批处理框架，用于处理大规模、重复性和高可靠性的任务。它提供了一种简单而强大的方式来处理批处理作业，如数据导入/导出、报表生成、批量处理等。

什么是Spring Batch？

Spring Batch旨在简化批处理作业的开发和管理。它提供了一种可扩展的模型来定义和执行批处理作业，将作业划分为多个步骤（Step），每个步骤又由一个或多个任务块（Chunk）组成。通过使用Spring Batch，可以轻松处理大量的数据和复杂的业务逻辑。

Spring Batch的特点和优势

  1. 可扩展性和可重用性：Spring Batch采用模块化的设计，提供了丰富的可扩展性和可重用性。可以根据具体需求自定义作业流程，添加或删除步骤，灵活地适应不同的批处理场景。

  2. 事务管理：Spring Batch提供了强大的事务管理机制，确保批处理作业的数据一致性和完整性。可以配置事务边界，使每个步骤或任务块在单独的事务中执行，保证了作业的可靠性。

  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 {@Autowiredprivate JobBuilderFactory jobBuilderFactory;@Autowiredprivate StepBuilderFactory stepBuilderFactory;@Beanpublic Step step1() {return stepBuilderFactory.get("step1").tasklet((contribution, chunkContext) -> {System.out.println("Hello, Spring Batch!");return RepeatStatus.FINISHED;}).build();}@Beanpublic Job job(Step step1) {return jobBuilderFactory.get("job").start(step1).build();}
}

解析

首先使用@Configuration和@EnableBatchProcessing注解将类标记为Spring Batch的配置类。然后，使用JobBuilderFactory和StepBuilderFactory创建作业和步骤的构建器。在step1方法中，定义了一个简单的任务块，打印"Hello, Spring Batch!"并返回RepeatStatus.FINISHED。最后，在job方法中，使用jobBuilderFactory创建一个作业，并将step1作为作业的起始步骤。

3. 理解Job、Step和任务块

  • Job（作业）：作业是一个独立的批处理任务，由一个或多个步骤组成。它描述了整个批处理过程的流程和顺序，并可以有自己的参数和配置。

  • Step（步骤块）：步骤是作业的组成部分，用于执行特定的任务。一个作业可以包含一个或多个步骤，每个步骤都可以定义自己的任务和处理逻辑。

  • 任务块（Chunk）：任务块是步骤的最小执行单元，用于处理一定量的数据。任务块将数据分为一块一块进行处理，可以定义读取数据、处理数据和写入数据的逻辑。

需求缔造：
假设我们有一个需求，需要从一个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 Batch提供了数据转换和校验的机制。可以使用ItemProcessor对读取的数据进行转换、过滤和校验。ItemProcessor可以应用自定义的业务逻辑来处理每个数据项。

      我们配置了一个FlatFileItemReader，设置了CSV文件的位置和行映射器，指定了字段分隔符和字段到模型属性的映射关系。

    接下来，我们可以定义一个ItemProcessor来对读取的学生信息进行转换和校验：

    @Bean
public ItemProcessor<Student, Student> studentItemProcessor() {return new ItemProcessor<Student, Student>() {@Overridepublic 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 Batch提供了作业调度器来配置和管理批处理作业的执行。可以使用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 Batch提供了丰富的监控和管理功能。可以使用Spring Batch的管理接口和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 Batch对不同数据源和数据格式的支持，以及如何配置和组装作业步骤来完成整个批处理任务。

错误处理和重试机制

  • Spring Batch提供了错误处理和重试机制，以确保批处理作业的稳定性和可靠性。可以配置策略来处理读取、处理和写入过程中的错误和异常情况。可以设置重试次数、重试间隔和错误处理策略，以适应不同的错误场景和需求。
    首先，我们可以在步骤配置中设置错误处理策略。例如，我们可以使用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>() {@Overridepublic void onSkipInRead(Throwable throwable) {// 处理读取过程中发生的异常}@Overridepublic void onSkipInWrite(Student student, Throwable throwable) {// 处理写入过程中发生的异常}@Overridepublic void onSkipInProcess(Student student, Throwable throwable) {// 处理处理过程中发生的异常}};
}@Bean
public RetryListener studentRetryListener() {return new RetryListener() {@Overridepublic <T, E extends Throwable> boolean open(RetryContext retryContext, RetryCallback<T, E> retryCallback) {// 在重试之前执行的逻辑return true;}@Overridepublic <T, E extends Throwable> void onError(RetryContext retryContext, RetryCallback<T, E> retryCallback, Throwable throwable) {// 处理重试过程中发生的异常}@Overridepublic <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();
}

批处理最佳实践

  • 数据量控制：在批处理作业中，应注意控制数据量的大小，以避免内存溢出或处理速度过慢的问题。可以通过分块（Chunk）处理和分页读取的方式来控制数据量。

  • 事务管理：在批处理作业中，对于需要保证数据一致性和完整性的操作，应使用适当的事务管理机制。可以配置事务边界，确保每个步骤或任务块在独立的事务中执行。

  • 错误处理和日志记录：合理处理错误和异常情况是批处理作业的重要部分。应使用适当的错误处理策略、日志记录和报警机制，以便及时发现和处理问题。

  • 性能调优：在批处理作业中，应关注性能调优的问题。可以通过合理的并行处理、合理配置的线程池和适当的数据读取和写入策略来提高作业的处理速度和效率。

  • 监控和管理：对于长时间运行的批处理作业，应设置适当的监控和管理机制。可以使用监控工具、警报系统和自动化任务管理工具来监控作业的执行情况和性能指标。

扩展Spring Batch

自定义读取器、写入器和处理器

Spring Batch提供了许多扩展点，可以通过自定义读取器、写入器和处理器以及其他组件来扩展和定制批处理作业的功能。

public class MyItemReader implements ItemReader<String> {private List<String> data = Arrays.asList("item1", "item2", "item3");private Iterator<String> iterator = data.iterator();@Overridepublic String read() throws Exception {if (iterator.hasNext()) {return iterator.next();} else {return null;}}
}

自定义写入器：

public class MyItemWriter implements ItemWriter<String> {@Overridepublic void write(List<? extends String> items) throws Exception {for (String item : items) {// 自定义写入逻辑}}
}

自定义处理器：

public class MyItemProcessor implements ItemProcessor<String, String> {@Overridepublic String process(String item) throws Exception {// 自定义处理逻辑return item.toUpperCase();}
}

批处理作业的并行处理:

Spring Batch支持将批处理作业划分为多个独立的步骤，并通过多线程或分布式处理来实现并行处理。

  1. 多线程处理：可以通过配置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()方法中。

  2. 分布式处理：如果需要更高的并行性和可伸缩性，可以考虑使用分布式处理。Spring Batch提供了与Spring Integration和Spring Cloud Task等项目的集成，以实现分布式部署和处理。

 与其他Spring项目的集成

  1. 与Spring Integration的集成：

首先，需要在Spring Batch作业中配置Spring Integration的消息通道和适配器。可以使用消息通道来发送和接收作业的输入和输出数据，使用适配器来与外部系统进行交互。

@Configuration
@EnableBatchProcessing
@EnableIntegration
public class BatchConfiguration {@Autowiredprivate JobBuilderFactory jobBuilderFactory;@Autowiredprivate StepBuilderFactory stepBuilderFactory;@Autowiredprivate MyItemReader reader;@Autowiredprivate MyItemProcessor processor;@Autowiredprivate MyItemWriter writer;@Beanpublic IntegrationFlow myJobFlow() {return IntegrationFlows.from("jobInputChannel").handle(jobLaunchingGateway()).get();}@Beanpublic MessageChannel jobInputChannel() {return new DirectChannel();}@Beanpublic MessageChannel jobOutputChannel() {return new DirectChannel();}@Beanpublic MessageChannel stepInputChannel() {return new DirectChannel();}@Beanpublic MessageChannel stepOutputChannel() {return new DirectChannel();}@Beanpublic JobLaunchingGateway jobLaunchingGateway() {SimpleJobLauncher jobLauncher = new SimpleJobLauncher();jobLauncher.setJobRepository(jobRepository());return new JobLaunchingGateway(jobLauncher);}@Beanpublic JobRepository jobRepository() {// 配置作业存储库}@Beanpublic Job myJob() {return jobBuilderFactory.get("myJob").start(step1()).build();}@Beanpublic 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 Task的集成：

首先，需要在Spring Batch作业中配置Spring Cloud Task的任务启动器和任务监听器。任务启动器用于启动和管理分布式任务，任务监听器用于在任务执行期间执行一些操作。

@Configuration
@EnableBatchProcessing
@EnableTask
public class BatchConfiguration {@Autowiredprivate JobBuilderFactory jobBuilderFactory;@Autowiredprivate StepBuilderFactory stepBuilderFactory;@Autowiredprivate MyItemReader reader;@Autowiredprivate MyItemProcessor processor;@Autowiredprivate MyItemWriter writer;@Beanpublic TaskConfigurer taskConfigurer() {return new DefaultTaskConfigurer();}@Beanpublic TaskExecutor taskExecutor() {return new SimpleAsyncTaskExecutor();}@Beanpublic Job myJob() {return jobBuilderFactory.get("myJob").start(step1()).build();}@Beanpublic Step step1() {return stepBuilderFactory.get("step1").<String, String>chunk(10).reader(reader).processor(processor).writer(writer).taskExecutor(taskExecutor()).build();}@Beanpublic TaskListener myTaskListener() {return new MyTaskListener();}@Beanpublic TaskExecutionListener myTaskExecutionListener() {return new MyTaskExecutionListener();}
}

这篇关于【Spring云原生】Spring Batch：海量数据高并发任务处理！数据处理纵享新丝滑！事务管理机制+并行处理+实例应用讲解的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/773781。

相关文章

如何使用celery进行异步处理和定时任务(django)

如何使用celery进行异步处理和定时任务(django)

《如何使用celery进行异步处理和定时任务(django)》文章介绍了Celery的基本概念、安装方法、如何使用Celery进行异步任务处理以及如何设置定时任务,通过Celery,可以在Web应用中... 目录一、celery的作用二、安装celery三、使用celery 异步执行任务四、使用celery
阅读更多...
在Ubuntu上部署SpringBoot应用的操作步骤

在Ubuntu上部署SpringBoot应用的操作步骤

《在Ubuntu上部署SpringBoot应用的操作步骤》随着云计算和容器化技术的普及,Linux服务器已成为部署Web应用程序的主流平台之一,Java作为一种跨平台的编程语言,具有广泛的应用场景,本... 目录一、部署准备二、安装 Java 环境1. 安装 JDK2. 验证 Java 安装三、安装 mys
阅读更多...
Springboot的ThreadPoolTaskScheduler线程池轻松搞定15分钟不操作自动取消订单

Springboot的ThreadPoolTaskScheduler线程池轻松搞定15分钟不操作自动取消订单

《Springboot的ThreadPoolTaskScheduler线程池轻松搞定15分钟不操作自动取消订单》：本文主要介绍Springboot的ThreadPoolTaskScheduler线... 目录ThreadPoolTaskScheduler线程池实现15分钟不操作自动取消订单概要1，创建订单后
阅读更多...
详谈redis跟数据库的数据同步问题

详谈redis跟数据库的数据同步问题

《详谈redis跟数据库的数据同步问题》文章讨论了在Redis和数据库数据一致性问题上的解决方案,主要比较了先更新Redis缓存再更新数据库和先更新数据库再更新Redis缓存两种方案,文章指出,删除R... 目录一、Redis 数据库数据一致性的解决方案1.1、更新Redis缓存、删除Redis缓存的区别二
阅读更多...
JAVA中整型数组、字符串数组、整型数和字符串 的创建与转换的方法

JAVA中整型数组、字符串数组、整型数和字符串 的创建与转换的方法

《JAVA中整型数组、字符串数组、整型数和字符串的创建与转换的方法》本文介绍了Java中字符串、字符数组和整型数组的创建方法,以及它们之间的转换方法,还详细讲解了字符串中的一些常用方法,如index... 目录一、字符串、字符数组和整型数组的创建1、字符串的创建方法1.1 通过引用字符数组来创建字符串1.2
阅读更多...
Redis事务与数据持久化方式

Redis事务与数据持久化方式

《Redis事务与数据持久化方式》该文档主要介绍了Redis事务和持久化机制,事务通过将多个命令打包执行,而持久化则通过快照（RDB）和追加式文件（AOF）两种方式将内存数据保存到磁盘,以防止数据丢失... 目录一、Redis 事务1.1 事务本质1.2 数据库事务与redis事务1.2.1 数据库事务1.
阅读更多...
Python中构建终端应用界面利器Blessed模块的使用

Python中构建终端应用界面利器Blessed模块的使用

《Python中构建终端应用界面利器Blessed模块的使用》Blessed库作为一个轻量级且功能强大的解决方案,开始在开发者中赢得口碑,今天,我们就一起来探索一下它是如何让终端UI开发变得轻松而高... 目录一、安装与配置：简单、快速、无障碍二、基本功能：从彩色文本到动态交互1. 显示基本内容2. 创建链
阅读更多...
SpringCloud集成AlloyDB的示例代码

SpringCloud集成AlloyDB的示例代码

《SpringCloud集成AlloyDB的示例代码》AlloyDB是GoogleCloud提供的一种高度可扩展、强性能的关系型数据库服务,它兼容PostgreSQL,并提供了更快的查询性能... 目录1.AlloyDBjavascript是什么？AlloyDB 的工作原理2.搭建测试环境3.代码工程1.
阅读更多...
Java调用Python代码的几种方法小结

Java调用Python代码的几种方法小结

《Java调用Python代码的几种方法小结》Python语言有丰富的系统管理、数据处理、统计类软件包,因此从java应用中调用Python代码的需求很常见、实用,本文介绍几种方法从java调用Pyt... 目录引言Java core使用ProcessBuilder使用Java脚本引擎总结引言python
阅读更多...
SpringBoot操作spark处理hdfs文件的操作方法

SpringBoot操作spark处理hdfs文件的操作方法

《SpringBoot操作spark处理hdfs文件的操作方法》本文介绍了如何使用SpringBoot操作Spark处理HDFS文件,包括导入依赖、配置Spark信息、编写Controller和Ser... 目录SpringBoot操作spark处理hdfs文件1、导入依赖2、配置spark信息3、cont
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2