Spring Batch+KafkaItemWriter：异步发送回调与错误处理技术问询

一、先澄清：KafkaItemWriter 的发送模式 #

KafkaItemWriter 默认并非异步发送。它底层依赖 Spring Kafka 的 KafkaTemplate，其 write 方法会循环处理每个待发送的 Item，调用 KafkaTemplate.send() 获取 ListenableFuture 后，会阻塞等待所有 Future 执行完成才会返回。也就是说，步骤执行到 KafkaItemWriter 的写入阶段时，会等到所有消息的发送结果（成功/失败）都返回后，才会进入后续流程，不存在「步骤结束前没收到回调」的问题——除非你手动修改了异步配置。

二、用监听器记录发送结果到数据库 #

完全可以通过监听器实现状态更新，以下是两种常用方案：

1. Spring Batch 原生监听器（ItemWriteListener/StepExecutionListener） #

• ItemWriteListener：监听写入后的成功/错误事件，直接拿到对应 Item 集合更新数据库状态：

  @Component
  public class KafkaWriteStatusListener implements ItemWriteListener<YourBizItem> {
  
      private final BizStatusRepository statusRepo;
  
      public KafkaWriteStatusListener(BizStatusRepository statusRepo) {
          this.statusRepo = statusRepo;
      }
  
      @Override
      public void afterWrite(List<? extends YourBizItem> items) {
          items.forEach(item -> statusRepo.updateSendStatus(item.getId(), "SENT", null));
      }
  
      @Override
      public void onWriteError(Exception e, List<? extends YourBizItem> items) {
          items.forEach(item -> statusRepo.updateSendStatus(item.getId(), "FAILED", e.getMessage()));
      }
  }
  

  注册到 Step 配置中：

  @Bean
  public Step kafkaSendStep(StepBuilderFactory stepFactory,
                            ItemReader<YourBizItem> itemReader,
                            KafkaItemWriter<String, YourBizItem> kafkaWriter,
                            KafkaWriteStatusListener statusListener) {
      return stepFactory.get("kafkaSendStep")
              .<YourBizItem, YourBizItem>chunk(100)
              .reader(itemReader)
              .writer(kafkaWriter)
              .listener(statusListener)
              .build();
  }
  

• StepExecutionListener：适合在整个步骤结束后统一处理批量状态（比如统计成功/失败总数），在 afterStep 方法中通过 StepExecution 获取执行结果后更新数据库。

2. Spring Kafka ProducerListener（细粒度单条消息回调） #

如果需要精准捕获每条消息的发送结果，可以给 KafkaTemplate 配置 ProducerListener，每条消息发送完成后立即更新对应状态：

@Bean
public ProducerListener<String, YourBizItem> kafkaProducerListener(BizStatusRepository statusRepo) {
    return new ProducerListener<>() {
        @Override
        public void onSuccess(ProducerRecord<String, YourBizItem> record, RecordMetadata meta) {
            YourBizItem item = record.value();
            statusRepo.updateSendStatus(item.getId(), "SENT", null);
        }

        @Override
        public void onError(ProducerRecord<String, YourBizItem> record, Exception e) {
            YourBizItem item = record.value();
            statusRepo.updateSendStatus(item.getId(), "FAILED", e.getMessage());
        }
    };
}

@Bean
public KafkaTemplate<String, YourBizItem> kafkaTemplate(ProducerFactory<String, YourBizItem> producerFactory,
                                                       ProducerListener<String, YourBizItem> listener) {
    KafkaTemplate<String, YourBizItem> template = new KafkaTemplate<>(producerFactory);
    template.setProducerListener(listener);
    return template;
}

由于 KafkaItemWriter 默认阻塞等待所有发送完成，这些回调会在 Step 的写入阶段内全部执行完毕，不会出现步骤结束后回调仍在运行的情况。

三、KafkaItemWriter 错误处理方案 #

1. Chunk 级重试与跳过：通过 Step 的容错配置，针对不同 Kafka 异常设置重试或跳过逻辑：

   return stepFactory.get("kafkaSendStep")
           .<YourBizItem, YourBizItem>chunk(100)
           .reader(itemReader)
           .writer(kafkaWriter)
           .faultTolerant()
           .retryLimit(3)
           .retry(RetriableKafkaException.class) // 只重试可恢复的异常（如网络超时）
           .skip(NonRetriableKafkaException.class) // 跳过不可恢复的异常（如主题不存在）
           .skipLimit(10)
           .listener(new SkipListener<YourBizItem, YourBizItem>() {
               @Override
               public void onSkipInWrite(YourBizItem item, Throwable t) {
                   statusRepo.updateSendStatus(item.getId(), "SKIPPED", t.getMessage());
               }
           })
           .build();
   

2. 自定义错误处理：包装 KafkaItemWriter，捕获异常后根据类型执行差异化处理（比如区分「发送超时」和「权限不足」异常，设置不同的状态码）。

3. 事务一致性：开启 Kafka 生产者事务，并结合 Spring Batch 的事务管理器，保证「消息发送成功」和「数据库状态更新」的原子性——只有消息发送成功，状态才会更新为成功；发送失败则回滚状态更新。

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