java 如何使用注释在 spring 批处理中运行多个作业

How to run multiple jobs in spring batch using annotations

提问by newstackoverflowuser5555

I am using Spring Boot + Spring Batch (annotation) , have come across a scenario where I have to run 2 jobs.

我正在使用 Spring Boot + Spring Batch（注释），遇到了我必须运行 2 个作业的场景。

I have Employee and Salary records which needs to updated using spring batch. I have configured BatchConigurationclasses by following this tutorial spring-batch getting started tutorialfor Employee and Salary objects, respectively named as BatchConfigurationEmployee & BatchConfigurationSalary.

我有 Employee 和 Salary 记录，需要使用 Spring Batch 进行更新。我已经BatchConiguration按照本教程spring-batch 入门教程为 Employee 和 Salary 对象配置了类，分别命名为 BatchConfigurationEmployee 和 BatchConfigurationSalary。

I have Defined the ItemReader, ItemProcessor, ItemWriterand Jobby following the tutorial which is mentioned above already.

我已经定义了ItemReader，ItemProcessor，ItemWriter并Job按照这上面已经提到的教程。

When I start my Spring Boot application either of the Job runs, I want to run both the BatchConfigured classes. How can I achieve this

当我启动 Spring Boot 应用程序时，其中一个 Job 运行，我想同时运行 BatchConfigured 类。我怎样才能做到这一点

********* BatchConfigurationEmployee.java *************

@Configuration
@EnableBatchProcessing
public class BatchConfigurationEmployee {
    public ItemReader<employee> reader() {
        return new EmployeeItemReader();
    }

    @Bean
    public ItemProcessor<Employee, Employee> processor() {
        return new EmployeeItemProcessor();
    }

    @Bean   
    public Job Employee(JobBuilderFactory jobs, Step s1) {
        return jobs.get("Employee")
                .incrementer(new RunIdIncrementer())
                .flow(s1)
                .end()
                .build();
    }

    @Bean
    public Step step1(StepBuilderFactory stepBuilderFactory, ItemReader<Employee> reader,
                    ItemProcessor<Employee, Employee> processor) {
        return stepBuilderFactory.get("step1")
                .<Employee, Employee> chunk(1)
                .reader(reader)
                .processor(processor)
                .build();
    }
}

Salary Class is here

薪资等级在这里

@Configuration
@EnableBatchProcessing
public class BatchConfigurationSalary {
    public ItemReader<Salary> reader() {
        return new SalaryItemReader();
    }

    @Bean
    public ItemProcessor<Salary, Salary> processor() {
        return new SalaryItemProcessor();
    }

    @Bean
    public Job salary(JobBuilderFactory jobs, Step s1) {
        return jobs.get("Salary")
                .incrementer(new RunIdIncrementer())
                .flow(s1)
                .end()
                .build();
    }

    @Bean
    public Step step1(StepBuilderFactory stepBuilderFactory, ItemReader<Salary> reader,
                    ItemProcessor<Salary, Salary> processor) {
        return stepBuilderFactory.get("step1")
                .<Salary, Salary> chunk(1)
                .reader(reader)
                .processor(processor)
                .build();
    }
}

回答by Hansjoerg Wingeier

The names of the Beans have to be unique in the whole Spring Context.

Bean 的名称在整个 Spring Context 中必须是唯一的。

In both jobs, you are instantiating the reader, writer and processor with the same methodname. The methodname is the name that is used to identifiy the bean in the context.

在这两个作业中，您都使用相同的方法名实例化读取器、写入器和处理器。methodname 是用于在上下文中标识 bean 的名称。

In both job-definitions, you have reader(), writer() and processor(). They will overwrite each other. Give them unique names like readerEmployee(), readerSalary() and so on.

在这两个作业定义中，您都有 reader()、writer() 和 processor()。它们会相互覆盖。给他们唯一的名字，比如 readerEmployee()、readerSalary() 等等。

That should solve your problem.

那应该可以解决您的问题。

回答by Hansjoerg Wingeier

You jobs are not annotated with @Bean, so the spring-context doesn't know them.

你的工作没有用@Bean 注释，所以 spring-context 不知道它们。

Have a look at the class JobLauncherCommandLineRunner. All Beans in the SpringContext implementing the Job interface will be injected. All jobs that are found will be executed. (this happens inside the method executeLocalJobs in JobLauncherCommandLineRunner)

查看 JobLauncherCommandLineRunner 类。将注入 SpringContext 中实现 Job 接口的所有 Bean。将执行找到的所有作业。（这发生在 JobLauncherCommandLineRunner 中的 executeLocalJobs 方法中）

If, for some reason, you don't want to have them as beans in the context, then you have to register your jobs with the jobregistry.( the method execute registeredJobs of JobLauncherCommandLineRunner will take care of launching the registered jobs)

如果出于某种原因，您不想将它们作为上下文中的 bean，那么您必须向 jobregistry 注册您的作业。（JobLauncherCommandLineRunner 的方法 executeregisteredJobs 将负责启动已注册的作业）

BTW, you can control with the property

顺便说一句，你可以用属性来控制

spring.batch.job.names= # Comma-separated list of job names to execute on startup (For instance
 `job1,job2`). By default, all Jobs found in the context are executed.

which jobs should be launched.

应该启动哪些作业。

回答by Joey587

I feel that this also is a pretty good way to run mutiple Jobs.

我觉得这也是一个很好的运行多个Jobs的方式。

I am making use of a Job Launcher to configure and execute the job and independent commandLineRunner implementation to run them. These are ordered to make sure they are executed sequentially in the required though

我正在使用 Job Launcher 来配置和执行作业，并使用独立的 commandLineRunner 实现来运行它们。这些命令是为了确保它们按要求顺序执行

Apologies for the big post but I wanted to give a clear picture of what can be achieved using JobLauncher configurations with multiple command line runners

为这篇大文章道歉，但我想清楚地说明使用带有多个命令行运行程序的 JobLauncher 配置可以实现的目标

This is the current BeanConfiguration that I have

这是我当前的 BeanConfiguration

@Configuration
public class BeanConfiguration {

    @Autowired
    DataSource dataSource;

    @Autowired
    PlatformTransactionManager transactionManager;

    @Bean(name="jobOperator")
     public JobOperator jobOperator(JobExplorer jobExplorer,

                                    JobRegistry jobRegistry) throws Exception {

            SimpleJobOperator jobOperator = new SimpleJobOperator();

            jobOperator.setJobExplorer(jobExplorer);
            jobOperator.setJobRepository(createJobRepository());
            jobOperator.setJobRegistry(jobRegistry);
            jobOperator.setJobLauncher(jobLauncher());

            return jobOperator;
     }

    /**
     * Configure joblaucnher to set the execution to be done asycn
     * Using the ThreadPoolTaskExecutor
     * @return
     * @throws Exception
     */
    @Bean
    public JobLauncher jobLauncher() throws Exception {
            SimpleJobLauncher jobLauncher = new SimpleJobLauncher();
            jobLauncher.setJobRepository(createJobRepository());
            jobLauncher.setTaskExecutor(taskExecutor());
            jobLauncher.afterPropertiesSet();
            return jobLauncher;
    }

    // Read the datasource and set in the job repo
    protected JobRepository createJobRepository() throws Exception {
        JobRepositoryFactoryBean factory = new JobRepositoryFactoryBean();
        factory.setDataSource(dataSource);
        factory.setTransactionManager(transactionManager);
        factory.setIsolationLevelForCreate("ISOLATION_SERIALIZABLE");
        //factory.setTablePrefix("BATCH_");
        factory.setMaxVarCharLength(10000);
        return factory.getObject();
    }

    @Bean
    public RestTemplateBuilder restTemplateBuilder() {
     return new RestTemplateBuilder().additionalInterceptors(new CustomRestTemplateLoggerInterceptor());
    }

    @Bean(name=AppConstants.JOB_DECIDER_BEAN_NAME_EMAIL_INIT)
    public JobExecutionDecider jobDecider() {
        return new EmailInitJobExecutionDecider();
    }

    @Bean
    public ThreadPoolTaskExecutor taskExecutor() {
    ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
    taskExecutor.setCorePoolSize(15);
    taskExecutor.setMaxPoolSize(20);
    taskExecutor.setQueueCapacity(30);
    return taskExecutor;
}
}

I have setup the database to hold the job exectuion details in postgre and hence the DatabaseConfiguration looks like this (two different beans for two different profiles -env)

我已经设置了数据库来保存 postgre 中的作业执行细节，因此 DatabaseConfiguration 看起来像这样（两个不同的 bean 用于两个不同的配置文件 -env）

@Configuration public class DatasourceConfiguration implements EnvironmentAware{

@Configuration 公共类 DatasourceConfiguration 实现 EnvironmentAware{

private Environment env;

@Bean
@Qualifier(AppConstants.DB_BEAN)
@Profile("dev")
public DataSource getDataSource() {
    HikariDataSource ds = new HikariDataSource();

    boolean isAutoCommitEnabled = env.getProperty("spring.datasource.hikari.auto-commit") != null ? Boolean.parseBoolean(env.getProperty("spring.datasource.hikari.auto-commit")):false;
    ds.setAutoCommit(isAutoCommitEnabled);
    // Connection test query is for legacy connections
    //ds.setConnectionInitSql(env.getProperty("spring.datasource.hikari.connection-test-query"));
    ds.setPoolName(env.getProperty("spring.datasource.hikari.pool-name"));
    ds.setDriverClassName(env.getProperty("spring.datasource.driver-class-name"));
    long timeout = env.getProperty("spring.datasource.hikari.idleTimeout") != null ? Long.parseLong(env.getProperty("spring.datasource.hikari.idleTimeout")): 40000;
    ds.setIdleTimeout(timeout);
    long maxLifeTime = env.getProperty("spring.datasource.hikari.maxLifetime") != null ? Long.parseLong(env.getProperty("spring.datasource.hikari.maxLifetime")): 1800000 ;
    ds.setMaxLifetime(maxLifeTime);
    ds.setJdbcUrl(env.getProperty("spring.datasource.url"));
    ds.setPoolName(env.getProperty("spring.datasource.hikari.pool-name"));
    ds.setUsername(env.getProperty("spring.datasource.username"));
    ds.setPassword(env.getProperty("spring.datasource.password"));
    int poolSize = env.getProperty("spring.datasource.hikari.maximum-pool-size") != null ? Integer.parseInt(env.getProperty("spring.datasource.hikari.maximum-pool-size")): 10;
    ds.setMaximumPoolSize(poolSize);

    return ds;
}

@Bean
@Qualifier(AppConstants.DB_PROD_BEAN)
@Profile("prod")

public DataSource getProdDatabase() {
    HikariDataSource ds = new HikariDataSource();

    boolean isAutoCommitEnabled = env.getProperty("spring.datasource.hikari.auto-commit") != null ? Boolean.parseBoolean(env.getProperty("spring.datasource.hikari.auto-commit")):false;
    ds.setAutoCommit(isAutoCommitEnabled);
    // Connection test query is for legacy connections
    //ds.setConnectionInitSql(env.getProperty("spring.datasource.hikari.connection-test-query"));
    ds.setPoolName(env.getProperty("spring.datasource.hikari.pool-name"));
    ds.setDriverClassName(env.getProperty("spring.datasource.driver-class-name"));
    long timeout = env.getProperty("spring.datasource.hikari.idleTimeout") != null ? Long.parseLong(env.getProperty("spring.datasource.hikari.idleTimeout")): 40000;
    ds.setIdleTimeout(timeout);
    long maxLifeTime = env.getProperty("spring.datasource.hikari.maxLifetime") != null ? Long.parseLong(env.getProperty("spring.datasource.hikari.maxLifetime")): 1800000 ;
    ds.setMaxLifetime(maxLifeTime);
    ds.setJdbcUrl(env.getProperty("spring.datasource.url"));
    ds.setPoolName(env.getProperty("spring.datasource.hikari.pool-name"));
    ds.setUsername(env.getProperty("spring.datasource.username"));
    ds.setPassword(env.getProperty("spring.datasource.password"));
    int poolSize = env.getProperty("spring.datasource.hikari.maximum-pool-size") != null ? Integer.parseInt(env.getProperty("spring.datasource.hikari.maximum-pool-size")): 10;
    ds.setMaximumPoolSize(poolSize);

    return ds;
}

public void setEnvironment(Environment environment) {
    // TODO Auto-generated method stub
    this.env = environment;
}

}

}

Make sure that the initial app launcher catches the app execution which will be returned once the job execution terminates (either gets failed or completed) so that you can gracefully shutdown the jvm. Else using joblauncher makes the jvm to be alive even after all jobs get completed

确保初始应用程序启动器捕获在作业执行终止（失败或完成）后将返回的应用程序执行，以便您可以正常关闭 jvm。否则，即使在所有作业完成后，使用 joblauncher 也会使 jvm 处于活动状态

@SpringBootApplication
@ComponentScan(basePackages="com.XXXX.Feedback_File_Processing.*")
@EnableBatchProcessing
public class FeedbackFileProcessingApp 
{
    public static void main(String[] args) throws Exception {
        ApplicationContext appContext = SpringApplication.run(FeedbackFileProcessingApp.class, args);
        // The batch job has finished by this point because the 
        //   ApplicationContext is not 'ready' until the job is finished
        // Also, use System.exit to force the Java process to finish with the exit code returned from the Spring App
        System.exit(SpringApplication.exit(appContext));
    }

}

............. so on , you can configure your own decider , your own job/steps as you said above for two different configurations like below and use them seperately in commandline runners (since the post is getting bigger, I am giving the details of just the job and command line runner)

………………等等，您可以像上面所说的那样为两种不同的配置配置自己的决策程序、自己的工作/步骤，并在命令行运行程序中单独使用它们（因为该帖子是越来越大，我只提供工作和命令行运行程序的详细信息）

These are the two jobs

这是两个工作

@Configuration
public class DefferalJobConfiguration {

    @Autowired
    JobLauncher joblauncher;

    @Autowired
    private JobBuilderFactory jobFactory;

    @Autowired
    private StepBuilderFactory stepFactory;

    @Bean
    @StepScope
    public Tasklet newSampleTasklet() {
        return ((stepExecution, chunkContext) -> {
            System.out.println("execution of step after flow");
            return RepeatStatus.FINISHED;
        });
    }

    @Bean
    public Step sampleStep() {
        return stepFactory.get("sampleStep").listener(new CustomStepExecutionListener())
                .tasklet(newSampleTasklet()).build();
    }

    @Autowired
    @Qualifier(AppConstants.FLOW_BEAN_NAME_EMAIL_INITIATION)
    private Flow emailInitFlow;

    @Autowired
    @Qualifier(AppConstants.JOB_DECIDER_BEAN_NAME_EMAIL_INIT)
    private JobExecutionDecider jobDecider;

    @Autowired
    @Qualifier(AppConstants.STEP_BEAN_NAME_ITEMREADER_FETCH_DEFERRAL_CONFIG)
    private Step deferralConfigStep;

    @Bean(name=AppConstants.JOB_BEAN_NAME_DEFERRAL)
    public Job deferralJob() {
        return jobFactory.get(AppConstants.JOB_NAME_DEFERRAL)
                .start(emailInitFlow)
                .on("COMPLETED").to(sampleStep())
                .next(jobDecider).on("COMPLETED").to(deferralConfigStep)
                .on("FAILED").fail()
                .end().build();


    }
}



@Configuration
public class TestFlowJobConfiguration {

    @Autowired
    private JobBuilderFactory jobFactory;

    @Autowired
    @Qualifier("testFlow")
    private Flow testFlow;

    @Bean(name = "testFlowJob")
    public Job testFlowJob() {

        return jobFactory.get("testFlowJob").start(testFlow).end().build();
    }
}

Here are the command line runners (I am making sure that the first job is completed before the second job is initialized but it is totally up to the user to execute them in parallel following a different stratergy)

这是命令行运行程序（我确保在第二个作业初始化之前完成第一个作业，但完全取决于用户按照不同的策略并行执行它们）

@Component
@Order(1)
public class DeferralCommandLineRunner implements CommandLineRunner, EnvironmentAware{
    // If the jobLauncher is not used, then by default jobs are launched using SimpleJobLauncher
    //  with default configuration(assumption)
    // hence modified the jobLauncher with vales set in BeanConfig
    // of spring batch
    private Environment env;

    @Autowired
    JobLauncher jobLauncher;

    @Autowired
    @Qualifier(AppConstants.JOB_BEAN_NAME_DEFERRAL)
    Job deferralJob;

    @Override
    public void run(String... args) throws Exception {
        // TODO Auto-generated method stub
        JobParameters jobparams = new JobParametersBuilder()
                .addString("run.time", LocalDateTime.now().
                        format(DateTimeFormatter.ofPattern(AppConstants.JOB_DATE_FORMATTER_PATTERN)).toString())
                .addString("instance.name", 
                        (deferralJob.getName() != null) ?deferralJob.getName()+'-'+UUID.randomUUID().toString() :
                            UUID.randomUUID().toString())
                .toJobParameters();
        jobLauncher.run(deferralJob, jobparams);
    }

    @Override
    public void setEnvironment(Environment environment) {
        // TODO Auto-generated method stub
        this.env = environment;
    }

}



@Component
@Order(2)
public class TestJobCommandLineRunner implements CommandLineRunner {

    @Autowired
    JobLauncher jobLauncher;

    @Autowired
    @Qualifier("testFlowJob")
    Job testjob;

    @Autowired
    @Qualifier("jobOperator")
    JobOperator operator;

    @Override
    public void run(String... args) throws Exception {
        // TODO Auto-generated method stub
        JobParameters jobParam = new JobParametersBuilder().addString("name", UUID.randomUUID().toString())
                .toJobParameters();
        System.out.println(operator.getJobNames());
        try {
            Set<Long> deferralExecutionIds = operator.getRunningExecutions(AppConstants.JOB_NAME_DEFERRAL);
            System.out.println("deferralExceutuibuds:" + deferralExecutionIds);

            operator.stop(deferralExecutionIds.iterator().next());

        } catch (NoSuchJobException | NoSuchJobExecutionException | JobExecutionNotRunningException e) {
            // just add a logging here
            System.out.println("exception caught:" + e.getMessage());
        }
        jobLauncher.run(testjob, jobParam);
    }

}

Hope this gives a complete idea of how it can be done. I am using spring-boot-starter-batch:jar:2.0.0.RELEASE

希望这能提供有关如何完成的完整想法。我正在使用 spring-boot-starter-batch:jar:2.0.0.RELEASE