使用分布式锁实现定时任务的精确调度

在分布式系统中，实现定时任务的精确调度是一项具有挑战性的任务。由于分布式环境中存在多个节点，传统的定时任务可能会出现并发执行、重复执行或者错过执行的问题。为了解决这些问题，我们可以使用分布式锁来实现定时任务的精确调度。

准备工作

在开始之前，我们需要准备以下环境和工具：

Spring框架
Redis作为分布式锁的存储介质

实现步骤

1. 创建自定义注解

首先，我们需要创建一个自定义注解，用于标记需要进行定时任务调度的方法。这个注解可以包含一些属性，用于配置定时任务的执行规则和其他参数。例如，我们可以添加cron属性来指定定时任务的执行时间表达式，以及duration属性来指定预期任务时长。此注解集成Scheduled，使其天然继承Scheduled的特性，使我们不再关系定时触发的细节。

import org.springframework.core.annotation.AliasFor;
import org.springframework.scheduling.annotation.Scheduled;

import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;

@Target({ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
@Scheduled
public @interface DistributedScheduled {

    @AliasFor(annotation = Scheduled.class, attribute = "cron")
    String cron() default "";

    /**
     * 任务持续时长
     */
    String duration() default "";

}

2. 创建切面类

接下来，我们需要创建一个切面类，用于拦截带有DistributedScheduled注解的方法，并实现定时任务的精确调度逻辑。在切面类中，我们可以使用Redis作为分布式锁的存储介质，并使用AOP技术来拦截目标方法。

import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Aspect
@Component
public class DistributedScheduledAspect {

    @Autowired
    private RedisFatClient redisFatClient;

    @Around("@annotation(distributedScheduled)")
    public Object executeScheduledTask(ProceedingJoinPoint joinPoint, DistributedScheduled distributedScheduled) {
        String lockKey = joinPoint.getSignature().toLongString().replace(" ", "-");
        RLock lock = null;
        try {
            String duration = distributedScheduled.duration();
            long lockDuration = 30 * 1000;
            if (!duration.isEmpty()) {
                lockDuration = Long.parseLong(duration);
            } 
            lock = redisFatClient.getLock(lockDuration, lockKey);
        
            if (lock.tryLock()) {
                return joinPoint.proceed();
            } else {
                log.warn("lockKey: {}, 抢锁失败，此任务本节点不执行", lockKey);
            }
        } catch (Throwable throwable) {
            log.error("执行定时任务发生异常", throwable);
        } finally {
            if (lock != null && lock.isHeldByCurrentThread()) {
                lock.unlock();
            }
        }
        return null;
    }
}

3. 使用自定义注解标记定时任务方法

现在，我们可以在需要进行定时任务调度的方法上使用DistributedScheduled注解来标记。可以根据需要配置cron和duration属性。

@Slf4j
@Component
public class MyScheduledTasks {

    @DistributedScheduled(cron = "0 * * * * ?", duration = "60000")
    public void myScheduledTask() {
        // 执行定时任务的逻辑
    }
}