如何为春季批处理配置弹簧数据流

Question

我有春季批处理项目，我想在春季云数据流上配置它，我可以在SCDF上注册它以下是我的配置文件

@SpringBootApplication
@EnableBatchProcessing
@EnableTask
public class BatchApplication {
/*@Autowired
BatchCommandLineRunner batchcommdrunner;
@Bean
public CommandLineRunner commandLineRunner() {
    System.out.println("Executed at :" +  new SimpleDateFormat().format(new Date()));
    return batchcommdrunner ;
}*/
public static void main(String[] args) {
    SpringApplication.run(BatchApplication.class, args);
}
}

这是我的批处理结构文件

@Configuration
public class BatchConfiguaration {
@Autowired
private DataSource datasouce;
@Autowired
private JobBuilderFactory jobBuilderFactory;
@Autowired
private StepBuilderFactory stepBuilderFactory;
@Autowired
public Environment env;
@Bean(name = "reader")
@StepScope
public ItemReader<Schedules> reader(@Value("#{stepExecutionContext[scheduleRecs]}") List<Schedules> scherecs) {
    ItemReader<Schedules> reader = new IteratorItemReader<Schedules>(scherecs);
    return reader;
}
@Bean(name = "CWSreader")
@StepScope
public ItemReader<Contents> CWSreader(@Value("#{stepExecutionContext[scheduleRecs]}") List<Contents> scherecs) {
    ItemReader<Contents> reader = new IteratorItemReader<Contents>(scherecs);
    return reader;
}
@SuppressWarnings("rawtypes")
@Bean
@StepScope
public BatchProcessor processor() {
    return new BatchProcessor();
}
@Bean(name = "batchSchedulePreparedStatement")
@StepScope
public BatchSchedulePreparedStatement batchSchedulePreparedStatement() {
    return new BatchSchedulePreparedStatement();
}

@SuppressWarnings({ "rawtypes", "unchecked" })
@Bean(name = "batchWriter")
@StepScope
public BatchWriter batchWriter() {
    BatchWriter batchWriter = new BatchWriter();
    batchWriter.setDataSource(datasouce);
    batchWriter.setSql(env.getProperty("batch.insert.schedule.query"));
    batchWriter.setItemPreparedStatementSetter(batchSchedulePreparedStatement());
    return batchWriter;
}

@Bean("acheronDbTm")
@Qualifier("acheronDbTm")
public PlatformTransactionManager platformTransactionManager() {
    return new ResourcelessTransactionManager();
}
@Bean
public JobExplorer jobExplorer() throws Exception {
    MapJobExplorerFactoryBean explorerFactoryBean = new MapJobExplorerFactoryBean();
    explorerFactoryBean.setRepositoryFactory(mapJobRepositoryFactoryBean());
    explorerFactoryBean.afterPropertiesSet();
    return explorerFactoryBean.getObject();
}
@Bean
public MapJobRepositoryFactoryBean mapJobRepositoryFactoryBean() {
    MapJobRepositoryFactoryBean mapJobRepositoryFactoryBean = new MapJobRepositoryFactoryBean();
    mapJobRepositoryFactoryBean.setTransactionManager(platformTransactionManager());
    return mapJobRepositoryFactoryBean;
}
@Bean
public JobRepository jobRepository() throws Exception {
    return mapJobRepositoryFactoryBean().getObject();
}
@Bean
public SimpleJobLauncher jobLauncher() throws Exception {
    SimpleJobLauncher jobLauncher = new SimpleJobLauncher();
    jobLauncher.setJobRepository(jobRepository());
    return jobLauncher;
}
@Bean(name = "batchPartition")
@StepScope
public BatchPartition batchPartition() {
    BatchPartition batchPartition = new BatchPartition();
    return batchPartition;
}

@Bean(name="taskExecutor")
public TaskExecutor taskExecutor() {
    ThreadPoolTaskExecutor poolTaskExecutor = new ThreadPoolTaskExecutor();
    poolTaskExecutor.setCorePoolSize(10);
    poolTaskExecutor.setMaxPoolSize(30);
    poolTaskExecutor.setQueueCapacity(35);
    poolTaskExecutor.setThreadNamePrefix("Acheron");
    poolTaskExecutor.afterPropertiesSet();
    return poolTaskExecutor;
}
@Bean(name = "masterStep")
public Step masterStep() {
    return stepBuilderFactory.get("masterStep").partitioner(slave()).partitioner("slave", batchPartition())
            .taskExecutor(taskExecutor()).build();
}

@Bean(name = "slave")
public Step slave() {
    return stepBuilderFactory.get("slave").chunk(100).faultTolerant().retryLimit(2)
            .retry(DeadlockLoserDataAccessException.class).reader(reader(null)).processor(processor())
            .writer(batchWriter()).build();
}

@Bean(name = "manageStagingScheduleMaster")
public Job manageStagingScheduleMaster(final Step masterStep) throws Exception {
    return jobBuilderFactory.get("manageStagingScheduleMaster").preventRestart().incrementer(new RunIdIncrementer())
            .start(masterStep).build();
}

任何人都可以帮助我正确配置它，还是有其他方法可以监视批处理作业我还尝试了Spring Boot Admin，但它不支持SBA中的Java配置

我正在从控制器

中掌握这项工作

JobParametersBuilder builder = new JobParametersBuilder();
    System.out.println("Job Builder " + builder);
    JobParameters jobParameters = builder.toJobParameters();
    JobExecution execution = jobLauncher.run(job, jobParameters);
    return execution.getStatus().toString();

Answer 1

此示例显示了一个基本的春季批处理应用程序，该应用程序可以作为弹簧云数据流中的任务启动。