Spring 批处理 - ItemWriter 正在写入由 ItemReader 读取的相同对象，但不是通过 ItemProcessor 处理后返回的对象

Question

我的场景是：JdbcPagingItemReader 正在从 Oracle 数据库读取并且 returning 对象假设 'Employee'。然后这个 'Employee' 对象被传递给 Processor 以再次调用 db 以从多个表和 return 'AggregatedEmployee' 对象（它实际上扩展了 Employee）中提取更多信息。我正在使用 KafkaItemWriter 将处理后的对象写入 Kafka，而不是写入 AggregatedEmployee，作者正在尝试自己写入 'Employee'。

@Mahmoud Ben Hassine：我看到了你在 Spring 上的很多建议 Batch.Please 分享你的想法。

处理器接口代码：

public interface PageProcessor<T> {
       <R extends Employee> R process(T page);
}

步骤 Bean 代码：

@Bean
protected Step step1 (CompositeJdbcPagingItemReader <Employee> reader, KafkaItemWriter <String, AggregatedEmployee> writer) {
    return steps.get("step1")
                .<Employee, AggregatedEmployee>chunk(5).
                reader(reader).
                writer(writer).build();
    }

ProcessorInterface 代码的实现 Class：

public class EmployeeProcessor implements PageProcessor<Employee> {
    private NamedParameterJdbcTemplate jdbcTemplate;

    public void setDataSource(DataSource dataSource) {
        jdbcTemplate = new NamedParameterJdbcTemplate(dataSource);
    }

    @SuppressWarnings("unchecked")
    @Override
    public <R extends Employee> R process(Employee page) {
             ... implementation goes here
    }

KafkaItemWriter Bean：

    @Bean
    KafkaItemWriter<String,AggregatedEmployee> writer(){
        return new KafkaItemWriterBuilder<String, AggregatedEmployee>()
                .kafkaTemplate(aggregatedEmployeekafkaTemplate)
                .itemKeyMapper(aggregatedEmployee -> String.valueOf(aggregatedEmployee.getEmployeeId()))
                .build();
    }

编辑显示处理器：

public class CompositeJdbcPagingItemReader<T> extends JdbcPagingItemReader<T> {
    private PageProcessor<T> pageProcessor;

    public void setPageProcessor(PageProcessor<T> pageProcessor) {
        this.pageProcessor = pageProcessor;
    }

并且当Reader bean被创建时，处理器对象也被创建并设置到reader中，通过上面显示setter并且在EmployeeProcessor中编写的处理器逻辑也被执行。

错误：

java.lang.ClassCastException: class com.sample.model.Employee cannot be cast to class com.sample.model.AggregatedEmployee (com.sample.model.Employee and com.sample.model.AggregatedEmployee are in unnamed module of loader 'app')
    at org.springframework.batch.item.KeyValueItemWriter.write(KeyValueItemWriter.java:43)

Answer 1

您没有在步骤中设置处理器：

@Bean
protected Step step1 (CompositeJdbcPagingItemReader <Employee> reader, KafkaItemWriter <String, AggregatedEmployee> writer) {
return steps.get("step1")
            .<Employee, AggregatedEmployee>chunk(5).
            reader(reader).
            writer(writer).build();
}

您需要设置执行类型转换的处理器 Employee -> AggregatedEmployee。

Spring 批处理 - ItemWriter 正在写入由 ItemReader 读取的相同对象，但不是通过 ItemProcessor 处理后返回的对象

Spring Batch - ItemWriter is writing same object read by ItemReader but not the one returned after processing through ItemProcessor

java

spring

spring-batch

apache-kafka

spring-boot