springbatch 配置并行启动job详解 (八) -凯发k8官方网

文章目录

• • 一、创建并行job

前言：在日常业务中可能需要job并行执行，springbatch支持job并行步执行，并且配置简单。

代码已上传github上面地址：https://github.com/fadehub/spring-boot-learn/tree/master/spring-boot-springbatch

springbatch其它文章直通车：

• springbatch读单个文件(flatfileitemreader)和写单个文件(flatfileitemwriter)(一)
• springbatch顺序读取多文件(multiresourceitemreader)和顺序写文件(multiresourceitemwriter)(二)
• springbatch读数据库(mybatispagingitemreader)(三)
• springbatch读文件(flatfileitemreader)写据库(mybatisbatchitemwriter)(四)
• springbatch 监听器之job监听器(jobexecutionlistener)和step监听器(stepexecutionlistener)(五)
• springbatch 监听器之chunk监听器(chunklistener)和skip监听器(skiplistener)(六)
• springbatch 多线程(taskexecutor)启动job详解 (七)

一、创建并行job

首先使用flowbuilder构建一个个小的flow流程，在这个流程里面指定步骤，两个流程flow是并行执行的，下面有两个并行流flow1和flow2，flow1里面有step1 step2先后顺序，flow2有step3，也就是说{step1,step2}一起和step3是并行的：

package com.sl.config;import com.sl.common.commonfileitemreader; import com.sl.common.commonfileitemwriter; import com.sl.common.commonmybatisitemreader; import com.sl.entity.cafecat; import com.sl.entity.cat; import com.sl.entity.people; import com.sl.entity.student; import com.sl.listener.catchunklistener; import com.sl.listener.catjoblistener; import com.sl.listener.catsteplistener; import com.sl.processor.cafecatprocessor; import com.sl.processor.studentprocessor; import org.apache.ibatis.session.sqlsessionfactory; 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.batch.core.configuration.annotation.stepscope; import org.springframework.batch.core.job.builder.flowbuilder; import org.springframework.batch.core.job.flow.flow; import org.springframework.batch.core.job.flow.support.simpleflow; import org.springframework.beans.factory.annotation.autowired; import org.springframework.context.annotation.bean; import org.springframework.context.annotation.configuration; import org.springframework.core.task.simpleasynctaskexecutor; import org.springframework.core.task.taskexecutor;/*** job并行执行* @author shuliangzhao* @title: catflowconfiguration* @projectname spring-boot-learn* @description: todo* @date 2019/9/14 20:10*/ @configuration @enablebatchprocessing public class catflowconfiguration {@autowiredprivate jobbuilderfactory jobbuilderfactory;@autowiredprivate stepbuilderfactory stepbuilderfactory;@autowiredprivate cafecatprocessor cafecatprocessor;@autowiredprivate sqlsessionfactory sqlsessionfactory;@autowiredprivate catjoblistener catjoblistener;@autowiredprivate catsteplistener catsteplistener;@autowiredprivate catchunklistener catchunklistener;@autowiredprivate studentprocessor studentprocessor;@beanpublic job catflowjob() {return jobbuilderfactory.get("catflowjob").start(splitflow()).next(catflowstep()).build().build();}@beanpublic step catflowstep() {return stepbuilderfactory.get("catflowstep")// .listener(catsteplistener).listener(catchunklistener).chunk(10).reader(catflowcommonmybatisitemreader()).processor(cafecatprocessor).writer(cafecatflowcommonfileitemwriter()).taskexecutor(flowtaskexecutor()).throttlelimit(8).build();}@bean@stepscopepublic commonmybatisitemreader catflowcommonmybatisitemreader() {return new commonmybatisitemreader(sqlsessionfactory,cat.class.getsimplename());}@bean@stepscopepublic commonfileitemwriter cafecatflowcommonfileitemwriter() {return new commonfileitemwriter<>(cafecat.class);}@beanpublic flow splitflow() {return new flowbuilder("splitflow").split(flowtaskexecutor()).add(flow1(), flow2()).build();}@beanpublic flow flow1() {return new flowbuilder("flow1").start(step1()).next(step2()).build();}@beanpublic flow flow2() {return new flowbuilder("flow2").start(step3()).build();}@beanpublic step step3() {return stepbuilderfactory.get("step3").chunk(10).reader(step3flow2commonfileitemreader()).processor(studentprocessor).writer(step3flow2commonfileitemwriter()).build();}@bean@stepscopepublic commonfileitemreader step3flow2commonfileitemreader() {return new commonfileitemreader<>(people.class);}@bean@stepscopepublic commonfileitemwriter step3flow2commonfileitemwriter() {return new commonfileitemwriter<>(student.class);}@beanpublic step step1() {return stepbuilderfactory.get("step1").chunk(10).reader(step1flow1commonfileitemreader()).processor(studentprocessor).writer(step1flow1commonfileitemwriter()).build();}@bean@stepscopepublic commonfileitemreader step1flow1commonfileitemreader() {return new commonfileitemreader<>(people.class);}@bean@stepscopepublic commonfileitemwriter step1flow1commonfileitemwriter() {return new commonfileitemwriter<>(student.class);}@beanpublic step step2() {return stepbuilderfactory.get("step2").chunk(10).reader(step2flow1commonfileitemreader()).processor(studentprocessor).writer(step2flow1commonfileitemwriter()).build();}@bean@stepscopepublic commonfileitemreader step2flow1commonfileitemreader() {return new commonfileitemreader<>(people.class);}@bean@stepscopepublic commonfileitemwriter step2flow1commonfileitemwriter() {return new commonfileitemwriter<>(student.class);}@beanpublic taskexecutor flowtaskexecutor(){return new simpleasynctaskexecutor("spring_batch");} }

需要指定taskexecutor 应该使用哪个实现来执行各个流。默认值为 synctaskexecutor没有用，需要异步taskexecutor才能并行运行这些步骤。

总结

以上是凯发k8官方网为你收集整理的springbatch 配置并行启动job详解 (八)的全部内容，希望文章能够帮你解决所遇到的问题。

如果觉得凯发k8官方网网站内容还不错，欢迎将凯发k8官方网推荐给好友。