如何确保两个Spring Batch作业完成后执行后续操作？

问题根源 #

你遇到的问题核心在于JobExecutionListener.afterJob()的执行时机：作业在内存中执行完毕就会触发该方法，但此时作业的最终状态还未持久化到JobRepository。消费者通过JobExplorer查询状态时，可能拿到未更新的旧状态，导致判断逻辑失效。

最佳实践方案 #

方案1：用Spring Batch作业流编排（最优选择） #

如果两个作业可以编排为一个整体流程，直接通过Spring Batch的作业流控制逻辑，确保两个作业都成功后再执行后续业务代码：

@Configuration
class CombinedJobConfiguration(
    private val jobBuilderFactory: JobBuilderFactory,
    private val stepBuilderFactory: StepBuilderFactory,
    private val aJob: Job, // 注入已定义的AJob
    private val bJob: Job  // 注入已定义的BJob
) {

    @Bean
    fun combinedJob(): Job {
        return jobBuilderFactory.get("combinedJob")
            .start(aJob)
            .on("COMPLETED").to(bJob)
            .from(bJob)
            .on("COMPLETED").to(finalBusinessStep())
            .end()
            .build()
    }

    private fun finalBusinessStep(): Step {
        return stepBuilderFactory.get("finalBusinessStep")
            .tasklet { _, _ ->
                // 在这里执行两个作业成功后的业务逻辑
                ExitStatus.COMPLETED
            }
            .build()
    }
}

这种方式完全由Spring Batch的流程控制保证状态可靠性，无需额外消息中间件，逻辑最稳定。

方案2：确保状态持久化后再发消息 #

自定义监听器，利用Spring事务同步机制，在JobRepository完成状态更新后再发送Kafka消息：

@Configuration
class AJobConfiguration {
    @Bean
    @JobScope
    fun jobCompletionListener(kafkaTemplate: KafkaTemplate<String, String>): JobExecutionListener {
        return object : JobExecutionListenerSupport() {
            override fun afterJob(jobExecution: JobExecution) {
                // 绑定到作业执行的事务，确保状态持久化后再发送消息
                TransactionSynchronizationManager.registerSynchronization(object : TransactionSynchronizationAdapter() {
                    override fun afterCommit() {
                        // 此时JobRepository已完成状态更新
                        kafkaTemplate.send("job-completion-topic", "AJob completed")
                    }
                })
            }
        }
    }
}

给BJob配置相同逻辑后，消费者收到消息时，作业状态已经可靠写入数据库，查询结果不会出错。

方案3：消费者侧轮询重试（兼容方案） #

如果作业必须独立执行，可在消费者侧增加轮询逻辑，给JobRepository足够的状态更新时间：

@KafkaListener(topics = ["job-completion-topic"])
fun finishJobs() {
    val maxRetries = 5
    var retryCount = 0
    var bothCompleted = false

    while (retryCount < maxRetries && !bothCompleted) {
        Thread.sleep(2000) // 每次间隔2秒重试
        val aExecution = jobExplorer.getLastJobExecution("AJob")
        val bExecution = jobExplorer.getLastJobExecution("BJob")
        
        bothCompleted = aExecution?.exitStatus?.equals(ExitStatus.COMPLETED) == true 
            && bExecution?.exitStatus?.equals(ExitStatus.COMPLETED) == true
            
        retryCount++
    }

    if (bothCompleted) {
        // 执行目标业务逻辑
    } else {
        // 处理超时情况，比如记录告警
    }
}

总结 #

• 若作业可编排成流程，方案1是最优选择，无额外依赖，可靠性最高；
• 若作业需独立运行，优先选方案2，从根源保证消息发送时机正确；
• 方案3作为临时兼容方案，适合无法修改作业监听器的场景。

内容的提问来源于stack exchange，提问作者jeongmin.cha