分佈式定時任務調度框架實踐

分佈式任務調度框架幾乎是每個大型應用必備的工具，本文介紹了任務調度框架使用的需求背景和痛點，對業界普遍使用的開源分佈式任務調度框架的使用進行了探究實踐，並分析了這幾種框架的優劣勢和對自身業務的思考。

一、業務背景

1.1 爲什麼需要使用定時任務調度

（1）時間驅動處理場景：整點發送優惠券，每天更新收益，每天刷新標籤數據和人羣數據。

（2）批量處理數據：按月批量統計報表數據，批量更新短信狀態，實時性要求不高。

（3）異步執行解耦：活動狀態刷新，異步執行離線查詢，與內部邏輯解耦。

1.2 使用需求和痛點

（1）任務執行監控告警能力。

（2）任務可靈活動態配置，無需重啓。

（3）業務透明，低耦合，配置精簡，開發方便。

（4）易測試。

（5）高可用，無單點故障。

（6）任務不可重複執行，防止邏輯異常。

（7）大任務的分發並行處理能力。

二、開源框架實踐與探索

 2.1 Java 原生 Timer 和 ScheduledExecutorService

2.1.1 Timer 使用

Timer 缺陷：

1. Timer 底層是使用單線程來處理多個 Timer 任務，這意味着所有任務實際上都是串行執行，前一個任務的延遲會影響到之後的任務的執行。

2. 由於單線程的緣故，一旦某個定時任務在運行時，產生未處理的異常，那麼不僅當前這個線程會停止，所有的定時任務都會停止。

3. Timer 任務執行是依賴於系統絕對時間，系統時間變化會導致執行計劃的變更。

由於上述缺陷，儘量不要使用 Timer， idea 中也會明確提示，使用 ScheduledThreadPoolExecutor 替代 Timer 

2.1.2 ScheduledExecutorService 使用

ScheduledExecutorService 對於 Timer 的缺陷進行了修補，首先 ScheduledExecutorService 內部實現是 ScheduledThreadPool 線程池，可以支持多個任務併發執行。

對於某一個線程執行的任務出現異常，也會處理，不會影響其他線程任務的執行，另外 ScheduledExecutorService 是基於時間間隔的延遲，執行不會由於系統時間的改變發生變化。

當然，ScheduledExecutorService 也有自己的侷限性：只能根據任務的延遲來進行調度，無法滿足基於絕對時間和日曆調度的需求。

2.2 Spring Task

2.2.1 Spring Task 使用

spring task 是 spring 自主開發的輕量級定時任務框架，不需要依賴其他額外的包，配置較爲簡單。

此處使用註解配置

2.2.2 Spring Task 缺陷

Spring Task 本身不支持持久化，也沒有推出官方的分佈式集羣模式，只能靠開發者在業務應用中自己手動擴展實現，無法滿足可視化，易配置的需求。

2.3 永遠經典的 Quartz

2.3.1 基本介紹

Quartz 框架是 Java 領域最著名的開源任務調度工具，也是目前事實上的定時任務標準，幾乎全部的開源定時任務框架都是基於 Quartz 核心調度構建而成。

2.3.2 原理解析

核心組件和架構

 關鍵概念

（1）Scheduler：任務調度器，是執行任務調度的控制器。本質上是一個計劃調度容器，註冊了全部 Trigger 和對應的 JobDetail， 使用線程池作爲任務運行的基礎組件，提高任務執行效率。

（2）Trigger：觸發器，用於定義任務調度的時間規則，告訴任務調度器什麼時候觸發任務，其中 CronTrigger 是基於 cron 表達式構建的功能強大的觸發器。

（3）Calendar：日曆特定時間點的集合。一個 trigger 可以包含多個 Calendar，可用於排除或包含某些時間點。

（4）JobDetail：是一個可執行的工作，用來描述 Job 實現類及其它相關的靜態信息，如 Job 的名稱、監聽器等相關信息。 

（5）Job：任務執行接口，只有一個 execute 方法，用於執行真正的業務邏輯。

（6）JobStore：任務存儲方式，主要有 RAMJobStore 和 JDBCJobStore，RAMJobStore 是存儲在 JVM 的內存中，有丟失和數量受限的風險，JDBCJobStore 是將任務信息持久化到數據庫中，支持集羣。

2.3.3 實踐說明

（1）關於 Quartz 的基本使用

• 可參考 Quartz 官方文檔和網上博客實踐教程。

（2）業務使用要滿足動態修改和重啓不丟失， 一般需要使用數據庫進行保存。

• Quartz 本身支持 JDBCJobStore，但是其配置的數據表比較多，官方推薦配置可參照官方文檔，超過 10 張表，業務使用比較重。

• 在使用的時候只需要存在基本 trigger 配置和對應任務以及相關執行日誌的表即可滿足絕大部分需求。

（3）組件化

• 將 quartz 動態任務配置信息持久化到數據庫，將數據操作包裝成基本 jar 包，供項目之間使用，引用項目只需要引入 jar 包依賴和配置對應的數據表，使用時就可以對 Quartz 配置透明。

（4）擴展

• 集羣模式

  通過故障轉移和負載均衡實現了任務的高可用性，通過數據庫的鎖機制來確保任務執行的唯一性，但是集羣特性僅僅只是用來 HA，節點數量的增加並不會提升單個任務的執行效率，不能實現水平擴展。

• Quartz 插件

  可以對特定需要進行擴展，比如增加觸發器和任務執行日誌，任務依賴串行處理場景，可參考：quartz 插件——實現任務之間的串行調度

2.3.4 缺陷和不足

（1）需要把任務信息持久化到業務數據表，和業務有耦合。

（2）調度邏輯和執行邏輯並存於同一個項目中，在機器性能固定的情況下，業務和調度之間不可避免地會相互影響。

（3）quartz 集羣模式下，是通過數據庫獨佔鎖來唯一獲取任務，任務執行並沒有實現完善的負載均衡機制。

2.4 輕量級神器 XXL-JOB

2.4.1 基本介紹

XXL-JOB 是一個輕量級分佈式任務調度平臺，主打特點是平臺化，易部署，開發迅速、學習簡單、輕量級、易擴展，代碼仍在持續更新中。

 “調度中心”是任務調度控制檯，平臺自身並不承擔業務邏輯，只是負責任務的統一管理和調度執行，並且提供任務管理平臺，  “執行器” 負責接收 “調度中心” 的調度並執行，可直接部署執行器，也可以將執行器集成到現有業務項目中。 通過將任務的調度控制和任務的執行解耦，業務使用只需要關注業務邏輯的開發。

主要提供了任務的動態配置管理、任務監控和統計報表以及調度日誌幾大功能模塊，支持多種運行模式和路由策略，可基於對應執行器機器集羣數量進行簡單分片數據處理。

2.4.2 原理解析

2.1.0 版本前核心調度模塊都是基於 quartz 框架，2.1.0 版本開始自研調度組件，移除 quartz 依賴 ，使用時間輪調度。

2.4.3 實踐說明

詳細配置和介紹參考官方文檔。

 2.4.3.1 demo 使用：

示例 1：實現簡單任務配置，只需要繼承 IJobHandler 抽象類，並聲明註解 

@JobHandler(value="offlineTaskJobHandler") ，實現業務邏輯即可。（注：此次引入了 dubbo，後文介紹）。

@JobHandler(value="offlineTaskJobHandler")
@Component
public class OfflineTaskJobHandler extends IJobHandler {
  
   @Reference(check = false,version = "cms-dev",group="cms-service")
   private OfflineTaskExecutorFacade offlineTaskExecutorFacade;
  
   @Override
   public ReturnT<String> execute(String param) throws Exception {
      XxlJobLogger.log(" offlineTaskJobHandler start.");
  
      try {
         offlineTaskExecutorFacade.executeOfflineTask();
      } catch (Exception e) {
         XxlJobLogger.log("offlineTaskJobHandler-->exception." , e);
         return FAIL;
      }
  
      XxlJobLogger.log("XXL-JOB, offlineTaskJobHandler end.");
      return SUCCESS;
   }
}

 示例 2：分片廣播任務。

@JobHandler(value="shardingJobHandler")
@Service
public class ShardingJobHandler extends IJobHandler {
  
   @Override
   public ReturnT<String> execute(String param) throws Exception {
  
      // 分片參數
      ShardingUtil.ShardingVO shardingVO = ShardingUtil.getShardingVo();
      XxlJobLogger.log("分片參數：當前分片序號 = {}, 總分片數 = {}", shardingVO.getIndex(), shardingVO.getTotal());
  
      // 業務邏輯
      for (int i = 0; i < shardingVO.getTotal(); i++) {
         if (i == shardingVO.getIndex()) {
            XxlJobLogger.log("第 {} 片, 命中分片開始處理", i);
         } else {
            XxlJobLogger.log("第 {} 片, 忽略", i);
         }
      }
  
      return SUCCESS;
   }
}

 2.4.3.2 整合 dubbo 

（1）引入 dubbo-spring-boot-starter 和業務 facade jar 包依賴。

<dependency>
    <groupId>com.alibaba.spring.boot</groupId>
    <artifactId>dubbo-spring-boot-starter</artifactId>
    <version>2.0.0</version>
</dependency>
  
<dependency>
    <groupId>com.demo.service</groupId>
    <artifactId>xxx-facade</artifactId>
    <version>1.9-SNAPSHOT</version>
</dependency>

（2）配置文件加入 dubbo 消費端配置（可根據環境定義多個配置文件，通過 profile 切換）。

## Dubbo 服務消費者配置
spring.dubbo.application.name=xxl-job
  
spring.dubbo.registry.address=zookeeper://zookeeper.xyz:2183
spring.dubbo.port=20880
  
spring.dubbo.version=demo
spring.dubbo.group=demo-service

（3）代碼中通過 @Reference 注入 facade 接口即可。

@Reference(check = false,version = "demo",group="demo-service")
private OfflineTaskExecutorFacade offlineTaskExecutorFacade;

（4）啓動程序加入 @EnableDubboConfiguration 註解。

@SpringBootApplication
@EnableDubboConfiguration
public class XxlJobExecutorApplication {
   public static void main(String[] args) {
        SpringApplication.run(XxlJobExecutorApplication.class, args);
   }
}

2.4.4 任務可視化配置 

內置了平臺項目，方便了開發者對任務的管理和執行日誌的監控，並提供了一些便於測試的功能。

2.4.5 擴展 

（1）任務監控和報表的優化。

（2）任務報警方式的擴展，比如加入告警中心，提供內部消息，短信告警。

（3）對實際業務內部執行出現異常情況下的不同監控告警和重試策略。

2.5 高可用 Elastic-Job

2.5.1 基本介紹

Elastic-Job 是一個分佈式調度解決方案，由兩個相互獨立的子項目 Elastic-Job-Lite 和 Elastic-Job-Cloud 組成。

Elastic-Job-Lite 定位爲輕量級無中心化解決方案，使用 jar 包的形式提供分佈式任務的協調服務。

Elastic-Job-Cloud 使用 Mesos + Docker 的解決方案，額外提供資源治理、應用分發以及進程隔離等服務。

可惜的是已經兩年沒有迭代更新記錄。   

2.5.2 原理解析

2.5.3 實踐說明

2.5.3.1 demo 使用

（1）安裝 zookeeper，配置註冊中心 config，配置文件加入註冊中心 zk 的配置。

@Configuration
@ConditionalOnExpression("'${regCenter.serverList}'.length() > 0")
public class JobRegistryCenterConfig {
  
    @Bean(initMethod = "init")
    public ZookeeperRegistryCenter regCenter(@Value("${regCenter.serverList}") final String serverList,
                                             @Value("${regCenter.namespace}") final String namespace) {
        return new ZookeeperRegistryCenter(new ZookeeperConfiguration(serverList, namespace));
    }
}

spring.application.name=demo_elasticjob
  
regCenter.serverList=localhost:2181
regCenter.namespace=demo_elasticjob
  
spring.datasource.url=jdbc:mysql://127.0.0.1:3306/xxl-job?Unicode=true&characterEncoding=UTF-8
spring.datasource.username=user
spring.datasource.password=pwd

（2）配置數據源 config，並配置文件中加入數據源配置。

@Getter
@Setter
@NoArgsConstructor
@AllArgsConstructor
@ToString
@Configuration
@ConfigurationProperties(prefix = "spring.datasource")
public class DataSourceProperties {
    private String url;
    private String username;
    private String password;
  
    @Bean
    @Primary
    public DataSource getDataSource() {
        DruidDataSource dataSource = new DruidDataSource();
        dataSource.setUrl(url);
        dataSource.setUsername(username);
        dataSource.setPassword(password);
        return dataSource;
    }
}

spring.datasource.url=jdbc:mysql://127.0.0.1:3306/xxl-job?Unicode=true&characterEncoding=UTF-8
spring.datasource.username=user
spring.datasource.password=pwd

（3）配置事件 config。

@Configuration
public class JobEventConfig {
    @Autowired
    private DataSource dataSource;
  
    @Bean
    public JobEventConfiguration jobEventConfiguration() {
        return new JobEventRdbConfiguration(dataSource);
    }
}

（4）爲了便於靈活配置不同的任務觸發事件，加入 ElasticSimpleJob 註解。

@Target({ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
public @interface ElasticSimpleJob {
  
    @AliasFor("cron")
    String value() default "";
  
    @AliasFor("value")
    String cron() default "";
  
    String jobName() default "";
  
    int shardingTotalCount() default 1;
  
    String shardingItemParameters() default "";
  
    String jobParameter() default "";
}

（5）對配置進行初始化。

@Configuration
@ConditionalOnExpression("'${elaticjob.zookeeper.server-lists}'.length() > 0")
public class ElasticJobAutoConfiguration {
  
    @Value("${regCenter.serverList}")
    private String serverList;
  
    @Value("${regCenter.namespace}")
    private String namespace;
  
    @Autowired
    private ApplicationContext applicationContext;
    @Autowired
    private DataSource dataSource;
  
    @PostConstruct
    public void initElasticJob() {
        ZookeeperRegistryCenter regCenter = new ZookeeperRegistryCenter(new ZookeeperConfiguration(serverList, namespace));
        regCenter.init();
        Map<String, SimpleJob> map = applicationContext.getBeansOfType(SimpleJob.class);
  
        for (Map.Entry<String, SimpleJob> entry : map.entrySet()) {
            SimpleJob simpleJob = entry.getValue();
            ElasticSimpleJob elasticSimpleJobAnnotation = simpleJob.getClass().getAnnotation(ElasticSimpleJob.class);
  
            String cron = StringUtils.defaultIfBlank(elasticSimpleJobAnnotation.cron(), elasticSimpleJobAnnotation.value());
            SimpleJobConfiguration simpleJobConfiguration = new SimpleJobConfiguration(JobCoreConfiguration.newBuilder(simpleJob.getClass().getName(), cron, elasticSimpleJobAnnotation.shardingTotalCount()).shardingItemParameters(elasticSimpleJobAnnotation.shardingItemParameters()).build(), simpleJob.getClass().getCanonicalName());
            LiteJobConfiguration liteJobConfiguration = LiteJobConfiguration.newBuilder(simpleJobConfiguration).overwrite(true).build();
  
            JobEventRdbConfiguration jobEventRdbConfiguration = new JobEventRdbConfiguration(dataSource);
            SpringJobScheduler jobScheduler = new SpringJobScheduler(simpleJob, regCenter, liteJobConfiguration, jobEventRdbConfiguration);
            jobScheduler.init();
        }
    }
}

（6）實現 SimpleJob 接口，按上文中方法整合 dubbo, 完成業務邏輯。

@ElasticSimpleJob(
        cron = "*/10 * * * * ?",
        jobName = "OfflineTaskJob",
        shardingTotalCount = 2,
        jobParameter = "測試參數",
        shardingItemParameters = "0=A,1=B")
@Component
public class MySimpleJob implements SimpleJob {
    Logger logger = LoggerFactory.getLogger(OfflineTaskJob.class);
  
    @Reference(check = false, version = "cms-dev", group = "cms-service")
    private OfflineTaskExecutorFacade offlineTaskExecutorFacade;
  
  
    @Override
    public void execute(ShardingContext shardingContext) {
  
        offlineTaskExecutorFacade.executeOfflineTask();
  
        logger.info(String.format("Thread ID: %s, 作業分片總數: %s, " +
                        "當前分片項: %s.當前參數: %s," +
                        "作業名稱: %s.作業自定義參數: %s"
                ,
                Thread.currentThread().getId(),
                shardingContext.getShardingTotalCount(),
                shardingContext.getShardingItem(),
                shardingContext.getShardingParameter(),
                shardingContext.getJobName(),
                shardingContext.getJobParameter()
        ));
    }
}

2.6 其餘開源框架

（1）Saturn：Saturn 是唯品會開源的一個分佈式任務調度平臺，在 Elastic Job 的基礎上進行了改造。

（2）SIA-TASK：是宜信開源的分佈式任務調度平臺。

三、優劣勢對比和業務場景適配思考

業務思考：

1. 豐富任務監控數據和告警策略。

2. 接入統一登錄和權限控制。

3. 進一步簡化業務接入步驟。

四、結語

對於併發場景不是特別高的系統來說，xxl-job 配置部署簡單易用，不需要引入多餘的組件，同時提供了可視化的控制檯，使用起來非常友好，是一個比較好的選擇。希望直接利用開源分佈式框架能力的系統，建議根據自身的情況來進行合適的選型。

本文由 Readfog 進行 AMP 轉碼，版權歸原作者所有。
來源：https://mp.weixin.qq.com/s/med31iWll0BDGjPcSq9p2Q