在Azure AKS上扩展Spring Batch时避免作业重复执行的方案咨询

一、作业重复执行的核心原因排查 #

• 作业实例幂等性缺失：Spring Batch通过JobParameters区分作业实例，若提交时未加入唯一标识（如时间戳、批次ID），多Pod提交时可能被判定为同一实例，且提交逻辑未做状态校验，导致重复启动。
• Pod启动时自动提交作业：若应用启动脚本/初始化逻辑包含作业提交代码，HPA扩容新Pod时，每个Pod都会执行这段逻辑，触发重复提交。
• JobRepository事务隔离问题：默认配置的事务隔离级别在高并发场景下，可能出现多Pod同时读取作业状态时的一致性问题，导致重复启动。
• 分布式调度冲突：使用Spring Scheduler等定时触发逻辑时，多Pod同时执行定时任务，未做分布式锁控制，引发重复提交。

二、同步管理解决方案 #

1. 确保作业实例的幂等性 #

每次提交作业时加入唯一JobParameter，并提交前检查实例状态：

JobParameters jobParams = new JobParametersBuilder()
    .addString("reportType", "daily")
    .addLong("timestamp", System.currentTimeMillis()) // 唯一标识
    .toJobParameters();

// 检查是否已有相同参数的作业实例
List<JobInstance> existingInstances = jobExplorer.getJobInstances("dailyReportJob", 0, 1);
if (existingInstances.isEmpty() || !Objects.equals(existingInstances.get(0).getJobParameters(), jobParams)) {
    jobLauncher.run(dailyReportJob, jobParams);
}

2. 剥离Pod启动时的作业提交逻辑 #

将作业提交与应用启动解耦，改用外部触发机制：

• 用Kubernetes CronJob负责定时提交作业，避免多Pod的调度冲突。
• 通过Kubernetes API或Azure Event Grid接收触发信号，仅在收到信号时执行提交逻辑。

3. 调整JobRepository事务隔离级别 #

配置更高的隔离级别，避免高并发下的读不一致：

@Bean
public JobRepository jobRepository(DataSource dataSource, PlatformTransactionManager transactionManager) throws Exception {
    JobRepositoryFactoryBean factory = new JobRepositoryFactoryBean();
    factory.setDataSource(dataSource);
    factory.setTransactionManager(transactionManager);
    factory.setIsolationLevelForCreate("ISOLATION_REPEATABLE_READ");
    factory.setTablePrefix("BATCH_");
    return factory.getObject();
}

4. 引入分布式锁控制提交 #

若需保留多Pod提交能力，用分布式锁（如Azure Redis Cache）确保同一时间仅一个Pod能提交作业：

@Autowired
private StringRedisTemplate redisTemplate;

public void submitReportJob(JobParameters jobParams) {
    String lockKey = "daily-report-job-lock";
    Boolean locked = redisTemplate.opsForValue().setIfAbsent(lockKey, "locked", Duration.ofMinutes(5));
    if (Boolean.TRUE.equals(locked)) {
        try {
            jobLauncher.run(dailyReportJob, jobParams);
        } finally {
            redisTemplate.delete(lockKey);
        }
    }
}

三、合理配置HPA与扩缩容策略 #

1. 基于作业队列长度配置HPA #

脱离单纯的CPU/内存指标，结合待处理作业队列长度作为自定义扩容指标：

• 用Azure Service Bus或Redis List维护待处理作业队列，实时统计队列长度。
• 配置AKS HPA监听该自定义指标，队列长度超过阈值时扩容，任务完成后自动缩容。

2. 限制并发作业执行数量 #

通过Spring Batch的任务执行器配置，限制单Pod及集群内的并发作业数：

@Bean
public TaskExecutor taskExecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(2);
    executor.setMaxPoolSize(5); // 单Pod最大并发数
    executor.setQueueCapacity(10);
    executor.setThreadNamePrefix("batch-exec-");
    executor.initialize();
    return executor;
}

@Bean
public JobLauncher jobLauncher(JobRepository jobRepository, TaskExecutor taskExecutor) throws Exception {
    SimpleJobLauncher launcher = new SimpleJobLauncher();
    launcher.setJobRepository(jobRepository);
    launcher.setTaskExecutor(taskExecutor);
    return launcher;
}

3. 配置Pod优雅关闭 #

在AKS Deployment中设置优雅关闭，确保缩容时Pod能完成当前作业再退出：

apiVersion: apps/v1
kind: Deployment
metadata:
  name: batch-report-app
spec:
  template:
    spec:
      containers:
      - name: batch-app
        image: your-image:tag
        lifecycle:
          preStop:
            exec:
              command: ["sh", "-c", "sleep 30"] # 预留作业完成时间
        terminationGracePeriodSeconds: 60

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