一、Quartz 了解

image.png

1.1、Scheduler

  • Scheduler为quartz中的调度器，Quartz通过调度器来注册、暂停、删除Trigger和JobDetail
  • Scheduler拥有SchedulerContext，顾名思义就是上下文，通过SchedulerContext可以获取到触发器和任务的一些信息

1.2、Trigger

  • Trigger为触发器，通过cron表达式或日历指定任务执行的周期
  • 系统时间走到触发器指定的时间时，触发器就会触发任务的执行

1.3、JobDetail

  • Job接口是真正需要执行的任务

  • JobDetail核心调度实现了Job类的任务类，Trigger和Scheduler实际用到的都是JobDetail

    1.4、 Job

  • 完成任务的最小实现类，如果需要被定时调度的类都需要实现此接口

1.5、问题

为什么 需要集群？

  • 防止单点故障，减少对业务的影响
  • 减少节点的压力，例如在 10 点要触发 1000 个任务，如果有 10 个节点，则每个节点之需要执行 100 个任务

集群需要解决的问题？

  • 1、任务重跑，因为节点部署的内容是一样的，到 10 点的时候，每个节点都会执行相同的操作，引起数据混乱。比如跑批，绝对不能执行多次。
  • 2、任务漏跑，假如任务是平均分配的，本来应该在某个节点上执行的任务，因为节点故障，一直没有得到执行。
  • 3、水平集群需要注意时间同步问题
  • 4、Quartz 使用的是随机的负载均衡算法，不能指定节点执行所以必须要有一种共享数据或者通信的机制。在分布式系统的不同节点中，我们可以采用什么样的方式，实现数据共享？两两通信，或者基于分布式的服务，实现数据共享。例如：ZK、Redis、DB。
    • 在 Quartz 中，提供了一种简单的方式，基于数据库共享任务执行信息。也就是说，一个节点执行任务的时候，会操作数据库，其他的节点查询数据库，便可以感知到了。同样的问题：建什么表？哪些字段？依旧使用系统自带的 11 张表。

Quartz如何保证任务只在一个节点运行(即任务不会重复执行)？

  • 在acquireNextTriggers()方法中，通过数据库锁 + 事务来保证Trigger只被集群中的一个节点获取到

在集群的哪个节点上运行，Quartz是如何进行选取的？

  • 随缘的。集群中各个节点做到时间同步很重要。如果待触发的任务少且运行快，那么很可能一直在时间最早的那一个节点上执行。

任务太多，而线程池中配置的线程太少时怎么办？

  • 选取Trigger的时候也会考虑空闲线程的数量，空闲线程少的话实例就少选取几个Trigger来执行

二、Spring 整合 Quartz

2.1、依赖

  1. <dependency>
  2.   <groupId>org.springframework.boot</groupId>
  3.   <artifactId>spring-boot-starter-web</artifactId>
  4. </dependency>
  6. <dependency>
  7.   <groupId>org.springframework.boot</groupId>
  8.   <artifactId>spring-boot-starter-quartz</artifactId>
  9. </dependency>
  11. <dependency>
  12.   <groupId>mysql</groupId>
  13.   <artifactId>mysql-connector-java</artifactId>
  14. </dependency>
  16. <dependency>
  17.   <groupId>org.springframework.boot</groupId>
  18.   <artifactId>spring-boot-starter-jdbc</artifactId>
  19. </dependency>

2.2、配置属性

  1. spring:
  2.   datasource:
  3.     driver-class-name: com.mysql.cj.jdbc.Driver
  4.     url: jdbc:mysql://127.0.0.1:33060/quartz_demo?useUnicode=true&characterEncoding=utf-8&useSSL=false
  5.     username: root
  6.     password: root@01234560
  7.   # 定时配置
  8.   quartz:
  9.     # 将任务等保存化到数据库
  10.     job-store-type: jdbc
  11.     # 程序结束时会等待quartz相关的内容结束
  12.     wait-for-jobs-to-complete-on-shutdown: true
  13.     # QuartzScheduler启动时更新己存在的Job,这样就不用每次修改targetObject后删除qrtz_job_details表对应记录
  14.     overwrite-existing-jobs: true
  15.     # 相关属性配置
  16.     properties:
  17.       org:
  18.         quartz:
  19.           # 自定义 Quartz 数据源
  20.           dataSource:
  21.             globalJobDataSource:
  22.               # URL必须大写 ，这里直接使用项目数据源
  23.               URL: ${spring.datasource.url}
  24.               driver:  ${spring.datasource.driver-class-name}
  25.               maxConnections: 5
  26.               username: ${spring.datasource.username}
  27.               password: ${spring.datasource.password}
  28.               # 必须指定数据源类型
  29.               provider: hikaricp
  30.           scheduler:
  31.             # scheduler的实例名
  32.             instanceName: scheduler
  33.             instanceId: AUTO
  34.           jobStore:
  35.             # 数据源
  36.             dataSource: globalJobDataSource
  37.             # JobStoreTX将用于独立环境，提交和回滚都将由这个类处理
  38.             class: org.quartz.impl.jdbcjobstore.JobStoreTX
  39.             # 驱动配置
  40.             driverDelegateClass: org.quartz.impl.jdbcjobstore.StdJDBCDelegate
  41.             # 表前缀
  42.             tablePrefix: QRTZ_
  43.             useProperties: false
  44.             #打开群集功能
  45.             #isClustered: true 
  46.           # 线程池配置
  47.           threadPool:
  48.             class: org.quartz.simpl.SimpleThreadPool
  49.             # 线程数
  50.             threadCount: 10
  51.             # 优先级
  52.             threadPriority: 5

三、Quartz 源码了解

  • 任务执行状态过程

3.1、Quartz 表结构了解

Table Name Description
QRTZ_CALENDARS 存储Quartz的Calendar信息
QRTZ_CRON_TRIGGERS 存储CronTrigger，包括Cron表达式和时区信息
QRTZ_FIRED_TRIGGERS 存储与已触发的Trigger相关的状态信息，以及相联Job的执行信息
QRTZ_PAUSED_TRIGGER_GRPS 存储已暂停的Trigger组的信息
QRTZ_SCHEDULER_STATE 存储少量的有关Scheduler的状态信息，和别的Scheduler实例
QRTZ_LOCKS 存储程序的悲观锁的信息
QRTZ_JOB_DETAILS 存储每一个已配置的Job的详细信息
QRTZ_JOB_LISTENERS 存储有关已配置的JobListener的信息
QRTZ_SIMPLE_TRIGGERS 存储简单的Trigger，包括重复次数、间隔、以及已触的次数
QRTZ_BLOG_TRIGGERS Trigger作为Blob类型存储
QRTZ_TRIGGER_LISTENERS 存储已配置的TriggerListener的信息
QRTZ_TRIGGERS 存储已配置的Trigger的信息

3.2、添加任务流程

3.2.1、流程

  • 校验 JobDetial
  • 校验 Trigger
  • 获取 JobStore 存储 Job 和 JobStore
    • 获取 Semaphore 管理锁对象
    • 判断是否是锁的拥有者
    • 查询锁，没有则插入锁记录
    • 如果有其他现在也操作，则阻塞，等待commit后释放
  • 通知了 监听器 添加了 Job
  • 通知了 调度线程，改触发器下次执行时间
  • 通知监听器添加了 JobStore

3.2.2、获取锁的过程（集群）

  • 锁名称定义

    1. CALENDAR_ACCESS 
    2. JOB_ACCESS      
    3. MISFIRE_ACCESS  
    4. STATE_ACCESS    
    5. TRIGGER_ACCESS

  • 获取锁SQL ```sql

    查询锁记录 QRTZ_ 是自定义前缀

    select * from QRTZ_LOCKS t where t.lock_name=’TRIGGER_ACCESS’ for update

插入锁记录 SCHED_NAME 实例名称

INSERT INTO QRTZ_LOCKS(SCHED_NAME, LOCK_NAME) VALUES (‘scheduler’, ‘TRIGGER_ACCESS’);

  2. - 流程图示
  4. ![image.png](https://cdn.nlark.com/yuque/0/2022/png/438760/1651052676852-ef25b43d-1c0d-447d-b5ac-6b0c8d0104fa.png#clientId=u16680f6c-2f99-4&crop=0&crop=0&crop=1&crop=1&from=paste&id=uc279ea49&margin=%5Bobject%20Object%5D&name=image.png&originHeight=575&originWidth=496&originalType=url&ratio=1&rotation=0&showTitle=false&size=37603&status=done&style=none&taskId=u768fb6f8-714a-4efc-9788-1368d0cbacf&title=)
  7. <a name="mCpTh"></a>
  8. ### 3.3、调度线程执行任务过程
  9. <a name="IT5xU"></a>
  10. #### 3.3.1、执行流程
  12. - 线程类  QuartzSchedulerThread
  13. - 调度器在trigger队列中寻找30秒内一定数目的trigger(需要保证集群节点的系统时间一致)
  14. - 判断触发器 Trigger状态 是否为  ACQUIRED
  15.    - 获取 JobDetail 
  16.    - 更新触发器 执行状态
  17.    - 更新触发器下次执行时间
  18.    - 如果禁止并发操作，则会暂停该任务的其它触发器
  19. - QuartzSchedulerResources 获取线程池执行 Job isClustered: true _#打开群集功能_
  21. <a name="dWo0V"></a>
  22. #### 3.3.2、触发器状态
  23. ![](https://cdn.nlark.com/yuque/0/2022/png/438760/1651117279058-90b7260e-8c2a-4e09-bf87-4e4769b7854e.png#clientId=u16680f6c-2f99-4&crop=0&crop=0&crop=1&crop=1&from=paste&id=UdntW&margin=%5Bobject%20Object%5D&originHeight=606&originWidth=715&originalType=url&ratio=1&rotation=0&showTitle=false&status=done&style=none&taskId=u3ca7e04a-9d7c-4234-8c1a-71d56381295&title=)
  24. ```java
  25.   // STATES
  26. String STATE_WAITING = "WAITING";  //等待
  27. String STATE_ACQUIRED = "ACQUIRED"; //获取到锁
  28. String STATE_EXECUTING = "EXECUTING"; //执行
  29. String STATE_COMPLETE = "COMPLETE"; //完成
  30. String STATE_BLOCKED = "BLOCKED"; //阻塞
  31. String STATE_ERROR = "ERROR"; //错误
  32. String STATE_PAUSED = "PAUSED"; //暂停
  33. String STATE_PAUSED_BLOCKED = "PAUSED_BLOCKED"; //暂停阻塞
  34. String STATE_DELETED = "DELETED";//删除

3.4、集群容错处理

  • 集群管理线程 ClusterManager
  • 查询注册实例状态 ```sql

    查询注册实例状态

    qrtz_scheduler_state

    SELECT * FROM {0}SCHEDULER_STATE WHERE SCHED_NAME = {1} AND INSTANCE_NAME = ?

SELECT * FROM {0}SCHEDULER_STATE WHERE SCHED_NAME = {1} ```

参考