背景
我正在设计一个文件读取层，它可以读取分隔文件并将其加载到List。我决定使用 Spring Batch，因为它提供了许多可扩展性选项，我可以根据文件的大小将其用于不同的文件集。
要求
- 我想设计一个通用的作业 API，可用于读取任何分隔文件。
- 应该有一个作业结构用于解析每个分隔文件。例如，如果系统需要读取5个文件，则将有5个作业（每个文件一个）。这 5 个作业唯一不同的地方是它们使用不同的
FieldSetMapper、列名、目录路径和其他缩放参数，例如commit-interval and throttle-limit.
- 该API的用户不需要配置Spring
当系统中引入新的文件类型时，可以自行进行批处理、步骤、分块、分区等。
- 用户需要做的就是提供
FieldsetMapper与作业一起使用commit-interval, throttle-limit以及每种类型的文件将放置的目录。
- 每个文件将有一个预定义目录。每个目录可以包含多个相同类型和格式的文件。 A
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 Batch 的新手，所以我可能忽略了这种方法的一些不太明显的问题，例如 Spring Batch 中的共享内部对象可能会导致两个作业一起运行失败，或者明显的问题，例如 bean 的范围。
- 有更好的方法来实现我的目标吗？
- The
fileName的属性@Value("#{stepExecutionContext['fileName']}")总是被赋值为I:/CK/invoices/partitions/这是返回的值getPath中的方法InvoiceLoader即使 getPathmethod inInventoryLoader`返回不同的值。
