spring batch

Posted 2020-11-20

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