如何将记录ID写入Spring Batch的BATCH_*表实现审计？

嘿，针对你想把记录ID存入Spring Batch的BATCH_*表做审计的需求，我给你梳理几个实用方案，帮你选最适合的：

如果你的记录数量不多，想直接用现有的BATCH_STEP_EXECUTION_CONTEXT表，可以通过Step Execution Context来存储记录ID集合。具体步骤如下：

1. 在Chunk处理中收集记录ID
   你可以包装现有的ItemWriter，在完成写入后收集当前Chunk里的所有记录ID，然后存入Step Execution Context：

   public class AuditingItemWriter<T> implements ItemWriter<T> {
       private final ItemWriter<T> delegate;
       private StepExecution stepExecution;
   
       // 通过构造注入原有Writer
       public AuditingItemWriter(ItemWriter<T> delegate) {
           this.delegate = delegate;
       }
   
       // 注入StepExecution
       @BeforeStep
       public void beforeStep(StepExecution stepExecution) {
           this.stepExecution = stepExecution;
       }
   
       @Override
       public void write(List<? extends T> items) throws Exception {
           // 先执行原有的Topic写入逻辑
           delegate.write(items);
   
           // 收集当前Chunk的记录ID
           List<Long> currentChunkIds = items.stream()
                   .map(item -> ((YourDataEntity) item).getId()) // 替换成你的实体类获取ID的方法
                   .collect(Collectors.toList());
   
           // 更新Step Execution Context
           ExecutionContext context = stepExecution.getExecutionContext();
           List<Long> allAuditedIds = (List<Long>) context.get("auditedRecordIds");
           if (allAuditedIds == null) {
               allAuditedIds = new ArrayList<>();
           }
           allAuditedIds.addAll(currentChunkIds);
           context.put("auditedRecordIds", allAuditedIds);
       }
   }
   

2. 注意事项

   • BATCH_STEP_EXECUTION_CONTEXT的short_context字段是VARCHAR类型（默认长度有限），如果记录ID太多会导致数据溢出，此时可以改用long_context（CLOB类型），Spring Batch会自动处理序列化后的存储。
   • Step Execution Context主要用于存储作业/步骤的状态信息，而非业务审计数据，所以这个方案更适合少量记录的场景。

如果你的数据量较大，或者想做更规范的审计，建议创建一个专门的审计表（比如BATCH_RECORD_AUDIT），而不是复用Spring Batch的系统表。这种方式更灵活，也符合审计数据的存储规范。

步骤1：创建审计表 #

CREATE TABLE BATCH_RECORD_AUDIT (
    ID BIGINT AUTO_INCREMENT PRIMARY KEY,
    JOB_EXECUTION_ID BIGINT NOT NULL,
    STEP_EXECUTION_ID BIGINT NOT NULL,
    RECORD_ID BIGINT NOT NULL,
    PROCESS_TIMESTAMP TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
    FOREIGN KEY (JOB_EXECUTION_ID) REFERENCES BATCH_JOB_EXECUTION(JOB_EXECUTION_ID),
    FOREIGN KEY (STEP_EXECUTION_ID) REFERENCES BATCH_STEP_EXECUTION(STEP_EXECUTION_ID)
);

步骤2：通过ItemWriteListener实现批量审计 #

用Spring Batch的ItemWriteListener可以在每次Chunk写入完成后，批量插入审计记录，性能更优：

@Component
public class RecordAuditListener implements ItemWriteListener<YourDataEntity> {
    private final JdbcTemplate jdbcTemplate;
    private StepExecution stepExecution;

    @Autowired
    public RecordAuditListener(JdbcTemplate jdbcTemplate) {
        this.jdbcTemplate = jdbcTemplate;
    }

    @BeforeStep
    public void beforeStep(StepExecution stepExecution) {
        this.stepExecution = stepExecution;
    }

    @Override
    public void afterWrite(List<? extends YourDataEntity> items) {
        // 批量准备审计参数
        List<Object[]> auditParams = items.stream()
                .map(item -> new Object[]{
                        stepExecution.getJobExecutionId(),
                        stepExecution.getId(),
                        item.getId()
                })
                .collect(Collectors.toList());

        // 批量插入审计记录
        jdbcTemplate.batchUpdate(
            "INSERT INTO BATCH_RECORD_AUDIT (JOB_EXECUTION_ID, STEP_EXECUTION_ID, RECORD_ID) VALUES (?, ?, ?)",
            auditParams
        );
    }
}

步骤3：在Step中注册Listener #

@Bean
public Step dataExportStep(ItemReader<YourDataEntity> reader, 
                           ItemProcessor<YourDataEntity, YourDataEntity> processor, 
                           ItemWriter<YourDataEntity> writer,
                           RecordAuditListener auditListener) {
    return stepBuilderFactory.get("dataExportStep")
            .<YourDataEntity, YourDataEntity>chunk(100) // 你的Chunk大小
            .reader(reader)
            .processor(processor)
            .writer(writer)
            .listener(auditListener) // 注册审计Listener
            .build();
}

如果想在处理每条数据时就完成审计（比如记录处理时间、状态等），可以在ItemProcessor中加入审计逻辑：

public class AuditingProcessor implements ItemProcessor<YourDataEntity, YourDataEntity> {
    private final JdbcTemplate jdbcTemplate;
    private StepExecution stepExecution;

    @Autowired
    public AuditingProcessor(JdbcTemplate jdbcTemplate) {
        this.jdbcTemplate = jdbcTemplate;
    }

    @BeforeStep
    public void beforeStep(StepExecution stepExecution) {
        this.stepExecution = stepExecution;
    }

    @Override
    public YourDataEntity process(YourDataEntity item) throws Exception {
        // 先执行数据处理逻辑（如果有的话）
        // ...

        // 插入审计记录
        jdbcTemplate.update(
            "INSERT INTO BATCH_RECORD_AUDIT (JOB_EXECUTION_ID, STEP_EXECUTION_ID, RECORD_ID) VALUES (?, ?, ?)",
            stepExecution.getJobExecutionId(),
            stepExecution.getId(),
            item.getId()
        );
        return item;
    }
}

总结 #

• 如果只是临时少量记录审计，方案一（Step Context）可以快速实现，但不适合大规模数据。
• 方案二（自定义审计表+Listener）是最推荐的方式，既符合Spring Batch的设计理念，又能满足审计的扩展性需求（比如后续要加处理状态、操作人等字段）。

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