Spring Batch多依赖Reader实现多数据源合并的技术问题咨询

问题1：Job应该配置1个步骤还是2个步骤？ #

优先选择单步骤方案，理由如下：

• 你的业务是对每一条API返回的TodoDto，实时关联查询数据库对应数据后合并，属于同一条数据的流式处理链路，拆分两步需要额外存储第一步读取到的API数据作为中间态，增加不必要的IO开销和复杂度
• 单步骤可以直接复用chunk的批量处理特性，30条一批的处理逻辑不需要做任何调整，天然适配你的业务流

问题2：动态ID参数传递及数据关联实现 #

你当前的写法错误点在于把JdbcCursorItemReader配置为了Step的二级Reader，Spring Batch的单Step默认只能有一个主Reader，动态按ID查询的逻辑应该迁移到ItemProcessor中实现，不需要配置成独立的Reader，具体实现方案如下：

调整方案步骤 #

1. 去掉你现有配置中的step2，整个Job只保留一个Step，主Reader用你已经实现的TodoItemRestReader读取API数据
2. 注入JdbcTemplate到自定义的Processor中，在Processor中接收Reader读取到的TodoDto，取到id后执行SQL查询数据库对应的TodoBackendDto，然后合并两个对象返回合并后的结果
3. 自定义Writer接收合并后的对象，调用目标Web API完成写入

修正后的代码示例 #

@Bean
public Job todoJob() {
    final String jobName = "todos-migration-job";
    Step mergeStep = stepBuilderFactory.get("merge-todo-data").<TodoDto, MergedTodoDto>chunk(30)
            .reader(todoItemReader())
            .processor(todoMergeProcessor())
            .writer(mergedTodoWriter())
            .build();
    Job todoJob = jobBuilderFactory.get(jobName)
            .incrementer(new RunIdIncrementer())
            .start(mergeStep)
            .build();
    this.jobService.registerJob(jobName);
    return todoJob;
}

// 自定义合并Processor
@Bean
public ItemProcessor<TodoDto, MergedTodoDto> todoMergeProcessor() {
    return new ItemProcessor<TodoDto, MergedTodoDto>() {
        @Autowired
        private JdbcTemplate todoJdbcTemplate;
        
        @Override
        public MergedTodoDto process(TodoDto apiTodo) throws Exception {
            // 按Reader返回的动态ID查询数据库
            TodoBackendDto dbTodo = todoJdbcTemplate.queryForObject(
                "select id, title from todo_data where id = ?",
                new Object[]{apiTodo.getId()},
                new TodoItemRowMapper()
            );
            // 合并两个对象逻辑自行实现
            MergedTodoDto merged = new MergedTodoDto();
            merged.setId(apiTodo.getId());
            merged.setApiField(apiTodo.getXXX());
            merged.setDbField(dbTodo.getTitle());
            return merged;
        }
    };
}

@Bean
public TodoItemRestReader todoItemReader() {
    return new TodoItemRestReader();
}

// 自定义Writer调用目标Web API写入合并后数据
@Bean
public ItemWriter<MergedTodoDto> mergedTodoWriter() {
    return list -> {
        // 批量调用目标Web API的逻辑自行实现
    };
}

可选优化方案 #

如果你的数据库数据量不大，可以提前全量加载到内存的Map结构中，在Processor中直接按ID取数据，避免每条都查库，提升处理效率。

内容的提问来源于stack exchange，提问作者Sébastien Vallet