Spring Batch 5.0.0中如何重启多租户已完成定时任务

核心思路 #

解决多进程下的任务重启冲突，核心是基于租户维度的分布式锁+任务执行状态监听，确保每个租户的任务完成后，只有一个进程能触发重启动作，同时复用固定的租户参数，避免引入动态参数导致多进程重复执行。

具体实现方案 #

1. 自定义JobExecutionListener监听任务完成事件 #

通过实现Spring Batch的JobExecutionListener，在任务执行完成后触发重启逻辑，重启前先获取对应租户的分布式锁，避免多进程同时触发：

@Component
public class TenantJobRestartListener implements JobExecutionListener {

    private final JobLauncher jobLauncher;
    private final Job inferenceJob;
    private final DistributedLockProvider distributedLockProvider;

    public TenantJobRestartListener(JobLauncher jobLauncher, Job inferenceJob, DistributedLockProvider distributedLockProvider) {
        this.jobLauncher = jobLauncher;
        this.inferenceJob = inferenceJob;
        this.distributedLockProvider = distributedLockProvider;
    }

    @Override
    public void afterJob(JobExecution jobExecution) {
        String tenantId = jobExecution.getJobParameters().getString(SchedulerConstants.TENANTID_CONTEXT);
        if (tenantId == null) {
            return;
        }

        // 以租户ID作为锁标识，确保同一租户只有一个进程能触发重启
        String lockKey = "batch-job-restart-lock:" + tenantId;
        try (DistributedLock lock = distributedLockProvider.acquireLock(lockKey, Duration.ofSeconds(30))) {
            if (lock.isAcquired()) {
                // 复用固定参数，避免动态参数引发多进程重复执行
                JobParameters jobParameters = new JobParametersBuilder()
                        .addLong("run.id", 1L, false)
                        .addString(SchedulerConstants.TENANTID_CONTEXT, tenantId)
                        .toJobParameters();
                jobLauncher.run(inferenceJob, jobParameters);
            }
        } catch (Exception e) {
            // 记录日志或告警，处理锁获取失败、任务启动异常等情况
        }
    }
}

2. 实现分布式锁组件 #

基于Redis（以Redisson为例）实现分布式锁，保证同一租户的重启动作唯一执行：

@Component
public class RedisDistributedLockProvider implements DistributedLockProvider {

    private final RedissonClient redissonClient;

    public RedisDistributedLockProvider(RedissonClient redissonClient) {
        this.redissonClient = redissonClient;
    }

    @Override
    public DistributedLock acquireLock(String lockKey, Duration leaseTime) {
        RLock rLock = redissonClient.getLock(lockKey);
        boolean acquired = false;
        try {
            acquired = rLock.tryLock(0, leaseTime.toMillis(), TimeUnit.MILLISECONDS);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        return new RedisDistributedLock(rLock, acquired);
    }

    private static class RedisDistributedLock implements DistributedLock {
        private final RLock rLock;
        private final boolean isAcquired;

        public RedisDistributedLock(RLock rLock, boolean isAcquired) {
            this.rLock = rLock;
            this.isAcquired = isAcquired;
        }

        @Override
        public boolean isAcquired() {
            return isAcquired;
        }

        @Override
        public void close() {
            if (isAcquired && rLock.isHeldByCurrentThread()) {
                rLock.unlock();
            }
        }
    }
}

interface DistributedLockProvider {
    DistributedLock acquireLock(String lockKey, Duration leaseTime);
}

interface DistributedLock extends AutoCloseable {
    boolean isAcquired();
}

3. 为任务绑定监听器 #

在定义任务Bean时，将自定义监听器绑定到任务上：

@Bean
public Job inferenceJob(JobBuilderFactory jobBuilderFactory, Step inferenceStep, TenantJobRestartListener restartListener) {
    return jobBuilderFactory.get("inferenceJob")
            .listener(restartListener)
            .start(inferenceStep)
            .build();
}

关键注意事项 #

• 锁粒度控制：必须以租户ID作为锁的Key，确保不同租户的任务重启互不干扰。
• 锁有效期设置：设置合理的锁过期时间，避免进程异常导致锁无法释放，阻塞后续重启动作。
• 任务参数复用：保持run.id和租户ID固定，避免Spring Batch将重启判定为新任务；若需每次重启作为新任务执行，可将run.id的第三个参数改为true，结合分布式锁仍能避免多进程重复执行。
• 异常兜底处理：在监听器中捕获所有异常，避免单次重启失败影响后续任务触发。

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