如何让Processor/Writer终止Chunk Processing且不触发作业失败回滚？

针对你遇到的Processor/Writer检测到终止条件但Reader无法感知的场景，这里有几个实用的原生方案，能保证当前Chunk事务正常提交、Job不失败：

方案一：使用StepExecution的终止标记（最简便） #

Spring Batch提供了stepExecution.setTerminateOnly()方法，这是最直接的方式。当Processor或Writer检测到配额耗尽等终止条件时，调用这个方法，Spring Batch会在当前Chunk处理完成并提交事务后，自动终止Step循环，不会标记Job失败。

代码示例（Processor实现） #

public class QuotaCheckProcessor implements ItemProcessor<DbRecord, ProcessedRecord> {
    private final QuotaManager quotaManager;
    private StepExecution stepExecution;

    // 注入依赖
    public QuotaCheckProcessor(QuotaManager quotaManager) {
        this.quotaManager = quotaManager;
    }

    // 通过@BeforeStep获取StepExecution实例
    @BeforeStep
    public void initStepExecution(StepExecution stepExecution) {
        this.stepExecution = stepExecution;
    }

    @Override
    public ProcessedRecord process(DbRecord item) throws Exception {
        // 先正常处理当前记录
        ProcessedRecord processed = doProcess(item);
        
        // 检查是否达到终止条件
        if (quotaManager.isQuotaExhausted()) {
            // 设置终止标记，当前Chunk会正常提交
            stepExecution.setTerminateOnly();
        }
        
        return processed;
    }

    private ProcessedRecord doProcess(DbRecord item) {
        // 业务处理逻辑
    }
}

注意事项 #

• 这个标记是延迟生效的：只会在当前Chunk处理完、事务提交后，才会停止读取下一个Chunk，不会中断当前正在处理的记录。
• Step最终状态会被标记为COMPLETED或STOPPED，但Job不会被标记为失败，所有已处理的Chunk事务都会保留。

方案二：通过共享信号让Reader主动返回null #

如果需要让Reader立刻停止读取（而不是等当前Chunk处理完），可以通过线程安全的共享信号，让Reader在下次调用read()时返回null，触发Chunk循环正常终止。

实现步骤 #

1. 定义线程安全的终止信号类：

public class TerminationFlag {
    private final AtomicBoolean shouldStop = new AtomicBoolean(false);

    public void triggerStop() {
        shouldStop.set(true);
    }

    public boolean shouldTerminate() {
        return shouldStop.get();
    }
}

2. 在Processor中注入信号并触发终止：

public class QuotaCheckProcessor implements ItemProcessor<DbRecord, ProcessedRecord> {
    private final QuotaManager quotaManager;
    private final TerminationFlag terminationFlag;

    public QuotaCheckProcessor(QuotaManager quotaManager, TerminationFlag terminationFlag) {
        this.quotaManager = quotaManager;
        this.terminationFlag = terminationFlag;
    }

    @Override
    public ProcessedRecord process(DbRecord item) throws Exception {
        ProcessedRecord processed = doProcess(item);
        
        if (quotaManager.isQuotaExhausted()) {
            terminationFlag.triggerStop();
        }
        
        return processed;
    }
}

3. 包装JDBCReader为终止感知的Reader：

public class SignalAwareReader<T> implements ItemReader<T>, ItemStream {
    private final ItemReader<T> delegateReader;
    private final TerminationFlag terminationFlag;

    public SignalAwareReader(ItemReader<T> delegateReader, TerminationFlag terminationFlag) {
        this.delegateReader = delegateReader;
        this.terminationFlag = terminationFlag;
    }

    @Override
    public T read() throws Exception {
        // 先检查终止信号，再调用原Reader的read方法
        if (terminationFlag.shouldTerminate()) {
            return null;
        }
        return delegateReader.read();
    }

    // 实现ItemStream接口，代理原Reader的流操作（JDBCReader本身实现了ItemStream）
    @Override
    public void open(ExecutionContext executionContext) throws ItemStreamException {
        ((ItemStream) delegateReader).open(executionContext);
    }

    @Override
    public void update(ExecutionContext executionContext) throws ItemStreamException {
        ((ItemStream) delegateReader).update(executionContext);
    }

    @Override
    public void close() throws ItemStreamException {
        ((ItemStream) delegateReader).close();
    }
}

4. 在Step配置中替换原Reader：

@Bean
public ItemReader<DbRecord> jdbcReader(DataSource dataSource) {
    JdbcPagingItemReader<DbRecord> reader = new JdbcPagingItemReader<>();
    // 配置JDBCReader的SQL、映射等
    return new SignalAwareReader<>(reader, terminationFlag());
}

@Bean
public TerminationFlag terminationFlag() {
    return new TerminationFlag();
}

注意事项 #

• 这种方式会让Reader在下一次读取时直接返回null，当前Chunk的事务依然会正常提交，不会回滚。
• 适合需要立刻停止读取的场景，但要注意如果当前Chunk还有未处理的记录，依然会处理完再终止。

方案三：自定义ExitStatus控制流程（适合多Step场景） #

如果你的Job包含多个Step，需要根据终止条件决定后续Step是否执行，可以在Processor/Writer中设置自定义ExitStatus，再通过JobFlowDecision进行分支判断。

代码示例 #

public class QuotaCheckProcessor implements ItemProcessor<DbRecord, ProcessedRecord>, StepExecutionListener {
    private final QuotaManager quotaManager;

    public QuotaCheckProcessor(QuotaManager quotaManager) {
        this.quotaManager = quotaManager;
    }

    @Override
    public ExitStatus afterStep(StepExecution stepExecution) {
        if (quotaManager.isQuotaExhausted()) {
            // 返回自定义ExitStatus
            return new ExitStatus("QUOTA_FULL");
        }
        return stepExecution.getExitStatus();
    }

    @Override
    public ProcessedRecord process(DbRecord item) throws Exception {
        return doProcess(item);
    }
}

然后在Job配置中根据ExitStatus决定流程：

@Bean
public Job quotaJob(JobRepository jobRepository, Step processStep, Step notifyStep) {
    return new JobBuilder("quotaJob", jobRepository)
            .start(processStep)
            .next(new JobExecutionDecider() {
                @Override
                public FlowExecutionStatus decide(JobExecution jobExecution, StepExecution stepExecution) {
                    if ("QUOTA_FULL".equals(stepExecution.getExitStatus().getExitCode())) {
                        return new FlowExecutionStatus("QUOTA_FULL");
                    }
                    return new FlowExecutionStatus("CONTINUE");
                }
            })
            .on("QUOTA_FULL").to(notifyStep)
            .on("CONTINUE").end()
            .build();
}

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