当前位置 : 主页 > 编程语言 > java >

Schedule定时任务在分布式产生的问题详解

来源:互联网 收集:自由互联 发布时间:2023-01-30
目录 正文 一、搭建基本环境 二、问题::执行时间延迟和单线程执行 三、为什么会出现上述问题? 四、解决方式 4.1、修改配置文件 4.2、执行逻辑改为异步执行 4.3、异步定时任务 4.
目录
  • 正文
  • 一、搭建基本环境
  • 二、问题::执行时间延迟和单线程执行
  • 三、为什么会出现上述问题?
  • 四、解决方式
    • 4.1、修改配置文件
    • 4.2、执行逻辑改为异步执行
    • 4.3、异步定时任务
    • 4.4、小结
  • 五、分布式下的思考
    • 思考:并发执行
    • 解决方式:分布式锁
  • 后记

    正文

    定时任务的实现方式多种多样,框架也是层出不穷。

    本文所谈及的是 SpringBoot 本身所带有的@EnableScheduling 、 @Scheduled实现定时任务的方式。

    以及采用这种方式,在分布式调度中可能会出现的问题,又针对为什么会发生这种问题?又该如何解决,做出了一些叙述。

    为了适合每个阶段的读者,我把前面测试的代码都贴出来啦~

    确保每一步都是有迹可循的,希望大家不要嫌啰嗦,感谢

    一、搭建基本环境

    基本依赖

     <parent>
         <artifactId>spring-boot-parent</artifactId>
         <groupId>org.springframework.boot</groupId>
         <version>2.7.2</version>
     </parent>
     <dependencies>
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot</artifactId>
         </dependency>
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter</artifactId>
         </dependency>
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-web</artifactId>
         </dependency>
         <dependency>
             <dependency>
                 <groupId>org.projectlombok</groupId>
                 <artifactId>lombok</artifactId>
             </dependency>
      </dependencies>
    

    创建个启动类及定时任务

     @SpringBootApplication
     public class ApplicationScheduling {
         public static void main(String[] args) {
             SpringApplication.run(ApplicationScheduling.class, args);
         }
     }
    
     /**
      * @description:
      * @author: Ning Zaichun
      * @date: 2022年09月06日 0:02
      */
     @Slf4j
     @Component
     @EnableScheduling
     public class ScheduleService {
         // 每五秒执行一次,cron的表达式就不再多说明了
         @Scheduled(cron = "0/5 * * * * ? ")
         public void testSchedule() {
                 log.info("当前执行任务的线程号ID===>{}", Thread.currentThread().getId());
         }
     }
    

    二、问题::执行时间延迟和单线程执行

    按照上面代码中给定的cron表达式@Scheduled(cron = "0/5 * * * * ? ")每五秒执行一次,那么最近五次的执行结果应当为:

     2022-09-06 00:21:10
     2022-09-06 00:21:15
     2022-09-06 00:21:20
     2022-09-06 00:21:25
     2022-09-06 00:21:30

    如果定时任务中是执行非常快的任务的,时间非常非常短,确实不会有什么的延迟性。

    上面代码执行结果:

     2022-09-06 19:42:10.018  INFO 24496 --- [   scheduling-1] com.nzc.service.ScheduleService          : 当前执行任务的线程号ID===>64
     2022-09-06 19:42:15.015  INFO 24496 --- [   scheduling-1] com.nzc.service.ScheduleService          : 当前执行任务的线程号ID===>64
     2022-09-06 19:42:20.001  INFO 24496 --- [   scheduling-1] com.nzc.service.ScheduleService          : 当前执行任务的线程号ID===>64
     2022-09-06 19:42:25.005  INFO 24496 --- [   scheduling-1] com.nzc.service.ScheduleService          : 当前执行任务的线程号ID===>64
     2022-09-06 19:42:30.007  INFO 24496 --- [   scheduling-1] com.nzc.service.ScheduleService          : 当前执行任务的线程号ID===>64

    如果说从时间上来看,说不上什么延迟性,但真实的业务场景中,业务的执行时间可能远比这里时间长。

    我主动让线程睡上10秒,让我们再来看看输出结果是如何的吧

         @Scheduled(cron = "0/5 * * * * ? ")
         public void testSchedule() {
             try {
                 Thread.sleep(10000);
                 log.info("当前执行任务的线程号ID===>{}", Thread.currentThread().getId());
             } catch (Exception e) {
                 e.printStackTrace();
             } 
         }
    

    输出结果

     2022-09-06 19:46:50.019  INFO 27236 --- [   scheduling-1] com.nzc.service.ScheduleService          : 当前执行任务的线程号ID===>64
     2022-09-06 19:47:05.024  INFO 27236 --- [   scheduling-1] com.nzc.service.ScheduleService          : 当前执行任务的线程号ID===>64
     2022-09-06 19:47:20.016  INFO 27236 --- [   scheduling-1] com.nzc.service.ScheduleService          : 当前执行任务的线程号ID===>64
     2022-09-06 19:47:35.005  INFO 27236 --- [   scheduling-1] com.nzc.service.ScheduleService          : 当前执行任务的线程号ID===>64
     2022-09-06 19:47:50.006  INFO 27236 --- [   scheduling-1] com.nzc.service.ScheduleService          : 当前执行任务的线程号ID===>64

    请注意两个问题:

    • 执行时间延迟:从时间上可以明显看出,不再是每五秒执行一次,执行时间延迟很多,造成任务的
    • 单线程执行:从始至终都只有一个线程在执行任务,造成任务的堵塞.

    三、为什么会出现上述问题?

    问题的根本:线程阻塞式执行,执行任务线程数量过少。

    那到底是为什么呢?

    回到启动类上,我们在启动上标明了一个@EnableScheduling注解。

    大家在看到诸如@Enablexxxx这样的注解的时候,就要知道它一定有一个xxxxxAutoConfiguration的自动装配的类。

    @EnableScheduling也不例外,它的自动装配的类是TaskSchedulingAutoConfiguration

    我们来看看它到底做了一些什么设置?我们如何修改?

     @ConditionalOnClass(ThreadPoolTaskScheduler.class)
     @Configuration(proxyBeanMethods = false)
     @EnableConfigurationProperties(TaskSchedulingProperties.class)
     @AutoConfigureAfter(TaskExecutionAutoConfiguration.class)
     public class TaskSchedulingAutoConfiguration {
         @Bean
         @ConditionalOnBean(name = TaskManagementConfigUtils.SCHEDULED_ANNOTATION_PROCESSOR_BEAN_NAME)
         @ConditionalOnMissingBean({ SchedulingConfigurer.class, TaskScheduler.class, ScheduledExecutorService.class })
         public ThreadPoolTaskScheduler taskScheduler(TaskSchedulerBuilder builder) {
             return builder.build();
         }
        // ......
     }
    

    可以看到它也是构造了一个 线程池注入到Spring 中

    build()调用继续看下去,

     public ThreadPoolTaskScheduler build() {
         return configure(new ThreadPoolTaskScheduler());
     }
    

    ThreadPoolTaskScheduler中,给定的线程池的核心参数就为1,这也表明了之前为什么只有一条线程在执行任务。private volatile int poolSize = 1;

    这一段是分开的用代码不好展示,我用图片标明出来。

    主要逻辑在这里,创建线程池的时候,只使用了三个参数,剩下的都是使用ScheduledExecutorService的默认的参数

         protected ScheduledExecutorService createExecutor(
                 int poolSize, ThreadFactory threadFactory, RejectedExecutionHandler rejectedExecutionHandler) 
    

    而这默认参数是不行的,生产环境的大坑,阿里的 Java 开发手册中也明确规定,要手动创建线程池,并给定合适的参数值~是为什么呢?

    因为默认的线程池中, 池中允许的最大线程数和最大任务等待队列都是Integer.MAX_VALUE.

    大家都懂的,如果使用这玩意,只要出了问题,必定挂~

    configure(new ThreadPoolTaskScheduler())这里就是构造,略过~

    如果已经较为熟悉SpringBoot的朋友,现在已然明白解决当前问题的方式~

    四、解决方式

    1、@EnableConfigurationProperties(TaskSchedulingProperties.class) ,自动装配类通常也都会对应有个xxxxProperties文件滴,TaskSchedulingProperties也确实可以配置核心线程数等基本参数,但是无法配置线程池中最大的线程数量和等待队列数量,这种方式还是不合适的。

    2、可以手动异步编排,交给某个线程池来执行。

    3、将定时任务加上异步注解@Async,将其改为异步的定时任务,另外自定义一个系统通用的线程池,让异步任务使用该线程执行任务~

    我们分别针对上述三种方式来实现一遍

    4.1、修改配置文件

    可以配置的就下面几项~

     spring:
       task:
         scheduling:
           thread-name-prefix: nzc-schedule- #线程名前缀
           pool:
             size: 10 #核心线程数
          # shutdown:
          #  await-termination: true #执行程序是否应等待计划任务在关机时完成。
          #   await-termination-period:  #执行程序应等待剩余任务完成的最长时间。
    

    测试结果:

     2022-09-06 20:49:15.015  INFO 7852 --- [ nzc-schedule-1] com.nzc.service.ScheduleService          : 当前执行任务的线程号ID===>64
     2022-09-06 20:49:30.004  INFO 7852 --- [ nzc-schedule-2] com.nzc.service.ScheduleService          : 当前执行任务的线程号ID===>66
     2022-09-06 20:49:45.024  INFO 7852 --- [ nzc-schedule-1] com.nzc.service.ScheduleService          : 当前执行任务的线程号ID===>64
     2022-09-06 20:50:00.025  INFO 7852 --- [ nzc-schedule-3] com.nzc.service.ScheduleService          : 当前执行任务的线程号ID===>67
     2022-09-06 20:50:15.023  INFO 7852 --- [ nzc-schedule-2] com.nzc.service.ScheduleService          : 当前执行任务的线程号ID===>66
     2022-09-06 20:50:30.008  INFO 7852 --- [ nzc-schedule-4] com.nzc.service.ScheduleService          : 当前执行任务的线程号ID===>68

    请注意:这里的配置并非是一定生效的,修改后有可能成功,有可能失败,具体原因未知,但这一点是真实存在的。

    不过从执行结果中可以看出,这里的执行的线程不再是孤单单的一个。

    4.2、执行逻辑改为异步执行

    首先我们先向Spring中注入一个我们自己编写的线程池,参数自己设置即可,我这里比较随意。

     @Configuration
     public class MyTheadPoolConfig {
         @Bean
         public TaskExecutor taskExecutor() {
             ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
             //设置核心线程数
             executor.setCorePoolSize(10);
             //设置最大线程数
             executor.setMaxPoolSize(20);
             //缓冲队列200:用来缓冲执行任务的队列
             executor.setQueueCapacity(200);
             //线程活路时间 60 秒
             executor.setKeepAliveSeconds(60);
             //线程池名的前缀:设置好了之后可以方便我们定位处理任务所在的线程池
             // 这里我继续沿用 scheduling 默认的线程名前缀
             executor.setThreadNamePrefix("nzc-create-scheduling-");
             //设置拒绝策略
             executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
             executor.setWaitForTasksToCompleteOnShutdown(true);
             return executor;
         }
     }
    

    然后在定时任务这里注入进去:

     /**
      * @description:
      * @author: Ning Zaichun
      * @date: 2022年09月06日 0:02
      */
     @Slf4j
     @Component
     @EnableScheduling
     public class ScheduleService {
         @Autowired
         TaskExecutor taskExecutor;
         @Scheduled(cron = "0/5 * * * * ? ")
         public void testSchedule() {
             CompletableFuture.runAsync(()->{
                 try {
                     Thread.sleep(10000);
                     log.info("当前执行任务的线程号ID===>{}", Thread.currentThread().getId());
                 } catch (Exception e) {
                     e.printStackTrace();
                 } 
             },taskExecutor);
         }
     }
    

    测试结果:

     2022-09-06 21:00:00.019  INFO 18356 --- [te-scheduling-1] com.nzc.service.ScheduleService          : 当前执行任务的线程号ID===>66
     2022-09-06 21:00:05.022  INFO 18356 --- [te-scheduling-2] com.nzc.service.ScheduleService          : 当前执行任务的线程号ID===>67
     2022-09-06 21:00:10.013  INFO 18356 --- [te-scheduling-3] com.nzc.service.ScheduleService          : 当前执行任务的线程号ID===>68
     2022-09-06 21:00:15.020  INFO 18356 --- [te-scheduling-4] com.nzc.service.ScheduleService          : 当前执行任务的线程号ID===>69
     2022-09-06 21:00:20.026  INFO 18356 --- [te-scheduling-5] com.nzc.service.ScheduleService          : 当前执行任务的线程号ID===>70

    可以看到虽然业务执行时间比较长,但是木有再出现,延迟执行定时任务的情况。

    4.3、异步定时任务

    异步定时任务其实和上面的方式原理是一样的,不过实现稍稍不同罢了。

    在定时任务的类上再加一个@EnableAsync注解,给方法添加一个@Async即可。

    不过一般@Async都会指定线程池,比如写成这样@Async(value = "taskExecutor"),

     /**
      * @description:
      * @author: Ning Zaichun
      * @date: 2022年09月06日 0:02
      */
     @Slf4j
     @Component
     @EnableAsync
     @EnableScheduling
     public class ScheduleService {
         @Autowired
         TaskExecutor taskExecutor;
         @Async(value = "taskExecutor")
         @Scheduled(cron = "0/5 * * * * ? ")
         public void testSchedule() {
                 try {
                     Thread.sleep(10000);
                     log.info("当前执行任务的线程号ID===>{}", Thread.currentThread().getId());
                 } catch (Exception e) {
                     e.printStackTrace();
                 } 
         }
     }
    

    执行结果:

     2022-09-06 21:10:15.022  INFO 22760 --- [zc-scheduling-1] com.nzc.service.ScheduleService          : 当前执行任务的线程号ID===>66
     2022-09-06 21:10:20.021  INFO 22760 --- [zc-scheduling-2] com.nzc.service.ScheduleService          : 当前执行任务的线程号ID===>67
     2022-09-06 21:10:25.007  INFO 22760 --- [zc-scheduling-3] com.nzc.service.ScheduleService          : 当前执行任务的线程号ID===>68
     2022-09-06 21:10:30.020  INFO 22760 --- [zc-scheduling-4] com.nzc.service.ScheduleService          : 当前执行任务的线程号ID===>69
     2022-09-06 21:10:35.007  INFO 22760 --- [zc-scheduling-5] com.nzc.service.ScheduleService          : 当前执行任务的线程号ID===>70

    结果显而易见是可行的啦~

    分析

    @EnableAsync注解相应的也有一个自动装配类为TaskExecutionAutoConfiguration

    也有一个TaskExecutionProperties配置类,可以在yml文件中对参数进行设置,这里的话是可以配置线程池最大存活数量的。

    它的默认核心线程数为8,这里我不再进行演示了,同时它的线程池中最大存活数量以及任务等待数量也都为Integer.MAX_VALUE,这也是不建议大家使用默认线程池的原因。

    4.4、小结

     /**
      * 定时任务
      *      1、@EnableScheduling 开启定时任务
      *      2、@Scheduled开启一个定时任务
      *      3、自动装配类 TaskSchedulingAutoConfiguration
      *
      * 异步任务
      *      1、@EnableAsync:开启异步任务
      *      2、@Async:给希望异步执行的方法标注
      *      3、自动装配类 TaskExecutionAutoConfiguration
      */
    

    实现方式虽不同,但从效率而言,并无太大区别,觉得那种合适使用那种便可。

    不过总结起来,考查的都是对线程池的理解,对于线程池的了解是真的非常重要的,也很有用处

    五、分布式下的思考

    针对上述情况而言,这些解决方法在不引入第三包的情况下是足以应付大部分情况了。

    定时框架的实现有许多方式,在此并非打算讨论这个。

    在单体项目中,也许上面的问题是解决了,但是站在分布式的情况下考虑,就并非是安全的了。

    当多个项目在同时运行,那么必然会有多个项目同时这段代码。

    思考:并发执行

    如果一个定时任务同时在多个机器中运行,会产生怎么样的问题?

    假如这个定时任务是收集某个信息,发送给消息队列,如果多台机器同时执行,同时给消息队列发送信息,那么必然导致之后产生一系列的脏数据。这是非常不可靠的

    解决方式:分布式锁

    很简单也不简单,加分布式锁~ 或者是用一些分布式调度的框架

    如使用XXL-JOB实现,或者是其他的定时任务框架。

    大家在执行这个定时任务之前,先去获取一把分布式锁,获取到了就执行,获取不到就直接结束。

    我这里使用的是 redission,因为方便,打算写分布式锁的文章,还在准备当中。

    redission官方文档,我觉得应当算是比较友好的文档了哈哈

    加入依赖:

     <dependency>
         <groupId>org.springframework.boot</groupId>
         <artifactId>spring-boot-starter-data-redis</artifactId>
     </dependency>
     <dependency>
         <groupId>org.redisson</groupId>
         <artifactId>redisson-spring-boot-starter</artifactId>
         <version>3.17.6</version>
     </dependency>
    

    按照文档说的,编写配置类,注入 RedissonClient,redisson的全部操作都是基于此。

     /**
      * @description:
      * @author: Ning Zaichun
      * @date: 2022年09月06日 9:31
      */
     @Configuration
     public class MyRedissonConfig {
         /**
          * 所有对Redisson的使用都是通过RedissonClient
          * @return
          * @throws IOException
          */
         @Bean(destroyMethod="shutdown")
         public RedissonClient redissonClient() throws IOException {
             //1、创建配置
             Config config = new Config();
            // 这里规定要用 redis://+IP地址
               config.useSingleServer().setAddress("redis://xxxxx:6379").setPassword("000415");   // 有密码就写密码~ 木有不用写~
             //2、根据Config创建出RedissonClient实例
             //Redis url should start with redis:// or rediss://
             RedissonClient redissonClient = Redisson.create(config);
             return redissonClient;
         }
     }
    

    修改定时任务:

     
     /**
      * @description:
      * @author: Ning Zaichun
      * @date: 2022年09月06日 0:02
      */
     @Slf4j
     @Component
     @EnableAsync
     @EnableScheduling
     public class ScheduleService {
         @Autowired
         TaskExecutor taskExecutor;
         @Autowired
         RedissonClient redissonClient;
         private final String SCHEDULE_LOCK = "schedule:lock";
         @Async(value = "taskExecutor")
         @Scheduled(cron = "0/5 * * * * ? ")
         public void testSchedule() {
             //分布式锁
             RLock lock = redissonClient.getLock(SCHEDULE_LOCK);
             try {
                 //加锁 10 为时间,加上时间 默认会去掉 redisson 的看门狗机制(即自动续锁机制)
                 lock.lock(10, TimeUnit.SECONDS);
                 Thread.sleep(10000);
                 log.info("当前执行任务的线程号ID===>{}", Thread.currentThread().getId());
             } catch (Exception e) {
                 e.printStackTrace();
             } finally {
                 // 一定要记得解锁~
                 lock.unlock();
             }
         }
     }
    

    这里只是给出个大概的实现,实际上还是可以优化的,比如在给定一个flag,在获取锁之前判断。如果有人抢到锁,就修改这个值,之后的请求,判断这个flag,如果不是默认的值,则直接结束任务等等。

    思考:继续往深处思考,在分布式情况下如果一个定时任务抢到锁,但是它在执行业务过程中失败或者是宕机了,这又该如何处理呢?如何补偿呢?

    个人思考:

    失败还比较好说,我们可以直接try{}catch(){}中进行通知告警,及时检查出问题。

    如果是挂了,我还没想好怎么做。

    后记

    但实际上,我所阐述的这种方式,只能说适用于简单的单体项目,一旦牵扯到动态定时任务,使用这种方式就不再那么方便了。

    大部分都是使用定时任务框架集成了,尤其是分布式调度远比单体项目需要考虑多的多。

    以上就是Schedule定时任务在分布式产生的问题详解的详细内容,更多关于Schedule定时任务分布式的资料请关注自由互联其它相关文章!

    上一篇:Java多线程案例之定时器详解
    下一篇:没有了
    网友评论