行业资讯

Spring-batch学习总结（2）—Job,Flow创建及应用，多线程并发，决策器，监听器，参数

发布时间：2020-07-06 12:58:09 作者：墨营
来源：网络 阅读：12621

一．Job的创建及其应用
1.Job flow的介绍：
（1）状态机：例完成step1，是否继续完成step2,step3,我们就需要通过Job flow来控制
（2）进行演示:使用next()方法来达到顺序执行step1，step2...的目的，再使用on(),to(),from()方法达到与next()方法同样的目的，再展示fail()方法和stopAndRestart()方法；
例1：创建JobFlowDemoOne,以及三个Step使用next()让其顺序执行
JobFlowDemOne：

package com.dhcc.batch.batchDemo.config;

import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class JobFlowDemOne {
    @Autowired
    private JobBuilderFactory jobBuilderFactory;
    @Autowired
    private StepBuilderFactory stepBuilderFactory;

    @Bean
    public Job JobFlowDemo1() {
        return jobBuilderFactory.get("JobFlowDemo1").start(step1()).next(step2()).next(step3()).build();

    }

    @Bean
    public Step step1() {
        return stepBuilderFactory.get("step1").tasklet(new Tasklet() {

            @Override
            public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
                System.out.println("step1-->Hello Spring Batch....");
                return RepeatStatus.FINISHED;
            }
        }).build();

    }

    @Bean
    public Step step2() {
        return stepBuilderFactory.get("step2").tasklet((contribution, context) -> {
            System.out.println("step 2-->Hello Spring Batch..");
            return RepeatStatus.FINISHED;
        }).build();

    }

    @Bean
    public Step step3() {
        return stepBuilderFactory.get("step3").tasklet(new Tasklet() {

            @Override
            public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
                System.out.println("step3-->Hello Spring Batch....");
                return RepeatStatus.FINISHED;
            }
        }).build();

    }
}

在step2()中，我们使用了java8的新特性，使用了Lambda表达式

package com.dhcc.batch.batchDemo;

import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
@EnableBatchProcessing
public class BatchDemoApplication {

    public static void main(String[] args) {
        SpringApplication.run(BatchDemoApplication.class, args);
    }
}

开始执行，执行结果为：
Spring-batch学习总结（2）—Job,Flow创建及应用，多线程并发，决策器，监听器，参数
观察控制台执行结果，我们可以发现Step1,step2,step3,按照next()的顺序顺序执行完毕，此时我们在进入数据库中进行观察，我们主要观察JobExecution表以及StepExecution表，查看其表数据：
JobExecution:
Spring-batch学习总结（2）—Job,Flow创建及应用，多线程并发，决策器，监听器，参数
StepExecution:
Spring-batch学习总结（2）—Job,Flow创建及应用，多线程并发，决策器，监听器，参数
在查看一个JobInstance,观察Job实例：
Spring-batch学习总结（2）—Job,Flow创建及应用，多线程并发，决策器，监听器，参数
例2：我们使用on(),to(),from()方法顺序执行我们的Step,达到与例1相同的效果
原代码不变，我们将JobFlowDemo1略作修改（注：我在这里为例方便演示新建了一个数据库）：

@Bean
    public Job JobFlowDemo2() {
        return jobBuilderFactory.get("JobFlowDemo2")
//              .start(step1())
//              .next(step2())
//              .next(step3())
                .start(step1()).on("COMPLETED").to(step2())
                .from(step2()).on("COMPLETED").to(step3())
                .from(step3()).end()
                .build();

观察控制台：
Spring-batch学习总结（2）—Job,Flow创建及应用，多线程并发，决策器，监听器，参数
代码执行成功；
fail()方法和stopAndRestart()方法后面用到的时候我们在做详细了解

二．Flow的创建及使用
1.flow的介绍：
（1）flow是一个Step的集合，他规定了Step与Step之间的转换关系;
（2）创建Flow可以达到复用的效果，让其在不同的Job之间进行复用;
（3）使用FlowBuilder去创建一个Flow,他和Job类似，使用start(),next()以及end()来运行flow;
例：
创建JobFlowDemoTwoApplication 

package com.dhcc.batch.batchDemo.jobFlowDemoTwo;

import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
@EnableBatchProcessing
public class JobFlowDemoTwoApplication {

    public static void main(String[] args) {
        SpringApplication.run(JobFlowDemoTwoApplication.class, args);
    }
}

创建step,flow以及Job——》JobFlowDemoTwoConfiguration：

package com.dhcc.batch.batchDemo.jobFlowDemoTwo;

import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.job.builder.FlowBuilder;
import org.springframework.batch.core.job.flow.Flow;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class JobFlowDemoTwoConfiguration {
    @Autowired
    private JobBuilderFactory jobBuilderFactory;

    @Autowired
    private StepBuilderFactory stepBuilderFactory;

    @Bean
    public Step jobFlowDemoTwo1() {
        return stepBuilderFactory.get("jobFlowDemoTwo1").tasklet(new Tasklet() {

            @Override
            public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
                System.out.println("Hello-->jobFlowDemoTwo1");
                return RepeatStatus.FINISHED;
            }
        }).build();

    }

    @Bean
    public Step jobFlowDemoTwo2() {
        return stepBuilderFactory.get("jobFlowDemoTwo2").tasklet(new Tasklet() {

            @Override
            public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
                System.out.println("Hello-->jobFlowDemoTwo2");
                return RepeatStatus.FINISHED;
            }
        }).build();

    }

    @Bean
    public Step jobFlowDemoTwo3() {
        return stepBuilderFactory.get("jobFlowDemoTwo3").tasklet(new Tasklet() {

            @Override
            public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
                System.out.println("Hello-->jobFlowDemoTwo3");
                return RepeatStatus.FINISHED;
            }
        }).build();

    }

    // 创建Flow,它是一个Step集合
    @Bean
    public Flow jobFlowDemoFlow() {
        return new FlowBuilder<Flow>("jobFlowDemoFlow")
                .start(jobFlowDemoTwo1())
                .next(jobFlowDemoTwo2())
                .build();

    }

    //创建一个Job执行Flow以及Step
    @Bean
    public Job jobFlowDemoTwoJob() {
        return  jobBuilderFactory.get("jobFlowDemoTwoJob")
                .start(jobFlowDemoFlow())
                .next(jobFlowDemoTwo3())
                .end()
                .build();

    }

}

开始运行，我们观察控制台：
Spring-batch学习总结（2）—Job,Flow创建及应用，多线程并发，决策器，监听器，参数
运行结果我们可以看见Job先运行了Flow中的两个Step,然后再运行jobFlowDemoStep3(),与我们设定的顺序一样
在数据库中，我们以stepExection表为例观察：
Spring-batch学习总结（2）—Job,Flow创建及应用，多线程并发，决策器，监听器，参数
看见了3个Step顺序执行完成；
三．Spilt多线程并发任务定义及使用
1.Spilt异步执行Flow
2.举例说明：
（1）首先定义两个Flow,在每个Flow中定义一些Step,每一个Step将自身的名字以及当前运行的线程打印出来；
（2）创建一个Job使用Spilt异步的启动两个Flow；
（3）运行Job,查看结果；
例：
JobSpiltDemoApplication：

package com.dhcc.batch.batchDemo.jobSpiltDemo;

import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
@EnableBatchProcessing
public class JobSpiltDemoApplication {

    public static void main(String[] args) {
        SpringApplication.run(JobSpiltDemoApplication.class, args);
    }
}

JobSpiltDemoConfiguration：

package com.dhcc.batch.batchDemo.jobSpiltDemo;

import org.springframework.batch.core.Job;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.job.builder.FlowBuilder;
import org.springframework.batch.core.job.flow.Flow;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.repeat.RepeatStatus;
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;

@Configuration
public class JobSpiltDemoConfiguration {

    @Autowired
    private JobBuilderFactory jobBuilderFactory;
    @Autowired
    private StepBuilderFactory stepBuilderFactory;

    /**
     * 创建job运行Flow,我们利用split(new
     * SimpleAsyncTaskExecutor()).add()让flow异步执行,add()中可以添加多个Flow
     * 
     * @return
     */
    @Bean
    public Job SpiltJob() {
        return jobBuilderFactory.get("SpiltJob").start(jobSpiltFlow1()).split(new SimpleAsyncTaskExecutor())
                .add(jobSpiltFlow2()).end().build();

    }

    // 创建Flow1
    @Bean
    public Flow jobSpiltFlow1() {
        return new FlowBuilder<Flow>("jobSpiltFlow1")
                .start(stepBuilderFactory.get("jobSpiltStep1").tasklet(tasklet()).build())
                .next(stepBuilderFactory.get("jobSpiltStep2").tasklet(tasklet()).build()).build();

    }

    // 创建Flow1
    @Bean
    public Flow jobSpiltFlow2() {
        return new FlowBuilder<Flow>("jobSpiltFlow2")
                .start(stepBuilderFactory.get("jobSpiltStep3").tasklet(tasklet()).build())
                .next(stepBuilderFactory.get("jobSpiltStep4").tasklet(tasklet()).build()).build();

    }

    private Tasklet tasklet() {
        return new PrintTasklet();
    }

    // step执行的任务类（可以写为外部类，此处为了方便，我们写为内部类）
    private class PrintTasklet implements Tasklet {

        @Override
        public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
            System.out.println("has been execute on stepName:" + chunkContext.getStepContext().getStepName()
                    + ",has been execute on thread:" + Thread.currentThread().getName());
            return RepeatStatus.FINISHED;
        }

    }

}

运行结果：

Spring-batch学习总结（2）—Job,Flow创建及应用，多线程并发，决策器，监听器，参数
观察控制台，我们可以看出我们达到了让Step异步执行的目的
观察数据库表：
Spring-batch学习总结（2）—Job,Flow创建及应用，多线程并发，决策器，监听器，参数

四．决策器的定义及应用
1.Decision：为我们提供下一步执行哪一个Step提供条件决策
2.JobExecutionDecider:接口，提供决策条件
例：

package com.dhcc.batch.batchDemo.flowDecisionDemo;

import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
@EnableBatchProcessing
public class FlowDecisionDemoApplication {

    public static void main(String[] args) {
        SpringApplication.run(FlowDecisionDemoApplication.class, args);
    }
}


FlowDecisionDemoConfiguration：

package com.dhcc.batch.batchDemo.flowDecisionDemo;

import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.job.flow.JobExecutionDecider;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class FlowDecisionDemoConfiguration {
    @Autowired
    private JobBuilderFactory jobBuilderFactory;
    @Autowired
    private StepBuilderFactory stepBuilderFactory;

    //编写Job
    @Bean
    public Job FlowDecisionJob() {
        return jobBuilderFactory.get("FlowDecisionJob")
                .start(firstStep()).next(myDecider())
                .from(myDecider()).on("EVEN").to(evenStep())
                .from(myDecider()).on("ODD").to(oddStep())
                .from(oddStep()).on("*").to(myDecider())
                .end()
                .build();

    }
    @Bean
    public Step firstStep() {
        return stepBuilderFactory.get("firstStep").tasklet(new Tasklet() {

            @Override
            public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
                System.out.println("Hello firstStep..");
                return RepeatStatus.FINISHED;
            }
        }).build();

    }

    @Bean
    public Step evenStep() {
        return stepBuilderFactory.get("evenStep").tasklet(new Tasklet() {

            @Override
            public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
                System.out.println("Hello evenStep..");
                return RepeatStatus.FINISHED;
            }
        }).build();

    }

    @Bean
    public Step oddStep() {
        return stepBuilderFactory.get("oddStep").tasklet(new Tasklet() {

            @Override
            public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
                System.out.println("Hello oddStep..");
                return RepeatStatus.FINISHED;
            }
        }).build();

    }

    //编写决策器
    @Bean
    public JobExecutionDecider myDecider() {
        return new myDecider();

    }

}

myDecider：

package com.dhcc.batch.batchDemo.flowDecisionDemo;

import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.job.flow.FlowExecutionStatus;
import org.springframework.batch.core.job.flow.JobExecutionDecider;

public class myDecider implements JobExecutionDecider {
    private int count=0;

    @Override
    public FlowExecutionStatus decide(JobExecution jobExecution, StepExecution stepExecution) {
        count++;
        if(count%2==0) {
            return new FlowExecutionStatus("EVEN");
        }else {
            return new FlowExecutionStatus("ODD");
        }
    }

}

运行结果：
Spring-batch学习总结（2）—Job,Flow创建及应用，多线程并发，决策器，监听器，参数
我们观察控制台，分析结果：首先job中先运行firstStep(),然后进入到myDecider中Count++,此时count=1，返回”ODD”,job中执行oddStep(),然后无论什么状态再次进入myDecider中，此时count=2,故返回”EVEN”，下一步执行evenStep();
五．Job的嵌套定义及应用
1.job可以嵌套使用，嵌套的Job我们将其称为子job,被嵌套的Job我们将其称为父job;
2.一个父Job可以有多个子Job;
3.子job不能单独运行，需要其父Job去启动;
例：
NestedJobApplication:

package com.dhcc.batch.batchDemo.nestedJob;

import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
@EnableBatchProcessing
public class NestedJobApplication {

    public static void main(String[] args) {
        SpringApplication.run(NestedJobApplication.class, args);
    }
}

接下来建立两个子Job，每个子job中有两个Step,一个父job
让父job控制运行子job

ChildJobOneConfiguration:

package com.dhcc.batch.batchDemo.nestedJob;

import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class ChildJobOneConfiguration {
    @Autowired
    private JobBuilderFactory jobBuilderFactory;

    @Autowired
    private StepBuilderFactory stepBuilderFactory;

    @Bean
    public Job childJob1() {
        return jobBuilderFactory.get("childJob1")
                .start(childJob1Step1())
                .next(childJob1Step2())
                .build();

    }

    @Bean
    public Step childJob1Step1() {
        return stepBuilderFactory.get("childJob1Step1").tasklet(new Tasklet() {

            @Override
            public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
                System.out.println("Hello---->childJob1Step1");
                return RepeatStatus.FINISHED;
            }
        }).build();

    }

    @Bean
    public Step childJob1Step2() {
        return stepBuilderFactory.get("childJob1Step2").tasklet(new Tasklet() {

            @Override
            public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
                System.out.println("Hello---->childJob1Step2");
                return RepeatStatus.FINISHED;
            }
        }).build();

    }

}

ChildJobTwoConfiguration:

package com.dhcc.batch.batchDemo.nestedJob;

import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class ChildJobTwoConfiguration {
    @Autowired
    private JobBuilderFactory jobBuilderFactory;
    @Autowired
    private StepBuilderFactory stepBuilderFactory;

    @Bean
    public Job childJob2() {
        return jobBuilderFactory.get("childJob2")
                .start(childJob2Step1())
                .next(childJob2Step2())
                .build();

    }

    @Bean
    public Step childJob2Step1() {
        return stepBuilderFactory.get("childJob2Step1").tasklet(new Tasklet() {

            @Override
            public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
                System.out.println("Hello---->childJob2Step1");
                return RepeatStatus.FINISHED;
            }
        }).build();

    }

    @Bean
    public Step childJob2Step2() {
        return stepBuilderFactory.get("childJob2Step2").tasklet(new Tasklet() {

            @Override
            public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
                System.out.println("Hello---->childJob2Step2");
                return RepeatStatus.FINISHED;
            }
        }).build();

    }

}

ParentJobConfiguration:

package com.dhcc.batch.batchDemo.nestedJob;

import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.batch.core.step.builder.JobStepBuilder;
import org.springframework.batch.core.step.builder.StepBuilder;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.transaction.PlatformTransactionManager;

@Configuration
public class ParentJobConfiguration {
    @Autowired
    private JobBuilderFactory jobBuilderFactory;
    @Autowired
    private StepBuilderFactory stepBuilderFactory;
    @Autowired
    private Job childJob1;
    @Autowired
    private Job childJob2;
    @Autowired
    private JobLauncher jobLauncher;

    @Bean
    public Job parentJob(JobRepository repository,PlatformTransactionManager transactionManager) {
        return jobBuilderFactory.get("parentJob")
                .start(parentJobStep())
                .next(childJob1(repository, transactionManager))
                .next(childJob2(repository, transactionManager))
                .build();
    }

    private Step childJob1(JobRepository repository,PlatformTransactionManager transactionManager) {
        return new JobStepBuilder(new StepBuilder("childJob1"))
                .job(childJob1)
                .launcher(jobLauncher)
                .repository(repository)
                .transactionManager(transactionManager)
                .build();
    }

    private Step childJob2(JobRepository repository,PlatformTransactionManager transactionManager) {
        return new JobStepBuilder(new StepBuilder("childJob2"))
                .job(childJob2)
                .launcher(jobLauncher)
                .repository(repository)
                .transactionManager(transactionManager)
                .build();
    }

    @Bean
    public Step parentJobStep() {
        return stepBuilderFactory.get("parentJobStep")
                .tasklet(new Tasklet() {

                    @Override
                    public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
                        System.out.println("Helllo------>parentJobStep..");
                        return RepeatStatus.FINISHED;
                    }
                }).build();

    }

}

在配置文件中添加

spring.datasource.url=jdbc:mysql://localhost:3306/springbatch?useUnicode=true&characterEncoding=UTF-8&serverTimezone=GMT%2B8&useSSL=false
spring.datasource.username=root
spring.datasource.password=qitao1996
spring.datasource.driver-class-name=com.mysql.jdbc.Driver
spring.datasource.schema=classpath:/org/springframework/batch/core/schema-mysql.sql

spring.batch.initialize-schema=always

spring.batch.job.names=parentJob

运行结果：

Spring-batch学习总结（2）—Job,Flow创建及应用，多线程并发，决策器，监听器，参数

观察控制台，成功运行，我们达到了job嵌套的效果
六．监听器的定义及应用
1.Listener:控制Job执行的一种方式
2.可以通过接口或者注解实现监听器
3.在spring-batch中提供各个级别的监听器接口，从job级别到item级别都有
（1）JobExecutionListener(before..,after..);
（2）StepExecutionListener(before..,after..);
（3）ChunkListener(before..,after..);
（4）ItemReaderListener;ItemWriterListener;ItemProcessListener(before..,after..,error..);
例：
ListenerJobApplication :

package com.dhcc.batch.batchDemo.listenerJob;

import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
@EnableBatchProcessing
public class ListenerJobApplication {

    public static void main(String[] args) {
        SpringApplication.run(ListenerJobApplication.class, args);
    }
}

创建监听器（通过实现接口实现）
MyJobListener：

package com.dhcc.batch.batchDemo.listenerJob;

import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobExecutionListener;

/**
 * 实现接口构建监听器
 * @author Administrator
 *
 */
public class MyJobListener implements JobExecutionListener{

    @Override
    public void beforeJob(JobExecution jobExecution) {
        System.out.println(jobExecution.getJobInstance().getJobName()+"before running......");
    }

    @Override
    public void afterJob(JobExecution jobExecution) {
        System.out.println(jobExecution.getJobInstance().getJobName()+"before running......");
    }

}

创建监听器（通过注解实现）
MyChunkListener：

package com.dhcc.batch.batchDemo.listenerJob;

import org.springframework.batch.core.annotation.AfterChunk;
import org.springframework.batch.core.annotation.BeforeChunk;
import org.springframework.batch.core.scope.context.ChunkContext;
/**
 * 使用注解构建监听器
 * @author Administrator
 *
 */
public class MyChunkListener {

    @BeforeChunk
    public void beforeChunk(ChunkContext context) {
        System.out.println(context.getStepContext().getStepName()+"chunk before running.....");
    }

    @AfterChunk
    public void afterChunk(ChunkContext context) {
        System.out.println(context.getStepContext().getStepName()+"chunk after running.....");
    }

}

MyListenerJobConfiguration：

package com.dhcc.batch.batchDemo.listenerJob;

import java.util.Arrays;
import java.util.List;

import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.support.ListItemReader;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class MyListenerJobConfiguration {
    @Autowired
    private JobBuilderFactory jobBuilderFactory;
    @Autowired
    private StepBuilderFactory stepBuilderFactory;

    //监听Job执行
    @Bean
    public Job myListenerJob() {
        return jobBuilderFactory.get("myListenerJob")
                .start(myListenerStep())
                .listener(new MyJobListener())  
                .build();

    }

    private Step myListenerStep() {
        return stepBuilderFactory.get("myListenerStep")
                .<String,String>chunk(2)
                .faultTolerant()
                .listener(new MyChunkListener())
                .reader(reader())
                .writer(writer())
                .build();
    }
    private ItemReader<? extends String> reader() {
        return new ListItemReader<>(Arrays.asList("maozedong","zhude","pendehuai","zhouenlai","liushaoqi"));
    }

    private ItemWriter<? super String> writer() {
        return new ItemWriter<String>() {

            @Override
            public void write(List<? extends String> items) throws Exception {
                for(String item:items) {
                    System.out.println("Writing item: "+item);
                }

            }
        };
    }

}

运行结果：
Spring-batch学习总结（2）—Job,Flow创建及应用，多线程并发，决策器，监听器，参数
观察控制台，我们成功的完成了监听任务..
七．Job参数初探
1.JobParameters作用：在Job运行过程中，可以用来传递信息
2.通过”key---->value”键值对的形式传入，在代码中我们通过get(“key”)来获取value值
例：
JobParametersDemoApplication：

package com.dhcc.batch.batchDemo.jobParametersDemo;

import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
@EnableBatchProcessing
public class JobParametersDemoApplication {

    public static void main(String[] args) {
        SpringApplication.run(JobParametersDemoApplication.class, args);
    }
}

创建Job,Step:
JobParametersConfiguretion:

package com.dhcc.batch.batchDemo.jobParametersDemo;

import java.util.Map;

import org.springframework.batch.core.ExitStatus;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobParameter;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.StepExecutionListener;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class JobParametersConfiguretion implements StepExecutionListener{

    @Autowired
    private JobBuilderFactory jobBuilderFactory;
    @Autowired
    private StepBuilderFactory stepBuilderFactory;
    private Map<String, JobParameter> parames;

    @Bean
    public Job MyParametersJobThree() {
        return jobBuilderFactory.get("MyParametersJobThree")
                .start(MyParametersJobStep3())
                .build();   
    }

    private Step MyParametersJobStep3() {

        return stepBuilderFactory.get("MyParametersJobStep3")
                .listener(this)
                .tasklet(new Tasklet() {

                    @Override
                    public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
                        System.out.println("parame is: "+parames.get("info"));
                        return RepeatStatus.FINISHED;
                    }
                }).build();
    }

    @Override
    public void beforeStep(StepExecution stepExecution) {
        System.out.println(stepExecution.getStepName()+"运行之前...........");
        parames = stepExecution.getJobParameters().getParameters();
    }

    @Override
    public ExitStatus afterStep(StepExecution stepExecution) {
        System.out.println(stepExecution.getStepName()+"运行之完毕...........");
        return null;
    }
}

运行结果：
Spring-batch学习总结（2）—Job,Flow创建及应用，多线程并发，决策器，监听器，参数
观察控制台，此处我们没有进行传参;

推荐阅读：
  1. Veeam创建复制任务Replication Job
  2. JOB_QUEUE_PROCESSES 参数

免责声明：本站发布的内容（图片、视频和文字）以原创、转载和分享为主，文章观点不代表本网站立场，如果涉及侵权请联系站长邮箱：is@yisu.com进行举报，并提供相关证据，一经查实，将立刻删除涉嫌侵权内容。

job flow 创建

上一篇：EXT文件系统特性介绍

下一篇：给php加拓展包

相关阅读

您好，登录后才能下订单哦！

密码登录
登录注册
获取短信验证码
其他方式登录
点击 登录注册 即表示同意《亿速云用户服务条款》
产品服务
地区划分
帮助支持
关于我们
行业资讯-文章归档 问答-问答归档
广州亿速云计算有限公司

7*24小时在线电话：400-100-2938

7*24小时在线QQ：800811969

Copyright © Yisu Cloud Ltd. All Rights Reserved. 2018 版权所有
粤ICP备17096448号-1 粤公网安备 44010402001142号