spring batch

Posted

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了spring batch相关的知识,希望对你有一定的参考价值。

A lightweight, comprehensive batch framework designed to enable the development of robust batch applications vital for the daily operations of enterprise systems.(一款轻量的、全面的批处理框架,用于开发强大的日常运营的企业级批处理应用程序。)

框架主要功能:

Transaction management(事务管理)

Chunk based processing(基于块的处理)

Declarative I/O(声明式的输入输出)

Start/Stop/Restart(启动/停止/再启动)

Retry/Skip(重试/跳过)

技术分享图片技术分享图片

框架一共有4个主要角色:

JobLauncher是任务启动器,通过它来启动任务,可以看做是程序的入口。

Job代表着一个具体的任务。

Step代表着一个具体的步骤,一个Job可以包含多个Step。

JobRepository是存储数据的地方,可以看做是一个数据库的接口,在任务执行的时候需要通过它来记录任务状态等等信息。

JobLauncher

JobLauncher是任务启动器,该接口只有一个run方法:

public interface JobLauncher {

    public JobExecution run(Job job, JobParameters jobParameters) throws Exception;

}

除了传入Job对象之外,还需要传入JobParameters对象。通过JobLauncher可以在Java程序中调用批处理任务,也可以通过命令行或者其他框架(如定时调度框架Quartz、Web后台框架Spring MVC)中调用批处理任务。Spring Batch框架提供了一个JobLauncher的实现类SimpleJobLauncher。

Job

Job代表着一个任务,一个Job与一个或者多个JobInstance相关联,而一个JobInstance又与一个或者多个JobExecution相关联:

技术分享图片技术分享图片

考虑到任务可能不是只执行一次就再也不执行了,更多的情况可能是定时任务,如每天执行一次,每个星期执行一次等等,那么为了区分每次执行的任务,框架使用了JobInstance。

如上图所示,Job是一个EndOfDay(每天最后时刻执行的任务),其中一个JobInstance就代表着2007年5月5日那天执行的任务实例。框架通过在执行JobLauncher.run(Job, JobParameters)方法时传入的JobParameters来区分是哪一天的任务。由于2007年5月5日那天执行的任务可能不会一次就执行完成,比如中途被停止,或者出现异常导致中断,需要多执行几次才能完成,所以框架使用了JobExecution来表示每次执行的任务。

Step

一个Job任务可以分为几个Step步骤,与JobExection相同,每次执行Step的时候使用StepExecution来表示执行的步骤。每一个Step还包含着一个ItemReader、ItemProcessor、ItemWriter:

ItemReader

public interface ItemReader<T> {

    T read() throws Exception;

}

ItemReader代表着读操作,框架已经提供了多种ItemReader接口的实现类,包括对文本文件、XML文件、数据库、JMS消息等读的处理,当然我们也可以自己实现该接口

ItemProcessor

public interface ItemProcessor<I, O> {

    O process(I item) throws Exception;

}

ItemReader代表着处理操作,process方法的形参传入I类型的对象,通过处理后返回O型的对象。开发者可以实现自己的业务代码来对数据进行处理

ItemWriter

public interface ItemWriter<T> {

    void write(List<? extends T> items) throws Exception;

}

ItemReader代表着写操作,框架已经提供了多种ItemWriter接口的实现类,包括对文本文件、XML文件、数据库、JMS消息等写的处理,当然我们也可以自己实现该接口

JobRepository

JobRepository用于存储任务执行的状态信息,比如什么时间点执行了什么任务、任务执行结果如何等等。框架提供了2种实现,一种是通过Map形式保存在内存中,当Java程序重启后任务信息也就丢失了,并且在分布式下无法获取其他节点的任务执行情况;另一种是保存在数据库中

例子:

引入依赖

<dependency>    <groupId>org.springframework.batch</groupId>    <artifactId>spring-batch-core</artifactId>    <version>3.0.8.RELEASE</version> </dependency>

装载bean

<beans xmlns="http://www.springframework.org/schema/beans"

       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

       xsi:schemaLocation="

http://www.springframework.org/schema/beans

http://www.springframework.org/schema/beans/spring-beans.xsd">

    <!-- 事务管理器 -->

    <bean id="transactionManager" class="org.springframework.batch.support.transaction.ResourcelessTransactionManager"/>

    <!-- 任务仓库 -->

    <bean id="jobRepository" class="org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean">

        <property name="transactionManager" ref="transactionManager"/>

    </bean>

    <!-- 任务加载器 -->

    <bean id="jobLauncher" class="org.springframework.batch.core.launch.support.SimpleJobLauncher">

        <property name="jobRepository" ref="jobRepository"/>

    </bean>

</beans>

创建Reader

我们直接在resources目录下创建一个batch-data.csv文件,内容如下:

1,PENDING

2,PENDING

3,PENDING

4,PENDING

5,PENDING

6,PENDING

7,PENDING

8,PENDING

9,PENDING

10,PENDING

读操作需要实现ItemReader<T>接口,框架提供了一个现成的实现类FlatFileItemReader。使用该类需要设置Resource和LineMapper。Resource代表着数据源,即我们的batch-data.csv文件;LineMapper则表示如何将文件的每行数据转成对应的DTO对象。

创建DTO对象

public class DeviceCommand {

    private String id;

    private String status;

}

自定义LineMapper

我们需要自己实现一个LineMapper实现类,用于将batch-data.csv文件的每行数据,转成程序方便处理的DeviceCommand对象。

public class HelloLineMapper implements LineMapper<DeviceCommand> {

    @Override

    public DeviceCommand mapLine(String line, int lineNumber) throws Exception {

        // 逗号分割每一行数据

        String[] args = line.split(",");

        // 创建DeviceCommand对象

        DeviceCommand deviceCommand = new DeviceCommand();

        // 设置id值到对象中

        deviceCommand.setId(args[0]);

        // 设置status值到对象中

        deviceCommand.setStatus(args[1]);

        // 返回对象

        return deviceCommand;

    }

}

创建Processor

读完数据后,我们就需要处理数据了。处理操作需要实现ItemProcessor<I, O>接口,我们自己实现一个HelloItemProcessor.java即可,代码如下:

public class HelloItemProcessor implements ItemProcessor<DeviceCommand, DeviceCommand> {

    @Override

    public DeviceCommand process(DeviceCommand deviceCommand) throws Exception {

        // 模拟下发命令给设备

        System.out.println("send command to device, id=" + deviceCommand.getId());

        // 更新命令状态

        deviceCommand.setStatus("SENT");

        // 返回命令对象

        return deviceCommand;     

    }     

}

创建Writer

处理完数据后,我们需要更新命令状态到文件里,用于记录我们已经下发。与读文件类似,我们需要实现ItemWriter<T>接口,框架也提供了一个现成的实现类FlatFileItemWriter。使用该类需要设置Resource和LineAggregator。Resource代表着数据源,即我们的batch-data.csv文件;LineAggregator则表示如何将DTO对象转成字符串保存到文件的每行。

自定义LineAggregator

我们需要自己实现一个LineAggregator实现类,用于将DeviceCommand对象转成字符串,保存到batch-data.csv文件。

public class HelloLineAggregator implements LineAggregator<DeviceCommand> {

    @Override

    public String aggregate(DeviceCommand deviceCommand) {

        StringBuffer sb = new StringBuffer();

        sb.append(deviceCommand.getId());

        sb.append(",");

        sb.append(deviceCommand.getStatus());

        return sb.toString();

    }

}

主程序

技术分享图片

public class Main {

    public static void main(String[] args) throws Exception {

        // 加载上下文

        String[] configLocations = {"applicationContext.xml"};

        ApplicationContext applicationContext = new ClassPathXmlApplicationContext(configLocations);

        // 获取任务启动器,任务仓库,事物管理器

        JobLauncher jobLauncher = applicationContext.getBean(JobLauncher.class);

        JobRepository jobRepository = applicationContext.getBean(JobRepository.class);

        PlatformTransactionManager transactionManager = applicationContext.getBean(PlatformTransactionManager.class);

        // 创建reader

        FlatFileItemReader<DeviceCommand> flatFileItemReader = new FlatFileItemReader<>();

        flatFileItemReader.setResource(new FileSystemResource("src/main/resources/batch-data.csv"));

        flatFileItemReader.setLineMapper(new HelloLineMapper());

        // 创建processor

        HelloItemProcessor helloItemProcessor = new HelloItemProcessor();

        // 创建writer

        FlatFileItemWriter<DeviceCommand> flatFileItemWriter = new FlatFileItemWriter<>();

        flatFileItemWriter.setResource(new FileSystemResource("src/main/resources/batch-data.csv"));

        flatFileItemWriter.setLineAggregator(new HelloLineAggregator());

        // 创建Step

        StepBuilderFactory stepBuilderFactory = new StepBuilderFactory(jobRepository, transactionManager);

        Step step = stepBuilderFactory.get("step")

                   .<DeviceCommand, DeviceCommand>chunk(1)

                   .reader(flatFileItemReader)       // 读操作

                   .processor(helloItemProcessor)    // 处理操作

                   .writer(flatFileItemWriter)       // 写操作

                   .build();

        // 创建Job

        JobBuilderFactory jobBuilderFactory = new JobBuilderFactory(jobRepository);

        Job job = jobBuilderFactory.get("job")

                                   .start(step)

                                   .build();

        // 启动任务

        jobLauncher.run(job, new JobParameters());

    }

}

执行main方法之后,屏幕将会输出下面信息:

send command to device, id=1

send command to device, id=2

send command to device, id=3

send command to device, id=4

send command to device, id=5

send command to device, id=6

send command to device, id=7

send command to device, id=8

send command to device, id=9

send command to device, id=10

再查看batch-data.csv文件,将会发现命令状态全部更新为SENT:

1,SENT

2,SENT

3,SENT

4,SENT

5,SENT

6,SENT

7,SENT

8,SENT

9,SENT

10,SENT

引申----实现批处理

1.得到要批处理的数据

dataList-->Process.getNeedIds(Parm)

2.生产者(准备数据)

ProducerExecutor<T, Long> ProducerExecutor{

List<T> producer(List<Long> dataList){

Process.getNeedUpdate(dataList, Parm)执行的操作

}

}

3.消费者(准备要执行的数据)

CustomerExecutor<T> customerExecutor{

void execute(T data){

Process.updateOrAdd(data, Parm)执行的操作

}

}

4.process(定义数据的执行逻辑)

5.批量执行任务

生产者:

data = ProducerExecutor.producer(dataList.subList(e, 分批大小toIndex))

//执行完的数据写入LinkedBlockingQueue queue中

queue.put(data)

消费者:

ExecutorService executorService = Executors.newFixedThreadPool(consumeThreadPoolSize);

//从queue中取出数据执行

data = queue.poll()

executorService.submit

customerExecutor.execute(data)

6.执行完毕后

Process.after(Parm)

关于ExecutorService接口:

它扩展自Executor接口,Executor接口仅有一个方法:execute(runnable)

ExecutorService在Executor的基础上增加了“service”特性的方法:

shutdown()、shutdownNow():都是关闭当前service服务,释放Executor的所有资源;它所触发的动作就是取消队列中任务的执行

shutdown是一种“友好”的关闭,它将不再接受新的任务提交,同时把已经提交到队列中的任务执行完毕。shutdownNow更加直接一些,它将会把尚未执行的任务不再执行。shutdowNow是个有返回类型的方法,它返回那些等待执行的任务列表(List<Runnable>)

Future submit(callable/runnale):向Executor提交任务,并返回一个结果未定的Future。

通过 Executor来启动线程,比用Thread的start()更好:可以很容易控制线程的启动、执行和关闭过程,还可以很容易使用线程池的特性。

1.创建ExecutorService

通过工具类java.util.concurrent.Executors的静态方法创建。Executors包中所定义了 Executor、ExecutorService、ScheduledExecutorService、ThreadFactory 和 Callable 类的工厂和实用方法。

比如创建一个ExecutorService的实例,ExecutorService实际上是一个线程池的管理工具:

ExecutorService executorService = Executors.newCachedThreadPool();

ExecutorService executorService = Executors.newFixedThreadPool(3);

ExecutorService executorService = Executors.newSingleThreadExecutor();

2.将任务添加到线程去执行

当将一个任务添加到线程池中的时候,线程池会为每个任务创建一个线程,该线程会在之后的某个时刻自动执行。

3.关闭执行服务对象

executorService.shutdown();

4.获取任务的执行的返回值

任务分两类:一类是实现Runnable接口的类,一类是实现了Callable接口的类。两者都可以被 ExecutorService执行,但是Runnable任务没有返回值,而Callable任务有返回值。

并且Callable的call()方法只能通过ExecutorService的(<T> task) 方法来执行,并且返回一个 <T>,<T>是表示任务等待完成的 Future。

例子:

ExecutorService executorService = Executors.newCachedThreadPool();

//创建10个任务并执行

for (int i = 0; i < 10; i++) {

//使用ExecutorService执行Callable类型的任务,并将结果保存在future中

Future<String> future = executorService.submit(new TaskWithResult(i));

System.out.println(fs.get()); //打印各个线程(任务)执行的结果

//启动一次顺序关闭,执行以前提交的任务,但不接受新任务。

如果已经关闭,则调用没有其他作用。

executorService.shutdown();

}

定义任务:

class TaskWithResult implements Callable<String> {

privateint id;

public TaskWithResult(int id) {

this.id = id;

}

public String call() throws Exception {

System.out.println("call()方法被自动调用));

//一个模拟耗时的操作

for (int i = 999999; i > 0; i--) ;

return"call()方法的返回结果,保存在future";

}

}



部分转载 http://www.importnew.com/26177.html

以上是关于spring batch的主要内容,如果未能解决你的问题,请参考以下文章

你了解Spring从Spring3到Spring5的变迁吗?

Spring全家桶笔记:Spring+Spring Boot+Spring Cloud+Spring MVC

学习笔记——Spring简介;Spring搭建步骤;Spring的特性;Spring中getBean三种方式;Spring中的标签

Spring--Spring入门

Spring框架--Spring事务管理和Spring事务传播行为

Spring框架--Spring事务管理和Spring事务传播行为