我有一个 Spring 批处理作业的简单注释配置,如下所示:
@Configuration
@EnableBatchProcessing
public abstract class AbstractFileLoader<T> {
private static final String FILE_PATTERN = "*.dat";
@Bean
@StepScope
@Value("#{stepExecutionContext['fileName']}")
public FlatFileItemReader<T> reader(String file) {
FlatFileItemReader<T> reader = new FlatFileItemReader<T>();
String path = file.substring(file.indexOf(":") + 1, file.length());
FileSystemResource resource = new FileSystemResource(path);
reader.setResource(resource);
DefaultLineMapper<T> lineMapper = new DefaultLineMapper<T>();
lineMapper.setFieldSetMapper(getFieldSetMapper());
DelimitedLineTokenizer tokenizer = new DelimitedLineTokenizer(",");
tokenizer.setNames(getColumnNames());
lineMapper.setLineTokenizer(tokenizer);
reader.setLineMapper(lineMapper);
reader.setLinesToSkip(1);
return reader;
}
@Bean
public ItemProcessor<T, T> processor() {
// TODO add transformations here
return null;
}
//Exception when using JobScope for the writer
@Bean
public ItemWriter<T> writer() {
ListItemWriter<T> writer = new ListItemWriter<T>();
return writer;
}
@Bean
public Job loaderJob(JobBuilderFactory jobs, Step s1,
JobExecutionListener listener) {
return jobs.get(getLoaderName()).incrementer(new RunIdIncrementer())
.listener(listener).start(s1).build();
}
@Bean
public Step readStep(StepBuilderFactory stepBuilderFactory,
ItemReader<T> reader, ItemWriter<T> writer,
ItemProcessor<T, T> processor, TaskExecutor taskExecutor,
ResourcePatternResolver resolver) {
final Step readerStep = stepBuilderFactory
.get(getLoaderName() + " ReadStep:slave").<T, T> chunk(25254)
.reader(reader).processor(processor).writer(writer)
.taskExecutor(taskExecutor).throttleLimit(16).build();
final Step partitionedStep = stepBuilderFactory
.get(getLoaderName() + " ReadStep:master")
.partitioner(readerStep)
.partitioner(getLoaderName() + " ReadStep:slave",
partitioner(resolver)).taskExecutor(taskExecutor)
.build();
return partitionedStep;
}
@Bean
public TaskExecutor taskExecutor() {
return new SimpleAsyncTaskExecutor();
}
@Bean
public Partitioner partitioner(
ResourcePatternResolver resourcePatternResolver) {
MultiResourcePartitioner partitioner = new MultiResourcePartitioner();
Resource[] resources;
try {
resources = resourcePatternResolver.getResources("file:"
+ getFilesPath() + FILE_PATTERN);
} catch (IOException e) {
throw new RuntimeException(
"I/O problems when resolving the input file pattern.", e);
}
partitioner.setResources(resources);
return partitioner;
}
@Bean
public JobExecutionListener listener(ItemWriter<T> writer) {
/* org.springframework.batch.core.scope.StepScope scope; */
return new JobCompletionNotificationListener<T>(writer);
}
public abstract FieldSetMapper<T> getFieldSetMapper();
public abstract String getFilesPath();
public abstract String getLoaderName();
public abstract String[] getColumnNames();
}
当我使用两个不同的作业参数运行同一作业实例时,两个实例都按顺序运行而不是并行运行。我配置了一个 SimpleAysncTaskExecutor bean,我认为它应该会导致异步触发作业。
我是否需要向此类添加更多配置才能并行执行作业实例?
最佳答案
您必须配置用于启 Action 业的 jobLauncher
才能使用您的 TaskExecutor
(或单独的池)。最简单的方法是重写 bean:
@Bean
JobLauncher jobLauncher(JobRepository jobRepository) {
new SimpleJobLauncher(
taskExecutor: taskExecutor(),
jobRepository: jobRepository)
}
不要对将记录的警告感到困惑,该警告表示将使用同步任务执行器。这是由于 Spring Batch 使用非常尴尬的方式创建了一个额外的实例,它在 SimpleBatchConfiguration
中提供了它提供的 bean(长话短说,如果您想摆脱警告,那么您需要这样做)您需要提供一个 BatchConfigurer bean 并指定如何创建其他 4 个 bean,即使您只想更改其中一个)。
请注意,这里是否是同一工作无关紧要。问题是默认情况下作业启动器将在同一线程上启 Action 业。
关于java - Spring 批处理 : Job instances run sequentially when using annotaitons,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/35608765/