Spring Batch Remote Chunking(Kafka)：如何在ChunkResponse中传DTO并更新数据库计数

1. 扩展ChunkResponse携带自定义处理信息 #

Spring Batch的ChunkResponse支持扩展，可继承它添加自定义字段（比如处理完成的记录数），注意保留父类核心构造方法以兼容序列化逻辑：

public class ExtendedChunkResponse extends ChunkResponse {
    private int processedRecordCount;

    // 兼容父类反序列化的构造方法
    public ExtendedChunkResponse(long jobId, long stepExecutionId, int sequence, boolean status, Throwable throwable) {
        super(jobId, stepExecutionId, sequence, status, throwable);
    }

    // Worker构造成功响应时使用的自定义构造方法
    public ExtendedChunkResponse(long jobId, long stepExecutionId, int sequence, boolean status, int processedRecordCount) {
        super(jobId, stepExecutionId, sequence, status, null);
        this.processedRecordCount = processedRecordCount;
    }

    // Getter & Setter
    public int getProcessedRecordCount() {
        return processedRecordCount;
    }

    public void setProcessedRecordCount(int processedRecordCount) {
        this.processedRecordCount = processedRecordCount;
    }
}

2. 自定义Worker端ChunkHandler，构造扩展响应 #

重写Worker的ChunkHandler，在处理完Chunk后统计记录数，将结果封装进ExtendedChunkResponse返回：

@Component
public class CustomChunkHandler implements ChunkHandler {

    @Autowired
    private YourItemProcessor itemProcessor;

    @Override
    public ChunkResponse handleChunk(ChunkRequest chunkRequest) throws Exception {
        Chunk<?> chunk = chunkRequest.getChunk();
        int processedCount = 0;

        try {
            // 批量处理Chunk内记录并计数
            for (Object item : chunk.getItems()) {
                itemProcessor.process(item);
                processedCount++;
            }
            // 返回带处理记录数的成功响应
            return new ExtendedChunkResponse(
                    chunkRequest.getJobId(),
                    chunkRequest.getStepExecutionId(),
                    chunkRequest.getSequence(),
                    true,
                    processedCount
            );
        } catch (Exception e) {
            // 处理失败时返回带异常信息的响应
            return new ExtendedChunkResponse(
                    chunkRequest.getJobId(),
                    chunkRequest.getStepExecutionId(),
                    chunkRequest.getSequence(),
                    false,
                    e
            );
        }
    }
}

3. 自定义Manager端ChunkResponseHandler，处理扩展响应 #

自定义Manager的响应处理器，先完成父类默认的StepExecution状态更新，再提取扩展字段更新数据库计数：

@Component
public class CustomChunkResponseHandler extends SimpleChunkResponseHandler {

    @Autowired
    private ProgressRepository progressRepository;

    @Override
    public void handleChunkResponse(ChunkResponse chunkResponse) throws Exception {
        // 先执行父类逻辑，维护StepExecution的状态
        super.handleChunkResponse(chunkResponse);

        // 校验响应类型并提取处理记录数
        if (chunkResponse instanceof ExtendedChunkResponse) {
            ExtendedChunkResponse extendedResponse = (ExtendedChunkResponse) chunkResponse;
            int processedCount = extendedResponse.getProcessedRecordCount();

            // 原子更新数据库计数，避免并发冲突
            progressRepository.incrementProcessedCount(processedCount);
        }
    }
}

4. 配置Kafka消息序列化/反序列化 #

为了让自定义的ExtendedChunkResponse能在Kafka中正常传输，需配置消息转换器支持JSON序列化（以Jackson为例）：

@Configuration
public class KafkaRemoteChunkingConfig {

    // Manager发送ChunkRequest的Kafka模板
    @Bean
    public KafkaTemplate<String, ChunkRequest> managerKafkaTemplate(ProducerFactory<String, ChunkRequest> producerFactory) {
        KafkaTemplate<String, ChunkRequest> template = new KafkaTemplate<>(producerFactory);
        template.setMessageConverter(new MappingJackson2MessageConverter());
        return template;
    }

    // Worker接收ChunkRequest的容器工厂
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, ChunkRequest> workerChunkRequestContainerFactory(
            ConsumerFactory<String, ChunkRequest> consumerFactory) {
        ConcurrentKafkaListenerContainerFactory<String, ChunkRequest> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        factory.setMessageConverter(new MappingJackson2MessageConverter());
        return factory;
    }

    // Worker发送ExtendedChunkResponse的Kafka模板
    @Bean
    public KafkaTemplate<String, ExtendedChunkResponse> workerKafkaTemplate(ProducerFactory<String, ExtendedChunkResponse> producerFactory) {
        KafkaTemplate<String, ExtendedChunkResponse> template = new KafkaTemplate<>(producerFactory);
        template.setMessageConverter(new MappingJackson2MessageConverter());
        return template;
    }

    // Manager接收ExtendedChunkResponse的容器工厂
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, ExtendedChunkResponse> managerChunkResponseContainerFactory(
            ConsumerFactory<String, ExtendedChunkResponse> consumerFactory) {
        ConcurrentKafkaListenerContainerFactory<String, ExtendedChunkResponse> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        factory.setMessageConverter(new MappingJackson2MessageConverter());
        return factory;
    }
}

5. 保障计数更新的可靠性 #

• 事务控制：在数据库更新方法上添加@Transactional注解，确保计数操作原子性。
• 并发安全：使用数据库原子更新语句（如UPDATE progress SET processed_count = processed_count + ? WHERE id = ?），避免多Worker并发更新导致的数据丢失。
• 异常重试：如果数据库更新失败，可配置Spring Retry进行重试，或将失败响应存入死信队列后续手动处理。

内容的提问来源于stack exchange，提问作者akashsharma3030