Background
我正在设计一个文件读取层,可以读取分隔文件并将其加载到List
中。我决定使用 Spring Batch,因为它提供了许多可扩展性选项,我可以根据它们的大小将其用于不同的文件集。
要求
- 我想设计一个可用于读取任何分隔文件的通用作业 API。
- 应该有一个作业结构,应该用于解析每个分隔文件。例如,如果系统需要读取 5 个文件,则将有 5 个作业(每个文件一个)。这 5 个作业彼此不同的唯一方法是它们将使用不同的
FieldSetMapper
、列名、目录路径和其他缩放参数,例如commit-interval
和throttle-limit
。 - 此 API 的用户应该不需要配置 Spring当系统中引入新的文件类型时,他自己进行批处理作业、步骤、分块、分区等。
- 用户需要做的就是提供作业要使用的
FieldsetMapper
以及commit-interval
、throttle-limit
和放置每种类型文件的目录。 - 每个文件将有一个预定义的目录。每个目录可以包含相同类型和格式的多个文件。
MultiResourcePartioner
将用于查看目录内部。分区数 = 目录中的文件数。
我的要求是构建一个Spring Batch基础设施,为我提供一个独特的工作,一旦我有了构成这项工作的零碎内容,我就可以启动。
我的解决方案:
我创建了一个抽象的配置类,该类将由具体的配置类扩展(每个文件将有 1 个要读取的具体类)。
@Configuration
@EnableBatchProcessing
public abstract class AbstractFileLoader<T> {
private static final String FILE_PATTERN = "*.dat";
@Autowired
JobBuilderFactory jobs;
@Autowired
ResourcePatternResolver resourcePatternResolver;
public final Job createJob(Step s1, JobExecutionListener listener) {
return jobs.get(this.getClass().getSimpleName())
.incrementer(new RunIdIncrementer()).listener(listener)
.start(s1).build();
}
public abstract Job loaderJob(Step s1, JobExecutionListener listener);
public abstract FieldSetMapper<T> getFieldSetMapper();
public abstract String getFilesPath();
public abstract String[] getColumnNames();
public abstract int getChunkSize();
public abstract int getThrottleLimit();
@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;
}
@Bean
@JobScope
public ListItemWriter<T> writer() {
ListItemWriter<T> writer = new ListItemWriter<T>();
return writer;
}
@Bean
@JobScope
public Step readStep(StepBuilderFactory stepBuilderFactory,
ItemReader<T> reader, ItemWriter<T> writer,
ItemProcessor<T, T> processor, TaskExecutor taskExecutor) {
final Step readerStep = stepBuilderFactory
.get(this.getClass().getSimpleName() + " ReadStep:slave")
.<T, T> chunk(getChunkSize()).reader(reader)
.processor(processor).writer(writer).taskExecutor(taskExecutor)
.throttleLimit(getThrottleLimit()).build();
final Step partitionedStep = stepBuilderFactory
.get(this.getClass().getSimpleName() + " ReadStep:master")
.partitioner(readerStep)
.partitioner(
this.getClass().getSimpleName() + " ReadStep:slave",
partitioner()).taskExecutor(taskExecutor).build();
return partitionedStep;
}
/*
* @Bean public TaskExecutor taskExecutor() { return new
* SimpleAsyncTaskExecutor(); }
*/
@Bean
@JobScope
public Partitioner partitioner() {
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
@JobScope
public JobExecutionListener listener(ListItemWriter<T> writer) {
return new JobCompletionNotificationListener<T>(writer);
}
/*
* Use this if you want the writer to have job scope (JIRA BATCH-2269). Also
* change the return type of writer to ListItemWriter for this to work.
*/
@Bean
public TaskExecutor taskExecutor() {
return new SimpleAsyncTaskExecutor() {
@Override
protected void doExecute(final Runnable task) {
// gets the jobExecution of the configuration thread
final JobExecution jobExecution = JobSynchronizationManager
.getContext().getJobExecution();
super.doExecute(new Runnable() {
public void run() {
JobSynchronizationManager.register(jobExecution);
try {
task.run();
} finally {
JobSynchronizationManager.close();
}
}
});
}
};
}
}
假设为了讨论,我必须阅读发票数据。因此,我可以扩展上述类来创建InvoiceLoader
:
@Configuration
public class InvoiceLoader extends AbstractFileLoader<Invoice>{
private class InvoiceFieldSetMapper implements FieldSetMapper<Invoice> {
public Invoice mapFieldSet(FieldSet f) {
Invoice invoice = new Invoice();
invoice.setNo(f.readString("INVOICE_NO");
return e;
}
}
@Override
public FieldSetMapper<Invoice> getFieldSetMapper() {
return new InvoiceFieldSetMapper();
}
@Override
public String getFilesPath() {
return "I:/CK/invoices/partitions/";
}
@Override
public String[] getColumnNames() {
return new String[] { "INVOICE_NO", "DATE"};
}
@Override
@Bean(name="invoiceJob")
public Job loaderJob(Step s1,
JobExecutionListener listener) {
return createJob(s1, listener);
}
@Override
public int getChunkSize() {
return 25254;
}
@Override
public int getThrottleLimit() {
return 8;
}
}
假设我还有一个名为 Inventory
的类,它扩展了AbstractFileLoader.
在应用程序启动时,我可以加载这两个注释配置,如下所示:
AbstractApplicationContext context1 = new AnnotationConfigApplicationContext(InvoiceLoader.class, InventoryLoader.class);
在我的应用程序中的其他地方,两个不同的线程可以启动作业,如下所示:
线程 1 :
JobLauncher jobLauncher1 = context1.getBean(JobLauncher.class);
Job job1 = context1.getBean("invoiceJob", Job.class);
JobExecution jobExecution = jobLauncher1.run(job1, jobParams1);
线程 2 :
JobLauncher jobLauncher1 = context1.getBean(JobLauncher.class);
Job job1 = context1.getBean("inventoryJob", Job.class);
JobExecution jobExecution = jobLauncher1.run(job1, jobParams1);
这种方法的优点是,每次有新文件要读取时,开发人员/用户所要做的就是子类AbstractFileLoader
并实现所需的抽象方法,而无需深入了解如何组装作业的细节。
问题 :
- 我是 Spring 批处理
- 的新手,所以我可能忽略了这种方法的一些不太明显的问题,例如 Spring 批处理中的共享内部对象可能会导致两个一起运行的作业失败或明显的问题,例如 bean 的范围。
- 有没有更好的方法来实现我的目标?
@Value("#{stepExecutionContext['fileName']}")
的fileName
属性始终被分配值为I:/CK/invoices/partitions/
这是getPath
方法在InvoiceLoader
中返回的值,即使 getPathmethod in
InventoryLoader'返回不同的值。
一种选择是将它们作为作业参数传递。例如:
@Bean
Job job() {
jobs.get("myJob").start(step1(null)).build()
}
@Bean
@JobScope
Step step1(@Value('#{jobParameters["commitInterval"]}') commitInterval) {
steps.get('step1')
.chunk((int) commitInterval)
.reader(new IterableItemReader(iterable: [1, 2, 3, 4], name: 'foo'))
.writer(writer(null))
.build()
}
@Bean
@JobScope
ItemWriter writer(@Value('#{jobParameters["writerClass"]}') writerClass) {
applicationContext.classLoader.loadClass(writerClass).newInstance()
}
与MyWriter
:
class MyWriter implements ItemWriter<Integer> {
@Override
void write(List<? extends Integer> items) throws Exception {
println "Write $items"
}
}
然后执行:
def jobExecution = launcher.run(ctx.getBean(Job), new JobParameters([
commitInterval: new JobParameter(3),
writerClass: new JobParameter('MyWriter'), ]))
输出为:
信息:执行步骤:[步骤1]写入 [1, 2, 3]写 [4]2016-2-24下午02:30:22 org.springframework.batch.core.launch.support.SimpleJobLauncher$1 run信息: 作业: [SimpleJob: [name=myJob]] 已完成,参数如下:[{commitInterval=3, writerClass=MyWriter}] 和以下状态: [已完成]状态为:已完成,作业执行 ID 0 #1 步骤1 已完成
完整示例在这里。