定时任务分布式锁SchedulerLock

拉不拉稀肚拉稀  论坛元老 | 2025-4-21 13:30:24 | 显示全部楼层 | 阅读模式
打印 上一主题 下一主题
楼主

主题 1669|帖子 1669|积分 5009

马上注册，结交更多好友，享用更多功能，让你轻松玩转社区。

您需要 登录 才可以下载或查看，没有账号？立即注册

x
作用

SchedulerLock 作用：确保任务在同一时刻最多执行一次。如果一个任务正在一个节点上执行，则它将得到一个锁，该锁将阻止从另一个节点（或线程）执行同一任务。如果一个任务已经在一个节点上执行，则在其他节点上的执行不会等待，只需跳过它即可 。
SchedulerLock 主要通太过布式锁实现，可以利用：

  • 数据库锁（基于数据库行锁或唯一约束）
  • Redis 分布式锁（利用 SET NX EX）
  • Zookeeper 分布式锁（基于临时节点）
  • 基于 Quartz/ShedLock 的框架实现
相关注解

@EnableSchedulerLock
  1. @Target(ElementType.TYPE)
  2. @Retention(RetentionPolicy.RUNTIME)
  3. @Import(SchedulerLockConfigurationSelector.class)
  4. public @interface EnableSchedulerLock {
  5.     enum InterceptMode {
  6.         PROXY_SCHEDULER,
  8.         PROXY_METHOD
  9.     }
  11.     InterceptMode interceptMode() default InterceptMode.PROXY_METHOD;
  13.     String defaultLockAtMostFor();
  15.     String defaultLockAtLeastFor() default "PT0S";
  17.     AdviceMode mode() default AdviceMode.PROXY;
  19.     boolean proxyTargetClass() default false;
  21.     int order() default Ordered.LOWEST_PRECEDENCE;
  22. }
复制代码
指定在执行节点结束时应保存锁的默认时间利用 ISO8601 Duration 格式，作用就是在被加锁的节点挂了时，无法释放锁，造成其他节点无法进行下一任务，我们利用注解时间需要给定一个值。可以在每个 ScheduledLock 注解中被重写，也就是说每个定时任务都可以重新定义时间，来控制每个定时任务。

  • defaultLockAtMostFor：设定默认最大锁持有时间
  • defaultLockAtLeastFor：设定默认最小锁持有时间
@SchedulerLock
  1. @Target({ElementType.METHOD, ElementType.ANNOTATION_TYPE})
  2. @Retention(RetentionPolicy.RUNTIME)
  3. @Deprecated
  4. public @interface SchedulerLock {
  5.     String name() default "";
  7.     long lockAtMostFor() default -1L;
  9.     String lockAtMostForString() default "";
  11.     long lockAtLeastFor() default -1L;
  13.     String lockAtLeastForString() default "";
  14. }
复制代码

  • name：锁的名称，必须保证唯一，每个任务的锁名称应该唯一，因为它决定了这个锁在分布式环境中的唯一性
  • lockAtMostFor：乐成执行任务的节点所能拥有的独占锁的最长时间，设置的值要保证比定时任务正常执行完成的时间大一些，此属性保证了如果 task 节点忽然宕机，也能在超过设定值时释放任务锁
  • lockAtLeastFor：乐成执行任务的节点所能拥有的独占锁的最短时间，在指定的时间内，即使任务执行完成，锁也不会释放，这有助于防止任务被频仍触发
  • lockAtMostForString：最大时间的字符串形式，允许通过 Spring 的属性占位符（例如：${lock.duration}）来动态配置值，例如“PT14M”表示为 14 分钟
  • lockAtLeastForString：最小时间的字符串形式
根本利用

redis 整合
  1. <dependency>
  2.    <groupId>org.springframework.boot</groupId>
  3.    <artifactId>spring-boot-starter-data-redis</artifactId>
  4. </dependency>
  5. <dependency>
  6.     <groupId>net.javacrumbs.shedlock</groupId>
  7.     <artifactId>shedlock-spring</artifactId>
  8.     <version>4.38.0</version>
  9. </dependency>
  10. <dependency>
  11.     <groupId>net.javacrumbs.shedlock</groupId>
  12.     <artifactId>shedlock-provider-redis-spring</artifactId>
  13.     <version>4.38.0</version>
  14. </dependency>
复制代码
  1. spring:
  2.   redis:
  3.     #数据库索引
  4.     database: 0
  5.     host: 127.0.0.1
  6.     port: 6379
  7.     password:
  8.     jedis:
  9.       pool:
  10.         #最大连接数
  11.         max-active: 8
  12.         #最大阻塞等待时间(负数表示没限制)
  13.         max-wait: -1
  14.         #最大空闲
  15.         max-idle: 8
  16.         #最小空闲
  17.         min-idle: 0
  18.         #连接超时时间
  19.     timeout: 10000
复制代码
  1. // 开启定时任务注解
  2. @EnableScheduling
  3. // 开启定时任务锁，默认设置锁最大占用时间为 30s
  4. @EnableSchedulerLock(defaultLockAtMostFor = "PT30S")
  5. @SpringBootApplication
  6. public class HelloSpringbootApplication {
  8.    public static void main(String[] args) {
  9.       SpringApplication.run(HelloSpringbootApplication.class, args);
  10.    }
  11. }
复制代码
  1. @Configuration
  2. @EnableCaching
  3. public class RedisConfig extends CachingConfigurerSupport {
  4.    
  5.     @Bean(name = "redisTemplate")
  6.     public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory){
  7.         RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
  8.         //参照 StringRedisTemplate 内部实现指定序列化器
  9.         redisTemplate.setConnectionFactory(redisConnectionFactory);
  10.         redisTemplate.setKeySerializer(keySerializer());
  11.         redisTemplate.setHashKeySerializer(keySerializer());
  12.         redisTemplate.setValueSerializer(valueSerializer());
  13.         redisTemplate.setHashValueSerializer(valueSerializer());
  14.         return redisTemplate;
  15.     }
  17.     private RedisSerializer<String> keySerializer(){
  18.         return new StringRedisSerializer();
  19.     }
  21.     //使用 Jackson 序列化器
  22.     private RedisSerializer<Object> valueSerializer(){
  23.         return new GenericJackson2JsonRedisSerializer();
  24.     }
  25.    
  26.     @Bean
  27.     public LockProvider lockProvider(RedisTemplate redisTemplate) {
  28.         return new RedisLockProvider(redisTemplate.getConnectionFactory());
  29.     }
  30. }
复制代码
  1. @Slf4j
  2. @Component
  3. public class TestScheduled {
  5.     @Resource
  6.     RedisTemplate redisTemplate;
  8.     // @SchedulerLock 的作用是保证当前定时任务的方法执行时获得锁，忽略其他相同任务的执行
  9.     // name 必须要指定，ShedLock 就是根据这个 name 进行相同任务判定的
  10.     // name：定时任务的名字，就是数据库中的主键(name)
  11.     // lockAtMostFor：锁的最大时间单位为毫秒
  12.     // lockAtLeastFor：锁的最小时间单位为毫秒
  13.     @Scheduled(fixedDelay = 30 * 1000)
  14.     @SchedulerLock(name = "evaluateUnsubmit",lockAtLeastFor = 5*60*1000,lockAtMostFor = 20*60*1000 )
  15.     public void testMethod(){
  16.         log.info("开始执行 {}", DateFormatUtils.format(new Date(), "yyyy-MM-dd HH:mm:ss"));
  17.         try {
  18.             Thread.sleep(100);
  19.             redisTemplate.opsForValue().set("test" + System.currentTimeMillis(),"goodJob",100, TimeUnit.SECONDS);
  20.         } catch (InterruptedException e) {
  21.             throw new RuntimeException(e);
  22.         }
  23.         log.info("执行完成 {}", DateFormatUtils.format(new Date(), "yyyy-MM-dd HH:mm:ss"));
  24.     }
  26.     @Scheduled(cron = "*/15 * * * * *")
  27.     @SchedulerLock(name = "TaskScheduler_scheduledTask", lockAtLeastForString = "PT5M", lockAtMostForString = "PT14M")
  28.     public void scheduledTask() {
  29.         // ...
  30.     }
  31. }
复制代码
mysql 整合
  1. <dependency>
  2.     <groupId>net.javacrumbs.shedlock</groupId>
  3.     <artifactId>shedlock-spring</artifactId>
  4.     <version>4.23.0</version>
  5. </dependency>
  7. <dependency>
  8.     <groupId>net.javacrumbs.shedlock</groupId>
  9.     <artifactId>shedlock-provider-jdbc-template</artifactId>
  10.     <version>4.23.0</version>
  11. </dependency>
  12. <dependency>
  13.     <groupId>mysql</groupId>
  14.     <artifactId>mysql-connector-java</artifactId>
  15.     <scope>runtime</scope>
  16. </dependency>
复制代码
  1. # MySQL, MariaDB
  2. CREATE TABLE shedlock(name VARCHAR(64) NOT NULL, lock_until TIMESTAMP(3) NOT NULL,
  3.     locked_at TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), locked_by VARCHAR(255) NOT NULL, PRIMARY KEY (name));
  5. # Postgres
  6. CREATE TABLE shedlock(name VARCHAR(64) NOT NULL, lock_until TIMESTAMP NOT NULL,
  7.     locked_at TIMESTAMP NOT NULL, locked_by VARCHAR(255) NOT NULL, PRIMARY KEY (name));
  9. # Oracle
  10. CREATE TABLE shedlock(name VARCHAR(64) NOT NULL, lock_until TIMESTAMP(3) NOT NULL,
  11.     locked_at TIMESTAMP(3) NOT NULL, locked_by VARCHAR(255) NOT NULL, PRIMARY KEY (name));
  13. # MS SQL
  14. CREATE TABLE shedlock(name VARCHAR(64) NOT NULL, lock_until datetime2 NOT NULL,
  15.     locked_at datetime2 NOT NULL, locked_by VARCHAR(255) NOT NULL, PRIMARY KEY (name));
  17. # DB2
  18. CREATE TABLE shedlock(name VARCHAR(64) NOT NULL PRIMARY KEY, lock_until TIMESTAMP NOT NULL,
  19.     locked_at TIMESTAMP NOT NULL, locked_by VARCHAR(255) NOT NULL);
复制代码
  1. @Configuration
  2. // 开启定时器
  3. @EnableScheduling
  4. // 开启定时任务锁，指定一个默认的锁的时间 30 秒
  5. @EnableSchedulerLock(defaultLockAtMostFor = "PT30S")
  6. public class ShedlockJdbcConfig {
  8.     /**
  9.      * 配置锁的提供者
  10.      */
  11.     @Bean
  12.     public LockProvider lockProvider(DataSource dataSource) {
  13.         return new JdbcTemplateLockProvider(
  14.                 JdbcTemplateLockProvider.Configuration.builder()
  15.                         .withJdbcTemplate(new JdbcTemplate(dataSource))
  16.                         .usingDbTime()
  17.                         .build()
  18.         );
  19.     }
  20. }
复制代码
  1. @Component
  2. @Slf4j
  3. public class TimeTaskJob {
  5.     private static Integer count = 1;
  7.     /**
  8.      * 任务 1 每 5 秒执行一次
  9.      * lockAtLeastFor：虽然定时任务是每隔5秒执行一次, 但是分布式锁定义的是: 每次任务要锁住20秒，20秒是持有锁的最小时间，必须等20秒后才释放锁，并且确保在20秒钟内，该任务不会运行超过 1 次；
  10.      * lockAtMostFor：锁最大持有时间30秒，表示最多锁定30秒钟，主要用于防止执行任务的节点挂掉（即使这个节点挂掉，在30秒钟后锁也被释放），一般将其设置为明显大于任务的最大执行时长；如果任务运行时间超过该值（即任务30秒钟没有执行完），则该任务可能被重复执行。
  11.      */
  12.     @Scheduled(cron = "0/5 * * * * ? ")
  13.     @SchedulerLock(name = "testJob1",lockAtLeastFor = "20000", lockAtMostFor = "30000")
  14.     public void scheduledTask1() {
  15.         log.info(Thread.currentThread().getName() + "->>>任务1执行第：" + (count++) + "次");
  16.     }
  19.     @Scheduled(cron = "0/5 * * * * ? ")
  20.     @SchedulerLock(name = "testJob2")
  21.     public void scheduledTask2() {
  22.         log.info(Thread.currentThread().getName() + "->>>任务2执行第：" + (count++) + "次");
  23.     }
  24. }
复制代码
实现原理


  • 利用@EnableSchedulerLock 注解后，会引入 SchedulerLockConfigurationSelector 类，根据其对应的模式（默认 InterceptMode.PROXY_METHOD）天生 LockConfigurationExtractorConfiguration 和 MethodProxyLockConfiguration 类
  1. @Target(ElementType.TYPE)
  2. @Retention(RetentionPolicy.RUNTIME)
  3. @Import(SchedulerLockConfigurationSelector.class)
  4. public @interface EnableSchedulerLock {
  5.     enum InterceptMode {
  6.         PROXY_SCHEDULER,
  8.         PROXY_METHOD
  9.     }
  11.     InterceptMode interceptMode() default InterceptMode.PROXY_METHOD;
  13.     String defaultLockAtMostFor();
  15.     String defaultLockAtLeastFor() default "PT0S";
  17.     AdviceMode mode() default AdviceMode.PROXY;
  19.     boolean proxyTargetClass() default false;
  21.     int order() default Ordered.LOWEST_PRECEDENCE;
  22. }
复制代码
  1. public class SchedulerLockConfigurationSelector implements ImportSelector {
  3.     @Override
  4.     @NonNull
  5.     public String[] selectImports(@NonNull AnnotationMetadata metadata) {
  6.         AnnotationAttributes attributes = AnnotationAttributes.fromMap(metadata.getAnnotationAttributes(EnableSchedulerLock.class.getName(), false));
  7.         InterceptMode mode = attributes.getEnum("interceptMode");
  8.         if (mode == PROXY_METHOD) {
  9.             return new String[]{AutoProxyRegistrar.class.getName(), LockConfigurationExtractorConfiguration.class.getName(), MethodProxyLockConfiguration.class.getName()};
  10.         } else if (mode == PROXY_SCHEDULER) {
  11.             return new String[]{AutoProxyRegistrar.class.getName(), LockConfigurationExtractorConfiguration.class.getName(), SchedulerProxyLockConfiguration.class.getName(), RegisterDefaultTaskSchedulerPostProcessor.class.getName()};
  12.         } else {
  13.             throw new UnsupportedOperationException("Unknown mode " + mode);
  14.         }
  15.     }
  16. }
复制代码

  • LockConfigurationExtractorConfiguration 会获取@EnableSchedulerLock 注解上的属性进行配置，天生 SpringLockConfigurationExtractor
  1. @Configuration
  2. class LockConfigurationExtractorConfiguration extends AbstractLockConfiguration implements EmbeddedValueResolverAware {
  3.     private final StringToDurationConverter durationConverter = StringToDurationConverter.INSTANCE;
  5.     private StringValueResolver resolver;
  7.     @Bean
  8.     ExtendedLockConfigurationExtractor lockConfigurationExtractor() {
  9.         return new SpringLockConfigurationExtractor(defaultLockAtMostForDuration(), defaultLockAtLeastForDuration(), resolver, durationConverter);
  10.     }
  12.     private Duration defaultLockAtLeastForDuration() {
  13.         return toDuration(getDefaultLockAtLeastFor());
  14.     }
  16.     private Duration defaultLockAtMostForDuration() {
  17.         return toDuration(getDefaultLockAtMostFor());
  18.     }
  20.     private String getDefaultLockAtLeastFor() {
  21.         return getStringFromAnnotation("defaultLockAtLeastFor");
  22.     }
  24.     private String getDefaultLockAtMostFor() {
  25.         return getStringFromAnnotation("defaultLockAtMostFor");
  26.     }
  28.     private Duration toDuration(String string) {
  29.         return durationConverter.convert(resolver.resolveStringValue(string));
  30.     }
  32.     protected String getStringFromAnnotation(String name) {
  33.         return annotationAttributes.getString(name);
  34.     }
  36.     @Override
  37.     public void setEmbeddedValueResolver(@NonNull StringValueResolver resolver) {
  38.         this.resolver = resolver;
  39.     }
  40. }
复制代码

  • MethodProxyLockConfiguration 类会根据 LockProvider 和 ExtendedLockConfigurationExtractor 进行自动装配，天生 MethodProxyScheduledLockAdvisor
  1. @Configuration
  2. @Role(BeanDefinition.ROLE_INFRASTRUCTURE)
  3. class MethodProxyLockConfiguration extends AbstractLockConfiguration {
  4.     @Bean
  5.     @Role(BeanDefinition.ROLE_INFRASTRUCTURE)
  6.     MethodProxyScheduledLockAdvisor proxyScheduledLockAopBeanPostProcessor(
  7.         @Lazy LockProvider lockProvider,
  8.         @Lazy ExtendedLockConfigurationExtractor lockConfigurationExtractor
  9.     ) {
  10.         MethodProxyScheduledLockAdvisor advisor = new MethodProxyScheduledLockAdvisor(
  11.             lockConfigurationExtractor,
  12.             new DefaultLockingTaskExecutor(lockProvider)
  13.         );
  14.         advisor.setOrder(getOrder());
  15.         return advisor;
  16.     }
  17. }
复制代码

  • 天生一个切面 MethodProxyScheduledLockAdvisor 类，对方法进行拦截
  1. class MethodProxyScheduledLockAdvisor extends AbstractPointcutAdvisor {
  2.     // ...
  4.     private static class LockingInterceptor implements MethodInterceptor {
  5.         private final ExtendedLockConfigurationExtractor lockConfigurationExtractor;
  6.         private final LockingTaskExecutor lockingTaskExecutor;
  8.         LockingInterceptor(ExtendedLockConfigurationExtractor lockConfigurationExtractor, LockingTaskExecutor lockingTaskExecutor) {
  9.             this.lockConfigurationExtractor = lockConfigurationExtractor;
  10.             this.lockingTaskExecutor = lockingTaskExecutor;
  11.         }
  13.         @Override
  14.         public Object invoke(MethodInvocation invocation) throws Throwable {
  15.             Class<?> returnType = invocation.getMethod().getReturnType();
  16.             if (returnType.isPrimitive() && !void.class.equals(returnType)) {
  17.                 throw new LockingNotSupportedException("Can not lock method returning primitive value");
  18.             }
  20.             // 查找@SchedulerLock 注解
  21.             LockConfiguration lockConfiguration = lockConfigurationExtractor.getLockConfiguration(invocation.getThis(), invocation.getMethod()).get();
  22.             // 执行加锁方法
  23.             TaskResult<Object> result = lockingTaskExecutor.executeWithLock(invocation::proceed, lockConfiguration);
  25.             if (Optional.class.equals(returnType)) {
  26.                 return toOptional(result);
  27.             } else {
  28.                 return result.getResult();
  29.             }
  30.         }
  32.         private static Object toOptional(TaskResult<Object> result) {
  33.             if (result.wasExecuted()) {
  34.                 return result.getResult();
  35.             } else {
  36.                 return Optional.empty();
  37.             }
  38.         }
  39.     }
  40. }
复制代码

  • SpringLockConfigurationExtractor 会查找方法上是否存在@SchedulerLock 注解
  1. class SpringLockConfigurationExtractor implements ExtendedLockConfigurationExtractor {
  2.     // ...
  3.    
  4.     @Override
  5.     public Optional<LockConfiguration> getLockConfiguration(Object target, Method method) {
  6.         AnnotationData annotation = findAnnotation(target, method);
  7.         if (shouldLock(annotation)) {
  8.             return Optional.of(getLockConfiguration(annotation));
  9.         } else {
  10.             return Optional.empty();
  11.         }
  12.     }
  14.     AnnotationData findAnnotation(Object target, Method method) {
  15.         AnnotationData annotation = findAnnotation(method);
  16.         if (annotation != null) {
  17.             return annotation;
  18.         } else {
  19.             Class<?> targetClass = AopUtils.getTargetClass(target);
  20.             try {
  21.                 Method methodOnTarget = targetClass
  22.                     .getMethod(method.getName(), method.getParameterTypes());
  23.                 return findAnnotation(methodOnTarget);
  24.             } catch (NoSuchMethodException e) {
  25.                 return null;
  26.             }
  27.         }
  28.     }
  30.     private AnnotationData findAnnotation(Method method) {
  31.         net.javacrumbs.shedlock.core.SchedulerLock annotation = AnnotatedElementUtils.getMergedAnnotation(method, net.javacrumbs.shedlock.core.SchedulerLock.class);
  32.         if (annotation != null) {
  33.             return new AnnotationData(annotation.name(), annotation.lockAtMostFor(), annotation.lockAtMostForString(), annotation.lockAtLeastFor(), annotation.lockAtLeastForString());
  34.         }
  35.         SchedulerLock annotation2 = AnnotatedElementUtils.getMergedAnnotation(method, SchedulerLock.class);
  36.         if (annotation2 != null) {
  37.             return new AnnotationData(annotation2.name(), -1, annotation2.lockAtMostFor(), -1, annotation2.lockAtLeastFor());
  38.         }
  39.         return null;
  40.     }
  42.     // ...
  43. }
复制代码

  • DefaultLockingTaskExecutor 类对方法进行加解锁，执行 LockProvider 提供的加锁方法
  1. public class DefaultLockingTaskExecutor implements LockingTaskExecutor {
  2.     // ...
  3.    
  4.     @Override
  5.     @NonNull
  6.     public <T> TaskResult<T> executeWithLock(@NonNull TaskWithResult<T> task, @NonNull LockConfiguration lockConfig) throws Throwable {
  7.         Optional<SimpleLock> lock = lockProvider.lock(lockConfig);
  8.         String lockName = lockConfig.getName();
  10.         if (alreadyLockedBy(lockName)) {
  11.             logger.debug("Already locked '{}'", lockName);
  12.             return TaskResult.result(task.call());
  13.         } else if (lock.isPresent()) {
  14.             try {
  15.                 LockAssert.startLock(lockName);
  16.                 LockExtender.startLock(lock.get());
  17.                 logger.debug("Locked '{}', lock will be held at most until {}", lockName, lockConfig.getLockAtMostUntil());
  18.                 return TaskResult.result(task.call());
  19.             } finally {
  20.                 LockAssert.endLock();
  21.                 SimpleLock activeLock = LockExtender.endLock();
  22.                 if (activeLock != null) {
  23.                     activeLock.unlock();
  24.                 } else {
  25.                     // This should never happen, but I do not know any better way to handle the null case.
  26.                     logger.warn("No active lock, please report this as a bug.");
  27.                     lock.get().unlock();
  28.                 }
  29.                 if (logger.isDebugEnabled()) {
  30.                     Instant lockAtLeastUntil = lockConfig.getLockAtLeastUntil();
  31.                     Instant now = ClockProvider.now();
  32.                     if (lockAtLeastUntil.isAfter(now)) {
  33.                         logger.debug("Task finished, lock '{}' will be released at {}", lockName, lockAtLeastUntil);
  34.                     } else {
  35.                         logger.debug("Task finished, lock '{}' released", lockName);
  36.                     }
  37.                 }
  38.             }
  39.         } else {
  40.             logger.debug("Not executing '{}'. It's locked.", lockName);
  41.             return TaskResult.notExecuted();
  42.         }
  43.     }
  44. }
复制代码
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。
回复

使用道具 举报

0 个回复

倒序浏览
返回列表

快速回复

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 or 立即注册

本版积分规则

拉不拉稀肚拉稀

论坛元老
这个人很懒什么都没写!

楼主热帖

标签云

集成商 AI 运维 CIO 存储 服务器
微信订阅号
微信服务号
微信客服
小程序
H5
关于我们 商务合作 网站地图
©Copyright 2022-2024 杭州祥升科技有限公司 版权所有 浙ICP备20004199号
快速回复 返回顶部 返回列表