java - 使用 Tasklet 在 Spring Batch 中或在 CompositeItemWriter 中创建逻辑？

Question

我创建了 3 个文件编写器，它们创建了 3 个本地文件。但读取的所有信息都以相同的方式进入每个文件。我正在尝试设置一些逻辑，以便只有被调用的项目才会进入其正确的文件。例如:创建文件A、文件B、文件C。当读者访问 MONGODB 时，它应该解析 buisness 列并找到 3 种不同的类型:TypeA、TypeB 和 TypeC。一旦读者找到不同的类型，就应该将其添加到适当的文件中，而不是将所有内容打印到创建的每个文件中。 FileA 应该只包含 TypeA 的信息，除此之外不应写入任何其他信息。有办法做到吗？这是我的代码的样子。

项目阅读器:

  @Bean
  public MongoItemReader<PaymentAudit> mongoreader() {
    LOG.debug("Mongo-Reader");    
    @SuppressWarnings("unchecked")
    MongoItemReader<PaymentAudit> mongoreader = new MongoItemReader();
    mongoreader.setTemplate(mongoTemplate);
    mongoreader.setQuery("{}");
    mongoreader.setTargetType(PaymentAudit.class);
    mongoreader.setSort(new HashMap<String, Sort.Direction>() {
        {
            put("_id", Direction.ASC);
        }
    });
    return mongoreader;
  }

文件项编写器:

   @StepScope
@Bean
public FlatFileItemWriter<PaymentAudit> writer() {
    LOG.debug("Mongo-writer");  
    String exportFilePath="C:\\filewriter\\retail.txt";
    FlatFileItemWriter<PaymentAudit> flatFile = new 
 FlatFileItemWriterBuilder<PaymentAudit>()
            .name("")
            .resource(new FileSystemResource(exportFilePath))
            .lineAggregator(createPaymentPortalLineAggregator())
            .build();
    String exportFileHeader = "TypeA";
    StringHeaderWriter headerWriter = new 
  StringHeaderWriter(exportFileHeader);
    flatFile.setHeaderCallback(headerWriter);
    return flatFile;

}
@Bean
public FlatFileItemWriter<PaymentAudit> writer2() {
    LOG.debug("flatFileItemWriter");    
    String exportFilePath="C:\\filewriter\\hcc.txt";
    FlatFileItemWriter<PaymentAudit> flatFile = new 
  FlatFileItemWriterBuilder<PaymentAudit>()
            .name("")
            .resource(new FileSystemResource(exportFilePath))
            .lineAggregator(createPaymentPortalLineAggregator())
            .build();
    String exportFileHeader = "TypeB";
    StringHeaderWriter headerWriter = new 
    StringHeaderWriter(exportFileHeader);
    flatFile.setHeaderCallback(headerWriter);
    return flatFile;

}
@Bean
public FlatFileItemWriter<PaymentAudit> writer3() {
    LOG.debug("Mongo-writer");  
    String exportFilePath="C:\\filewriter\\srx.txt";
    FlatFileItemWriter<PaymentAudit> flatFile = new 
    FlatFileItemWriterBuilder<PaymentAudit>()
            .name("")
            .resource(new FileSystemResource(exportFilePath))
            .lineAggregator(createPaymentPortalLineAggregator())
            .build();
    String exportFileHeader = "TypeC";
    StringHeaderWriter headerWriter = new 
    StringHeaderWriter(exportFileHeader);
    flatFile.setHeaderCallback(headerWriter);
    return flatFile;

}

@SuppressWarnings({ "unchecked", "rawtypes" })
public CompositeItemWriter<PaymentAudit> compositeItemWriter(){
    CompositeItemWriter writer = new CompositeItemWriter();


  private LineAggregator<PaymentAudit> createPaymentPortalLineAggregator() {
    DelimitedLineAggregator<PaymentAudit> lineAggregator = new 
    DelimitedLineAggregator<>();
    lineAggregator.setDelimiter("|");
    FieldExtractor<PaymentAudit> fieldExtractor = 
    createPaymentPortalFieldExtractor();
    lineAggregator.setFieldExtractor(fieldExtractor);
    return lineAggregator;
}

private FieldExtractor<PaymentAudit> createPaymentPortalFieldExtractor() {
    BeanWrapperFieldExtractor<PaymentAudit> extractor = new 
BeanWrapperFieldExtractor<>();
    extractor.setNames(new String[] { "TypeA, TypeB, TypeC"});      
    return extractor;
}

Answer 1

您需要使用ClassifierCompositeItemWriter为了对项目进行分类并将每种类型写入其相应的文件中。这是一个您可以尝试的简单示例:

import java.util.Arrays;

import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.file.FlatFileItemWriter;
import org.springframework.batch.item.file.builder.FlatFileItemWriterBuilder;
import org.springframework.batch.item.file.transform.PassThroughLineAggregator;
import org.springframework.batch.item.support.ClassifierCompositeItemWriter;
import org.springframework.batch.item.support.ListItemReader;
import org.springframework.classify.Classifier;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.FileSystemResource;

@Configuration
@EnableBatchProcessing
public class MyJob {

    private JobBuilderFactory jobBuilderFactory;

    private StepBuilderFactory stepBuilderFactory;

    public MyJob(JobBuilderFactory jobBuilderFactory, StepBuilderFactory stepBuilderFactory) {
        this.jobBuilderFactory = jobBuilderFactory;
        this.stepBuilderFactory = stepBuilderFactory;
    }

    @Bean
    public ItemReader<Person> itemReader() {
        Person foo1 = new Person();foo1.setId(1);foo1.setName("foo1");
        Person foo2 = new Person();foo2.setId(2);foo2.setName("foo2");
        Person bar1 = new Person();bar1.setId(3);bar1.setName("bar1");
        Person bar2 = new Person();bar2.setId(4);bar2.setName("bar2");
        return new ListItemReader<>(Arrays.asList(foo1, foo2, bar1, bar2));
    }

    @Bean
    public ClassifierCompositeItemWriter<Person> classifierCompositeItemWriter(ItemWriter<Person> fooItemWriter, ItemWriter<Person> barItemWriter) {
        ClassifierCompositeItemWriter<Person> classifierCompositeItemWriter = new ClassifierCompositeItemWriter<>();
        classifierCompositeItemWriter.setClassifier((Classifier<Person, ItemWriter<? super Person>>) person -> {
            if (person.getName().startsWith("foo")) {
                return fooItemWriter;
            } else {
                return barItemWriter;
            }
        });
        return classifierCompositeItemWriter;
    }

    @Bean
    public FlatFileItemWriter<Person> fooItemWriter() {
        return new FlatFileItemWriterBuilder<Person>()
                .name("fooItemWriter")
                .resource(new FileSystemResource("foos.txt"))
                .lineAggregator(new PassThroughLineAggregator<>())
                .build();
    }

    @Bean
    public FlatFileItemWriter<Person> barItemWriter() {
        return new FlatFileItemWriterBuilder<Person>()
                .name("barItemWriter")
                .resource(new FileSystemResource("bars.txt"))
                .lineAggregator(new PassThroughLineAggregator<>())
                .build();
    }

    @Bean
    public Step dataExtractionStep() {
        return stepBuilderFactory.get("dataExtractionStep")
                .<Person, Person>chunk(2)
                .reader(itemReader())
                .writer(classifierCompositeItemWriter(fooItemWriter(), barItemWriter()))
                .stream(fooItemWriter())
                .stream(barItemWriter())
                .build();
    }

    @Bean
    public Job dataExtractionJob() {
        return jobBuilderFactory.get("dataExtractionJob")
                .start(dataExtractionStep())
                .build();
    }

    public static void main(String[] args) throws Exception {
        ApplicationContext context = new AnnotationConfigApplicationContext(MyJob.class);
        JobLauncher jobLauncher = context.getBean(JobLauncher.class);
        Job job = context.getBean(Job.class);
        jobLauncher.run(job, new JobParameters());
    }

    public static class Person {

        private int id;

        private String name;

        public Person() {
        }

        public Person(int id, String name) {
            this.id = id;
            this.name = name;
        }

        public int getId() {
            return id;
        }

        public void setId(int id) {
            this.id = id;
        }

        public String getName() {
            return name;
        }

        public void setName(String name) {
            this.name = name;
        }

        @Override
        public String toString() {
            return "Person{" +
                    "id=" + id +
                    ", name='" + name + '\'' +
                    '}';
        }
    }

}

此示例读取一些 Person 项目，并将名称为 foo* 的项目写入 foos.txt 以及名称为 bar* 的项目 到 bars.txt。

希望这有帮助。