Spring Batch - 如何从一个表中读取并将数据写入两个不同的表

Question

我正在使用Spring Boot and Spring Batch从源数据库表的一个表中读取数据并将数据拆分并写入目标数据库的两个表中。

我为此选择使用CompositeItemWriter，但CompositeItemWriter<?>只有一种类型。我想在一个表中写入几个字段，在另一个表中写入其他字段。

说：老客户和新客户。

错误：

构造函数 CustomerClassifier(JdbcBatchItemWriter, JdbcBatchItemWriter) 未定义

ClassifierCompositeItemApplication.java

@EnableBatchProcessing
@SpringBootApplication
public class ClassifierCompositeItemApplication {

private JobBuilderFactory jobBuilderFactory;
private StepBuilderFactory stepBuilderFactory;

public ClassifierCompositeItemApplication(JobBuilderFactory jobs, StepBuilderFactory steps) {
    this.jobBuilderFactory = jobs;
    this.stepBuilderFactory = steps;
}


@Value("classpath:input/customer.csv")
private Resource inputResource;

@Bean
@StepScope
public FlatFileItemReader<Customer> classifierCompositeWriterItemReader() {
    return new FlatFileItemReaderBuilder<Customer>()
            .name("customerFileReader")
            .resource(inputResource).delimited()
            .names(new String[] { "firstName", "middleInitial", "lastName", "address", "city", "state", "zip" })
            .targetType(Customer.class)
            .build();
}


@Bean
public ClassifierCompositeItemWriter<Customer> compositeItemWriter() throws IOException {
    final Classifier<Customer, ItemWriter<? super Customer>> classifier = new CustomerClassifier(
            this.customer1(null), this.customer2(null));

    return new ClassifierCompositeItemWriterBuilder<Customer>().classifier(classifier).build();
}



@Bean
public JdbcBatchItemWriter<Customer> customer1(DataSource dataSource) {
    return new JdbcBatchItemWriterBuilder<Customer>()
            .namedParametersJdbcTemplate(new NamedParameterJdbcTemplate(dataSource))
            .sql("INSERT INTO TBL_CUSTOMER_WRITER (firstname, middleinitial, lastname, address, city, " + "state, "
                    + "zipcode) " + "VALUES(:firstName, " + ":middleInitial, " + ":lastName, " + ":address, "
                    + ":city, " + ":state, " + ":zip)")
            .beanMapped().build();
}



@Bean
public JdbcBatchItemWriter<NewCustomer> customer2(DataSource dataSource) {
    return new JdbcBatchItemWriterBuilder<NewCustomer>()
            .namedParametersJdbcTemplate(new NamedParameterJdbcTemplate(dataSource))
            .sql("INSERT INTO TBL_CUSTOMER_WRITER (firstname, middleinitial, lastname, address, city, " + "state, "
                    + "zipcode) " + "VALUES(:firstName, " + ":middleInitial, " + ":lastName, " + ":address, "
                    + ":city, " + ":state, " + ":zip)")
            .beanMapped().build();
}



@Bean
public Step classifierCompositeWriterStep() throws IOException {
    return this.stepBuilderFactory.get("compositeWriterStep")
            .<Customer, Customer>chunk(10)
            .reader(this.classifierCompositeWriterItemReader())
            .writer(this.compositeItemWriter())
            .stream(this.customer1(null))
            .stream(this.customer2(null))
            .build();
}


@Bean
public Job classifierCompositeWriterJob() throws IOException {
    return this.jobBuilderFactory.get("compositeWriterJob").start(this.classifierCompositeWriterStep()).build();

}

public static void main(String[] args) {
    SpringApplication.run(ClassifierCompositeItemApplication.class, args);
}

}

CustomerClassifier.java

@AllArgsConstructor
public class CustomerClassifier implements Classifier<Customer, ItemWriter<? super Customer>> {
private static final long serialVersionUID = 1L;
private final ItemWriter<Customer> customer1;
private final ItemWriter<Customer> customer2;

@Override
public ItemWriter<? super Customer> classify(Customer customer) {
    if (customer.getState().matches("^[A-M].*")) {
        return customer1;
    } else {
        return customer2;
    }
}
}

Answer 1

您可以使用ClassifierCompositeItemWriter。此复合编写器旨在对项目进行分类并为每个类调用一个委托项目编写器。

因此，在您的情况下，您可以为老客户设置一个项目编写器，为新客户设置另一个项目编写器，并将它们扭曲到分类器项目编写器中。您可以使用 Spring 提供的 Classifier 实现之一或创建自定义实现（例如，项目处理器 cam 将您的项目标记为旧/新，并且分类器使用此标志对它们进行分类）。

编辑：添加示例：

import java.util.Arrays;

import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobParameters;
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.support.ClassifierCompositeItemWriter;
import org.springframework.batch.item.support.ListItemReader;
import org.springframework.batch.item.support.builder.ClassifierCompositeItemWriterBuilder;
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;

@Configuration
@EnableBatchProcessing
public class MyJob {

    @Bean
    public ItemReader<Customer> itemReader() {
        return new ListItemReader<>(Arrays.asList(new Customer("foo"), new Customer("bar")));
    }

    @Bean
    public ItemWriter<Customer> fooWriter() {
        return items -> {
            for (Customer item : items) {
                System.out.println("foo writer: item " + item.name);
            }
        };
    }

    @Bean
    public ItemWriter<Customer> barWriter() {
        return items -> {
            for (Customer item : items) {
                System.out.println("bar writer: item " + item.name);
            }
        };
    }

    @Bean
    public ClassifierCompositeItemWriter<Customer> classifierCompositeItemWriter() {
        final Classifier<Customer, ItemWriter<? super Customer>> classifier = 
                new CustomerClassifier(this.fooWriter(), this.barWriter());

        return new ClassifierCompositeItemWriterBuilder<Customer>()
                .classifier(classifier)
                .build();
    }

    @Bean
    public Job job(JobBuilderFactory jobs, StepBuilderFactory steps) {
        return jobs.get("job")
                .start(steps.get("step")
                        .<Customer, Customer>chunk(5)
                        .reader(itemReader())
                        .writer(classifierCompositeItemWriter())
                        .build())
                .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());
    }

    static class Customer {
        String name;

        public Customer(String name) {
            this.name = name;
        }
    }

    static class CustomerClassifier implements Classifier<Customer, ItemWriter<? super Customer>> {

        private ItemWriter<? super Customer> fooItemWriter;
        private ItemWriter<? super Customer> barItemWriter;

        public CustomerClassifier(ItemWriter<? super Customer> fooItemWriter, ItemWriter<? super Customer> barItemWriter) {
            this.fooItemWriter = fooItemWriter;
            this.barItemWriter = barItemWriter;
        }

        @Override
        public ItemWriter<? super Customer> classify(Customer customer) {
            return customer.name.startsWith("f") ? fooItemWriter : barItemWriter;
        }
    }

}

打印出来：

foo writer: item foo
bar writer: item bar