从多个源读取 Spring 批处理作业

引言

java
@Configuration
@EnableBatchProcessing
public class MultiSourceBatchJobConfig {
    @Autowired
    private JobBuilderFactory jobBuilderFactory;
    @Autowired
    private StepBuilderFactory stepBuilderFactory;
    @Autowired
    private DataSource dataSource; // 注入数据源
    @Bean
    public ItemReader userReader() {
        JdbcCursorItemReader reader = new JdbcCursorItemReader<>();
        reader.setDataSource(dataSource);
        reader.setSql("SELECT id, username, email FROM users");
        reader.setRowMapper(new BeanPropertyRowMapper<>(User.class));
        return reader;
    }
    @Bean
    public ItemReader orderReader() {
        JdbcCursorItemReader reader = new JdbcCursorItemReader<>();
        reader.setDataSource(dataSource);
        reader.setSql("SELECT id, user_id, total_amount FROM orders");
        reader.setRowMapper(new BeanPropertyRowMapper<>(Order.class));
        return reader;
    }
    @Bean
    public ItemProcessor userProcessor() {
        return new UserProcessor();
    }
    @Bean
    public ItemProcessor orderProcessor() {
        return new OrderProcessor();
    }
    @Bean
    public ItemWriter enrichedUserWriter() {
        JdbcBatchItemWriter writer = new JdbcBatchItemWriter<>();
        writer.setDataSource(dataSource);
        writer.setSql("INSERT INTO enriched_users (id, username, email, total_order_amount) VALUES (:id, :username, :email, :totalOrderAmount)");
        writer.setItemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<>());
        return writer;
    }
    @Bean
    public Step multiSourceStep(ItemReader userReader, ItemReader orderReader,
                                ItemProcessor userProcessor,
                                ItemProcessor orderProcessor,
                                ItemWriter enrichedUserWriter) {
        return stepBuilderFactory.get("multiSourceStep")
                .chunk(10)
                .reader(userReader)
                .processor(userProcessor)
                .writer(enrichedUserWriter)
                .build();
    }
    @Bean
    public Job multiSourceJob(Step multiSourceStep) {
        return jobBuilderFactory.get("multiSourceJob")
                .start(multiSourceStep)
                .build();
    }
}