原文地址：http://geek.csdn.net/news/detail/65738

【編者按】?TBSchedule是一款非常優(yōu)秀的高性能分布式調(diào)度框架，本文是作者結(jié)合多年使用TBSchedule的經(jīng)驗，在研讀三遍源碼的基礎(chǔ)上完成。期間作者也與阿里空玄有過不少技術(shù)交流，并非常感謝空玄給予的大力支持。另外，作者寫這篇文章的目的一是出于對TBSchedule的一種熱愛，二是現(xiàn)在是一個資源共享、技術(shù)共享的時代，希望把它展現(xiàn)給大家（送人玫瑰，手留余香），能給大家的工作帶來幫助。

以下為文章正文：

一、TBSchedule初識

? ? 時下互聯(lián)網(wǎng)和電商領(lǐng)域，各個平臺都存在大數(shù)據(jù)、高并發(fā)的特點，對數(shù)據(jù)處理的要求越來越高，既要保證高效性，又要保證安全性、準確性。TBSchedule的使命就是將調(diào)度作業(yè)從業(yè)務(wù)系統(tǒng)中分離出來，降低或者是消除和業(yè)務(wù)系統(tǒng)的耦合度，進行高效異步任務(wù)處理。其實在互聯(lián)網(wǎng)和電商領(lǐng)域TBSchedule的使用非常廣泛，目前被應(yīng)用于阿里巴巴、淘寶、支付寶、京東、聚美、汽車之家、國美等很多互聯(lián)網(wǎng)企業(yè)的流程調(diào)度系統(tǒng)。

? ? 在深入了解TBSchedule之前我們先從內(nèi)部和外部形態(tài)對它有個初步認識，如圖1.1、圖1.2。

圖1.1 TBSchedule關(guān)鍵字

圖1.2 TBSchedule外部形態(tài)

? ? 從TBSchedule的內(nèi)部形態(tài)來說，與他有關(guān)的關(guān)鍵詞包括批量任務(wù)、動態(tài)擴展、多主機、多線程、并發(fā)、分片……，這些詞看起來非常的高大上，都是時下互聯(lián)網(wǎng)技術(shù)比較流行的詞匯。從TBSchedule的外部架構(gòu)來看，一目了然，宿主在調(diào)度應(yīng)用中與ZooKeeper進行通信。一個框架結(jié)構(gòu)是否是優(yōu)秀的，從美感的角度就可以看出來，一個好的架構(gòu)一定是隱藏了內(nèi)部復(fù)雜的原理，外部視覺上美好的，讓用戶使用起來簡單易懂。

二、TBSchedule原理

? ? 為什么TBSchedule值得推廣呢？

傳統(tǒng)的調(diào)度框架spring task、quartz也是可以進行集群調(diào)度作業(yè)的，一個節(jié)點掛了可以將任務(wù)漂移給其他節(jié)點執(zhí)行從而避免單點故障，但是不支持分布式作業(yè)，一旦達到單機處理極限也會存在問題。

elastic-job支持分布式，是一個很好的調(diào)度框架，但是開源時間較短，還沒有經(jīng)歷大范圍市場考驗。

Beanstalkd基于C語言開發(fā)，使用范圍較小，無法引入到php、java系統(tǒng)平臺。

? ? TBSchedule到底有多強大呢？我對TBSchedule的優(yōu)勢特點進行了如下總結(jié)：

支持集群、分布式

靈活的任務(wù)分片

動態(tài)的服務(wù)擴容和資源回收

任務(wù)監(jiān)控支持

經(jīng)歷了多年市場考驗，阿里強大技術(shù)團隊支持

? ? TBSchedule支持Cluster，可以宿主在多臺服務(wù)器多個線程組并行進行任務(wù)調(diào)度，或者說可以將一個大的任務(wù)拆成多個小任務(wù)分配到不同的服務(wù)器。

? ? TBSchedule的分布式機制是通過靈活的Sharding方式實現(xiàn)的，比如可以按所有數(shù)據(jù)的ID按10取模分片（分片規(guī)則如圖2.1）、按月份分片等等，根據(jù)不同的需求，不同的場景由客戶端配置分片規(guī)則。然后就是TBSchedule的宿主服務(wù)器可以進行動態(tài)擴容和資源回收，這個特點主要是因為它后端依賴的ZooKeeper，這里的ZooKeeper對于TBSchedule來說是一個NoSQL，用于存儲策略、任務(wù)、心跳信息數(shù)據(jù)，它的數(shù)據(jù)結(jié)構(gòu)類似文件系統(tǒng)的目錄結(jié)構(gòu)，它的節(jié)點有臨時節(jié)點、持久節(jié)點之分。調(diào)度引擎上線后，隨著業(yè)務(wù)量數(shù)據(jù)量的增多，當前Cluster可能不能滿足目前的處理需求，那么就需要增加服務(wù)器數(shù)量，一個新的服務(wù)器上線后會在ZooKeeper中創(chuàng)建一個代表當前服務(wù)器的一個唯一性路徑（臨時節(jié)點），并且新上線的服務(wù)器會和ZooKeeper保持長連接，當通信斷開后，節(jié)點會自動摘除。

? ? TBSchedule會定時掃描當前服務(wù)器的數(shù)量，重新進行任務(wù)分配。TBSchedule不僅提供了服務(wù)端的高性能調(diào)度服務(wù)，還提供了一個scheduleConsole war隨著宿主應(yīng)用的部署直接部署到服務(wù)器，可以通過web的方式對調(diào)度的任務(wù)、策略進行監(jiān)控管理，以及實時更新調(diào)整。

圖2.1 TBSchedule分片規(guī)則

? ? 是不是已經(jīng)對TBSchedule稍微了有些好感呢？我們接著往下看。

? ? TBSchedule提供了兩個核心組件ScheduleServer、TBScheduleManagerFactory和兩類核心接口IScheduleTaskDeal、IScheduleTaskDealSingle、IScheduleTaskDealMuti，這兩部分是客戶端研發(fā)的關(guān)鍵部分，是使用TBSchedule必須要了解的。

? ? ScheduleServer即任務(wù)處理器，的主要作用是任務(wù)和策略的管理、任務(wù)采集和執(zhí)行，由一組工作線程組成，這組工作線程是基于隊列實現(xiàn)的，進行任務(wù)抓取和任務(wù)處理（有兩種處理模式，下面會講）。每個任務(wù)處理器和ZooKeeper有一個心跳通信連接，用于檢測Server的狀態(tài)和進行任務(wù)動態(tài)分配。舉個例子，比如3臺服務(wù)器的worker集群執(zhí)行出票消息生成任務(wù)，對于這個任務(wù)類型每臺服務(wù)器可以配置一個ScheduleSever（即一個線程組），也可以配置兩個線程組，那么就相當于6臺服務(wù)器在并行執(zhí)行此任務(wù)類型。當某臺服務(wù)器宕機或者其他原因與ZooKeeper通信斷開時，它的任務(wù)將被其他服務(wù)器接管。ScheduleServer參數(shù)定義如圖2.2

圖2.2 ScheduleServer參數(shù)定義

? ? 在這些參數(shù)中taskItems是一個非常重要的屬性，是客戶單可以自由發(fā)揮的地方，是任務(wù)分片的基礎(chǔ)，比如我們處理一個任務(wù)可以根據(jù)ID按10取模，那么任務(wù)項就是0-9，3臺服務(wù)器分別拿到4、 3、 3個任務(wù)項，服務(wù)器的上下線都會對任務(wù)項進行重新分配。任務(wù)項是進行任務(wù)分配的最小單位。一個任務(wù)項只能由一個ScheduleServer來進行處理，但一個Server可以處理任意數(shù)量的任務(wù)項。這就是剛才我們說的分片特性。

? ? 調(diào)度服務(wù)器TBScheduleManagerFactory的主要工作ZooKeeper連接參數(shù)配置和ZooKeeper的初始化、調(diào)度管理。

? ? 兩類核心接口是需要被我們定義的目標任務(wù)實現(xiàn)的，根據(jù)自己的需要進行任務(wù)采集（重寫selectTasks方法）和任務(wù)執(zhí)行（重寫execute方法），這兩類接口也是客戶端研發(fā)根據(jù)需求自由發(fā)揮的地方。

? ? 接下來我們深入了解下TBSchedule，看看它的內(nèi)部是如何實現(xiàn)的。圖2.3流程圖是我花了很多心血通過一周時間畫出來的，基本是清晰的展現(xiàn)了TBSchedule內(nèi)部的執(zhí)行流程以及每個步驟ZooKeeper節(jié)點路徑和數(shù)據(jù)的變化。因為圖中的注釋已經(jīng)描述的很詳細了，每個節(jié)點右側(cè)是ZooKeeper的信息（數(shù)據(jù)結(jié)構(gòu)見圖2.4），這里就不再做過多的文字描述了，有任何建議或者不明白的地方可以找我交流。

圖2.3 TBSchedule內(nèi)部流程

圖2.4 TBSchedule之ZooKeeper數(shù)據(jù)結(jié)構(gòu)

? ? TBSchedule還有個強大之處是它提供了兩種處理器模式模式：

? ? 1. SLEEP模式

? ? 當某一個線程任務(wù)處理完畢，從任務(wù)池中取不到任務(wù)的時候，檢查其它線程是否處于活動狀態(tài)。如果是，則自己休眠；如果其它線程都已經(jīng)因為沒有任務(wù)進入休眠，當前線程是最后一個活動線程的時候，就調(diào)用業(yè)務(wù)接口，獲取需要處理的任務(wù)，放入任務(wù)池中，同時喚醒其它休眠線程開始工作。

? ?2. NOTSLEEP模式

? ? 當一個線程任務(wù)處理完畢，從任務(wù)池中取不到任務(wù)的時候，立即調(diào)用業(yè)務(wù)接口獲取需要處理的任務(wù)，放入任務(wù)池中。

? ? SLEEP模式內(nèi)部邏輯相對較簡單，如果遇到大任務(wù)需要處理較長時間，可能會造成其他線程被動阻塞的情況。但其實生產(chǎn)環(huán)境一般都是小而快的任務(wù)，即使出現(xiàn)阻塞的情況ScheduleConsole也會及時的監(jiān)控到。NOTSLEEP模式減少了線程休眠的時間，避免了因大任務(wù)造成阻塞的情況，但為了避免數(shù)據(jù)被重復(fù)處理，增加了CPU在數(shù)據(jù)比較上的開銷。TBSchedule默認是SLEEP模式。

? ? 到目前為止我相信大家對TBSchedule有了一個深刻的了解，心中的疑霧逐漸散開了。理論是實踐的基礎(chǔ)，實踐才是最終的目的，下一節(jié)我們將結(jié)合理論知識進行TBSchedule實戰(zhàn)。

三、TBSchedule實戰(zhàn)

? ? 在項目中使用TBSchedule需要依賴ZooKeeper、TBSchedule。

? ? ZooKeeper依賴：

<dependency><groupId>org.apache.ZooKeeper</groupId><artifactId>ZooKeeper</artifactId><version>3.4.6</version></dependency>

? ? TBSchedule依賴：

<dependency><groupId>com.taobao.pamirs.schedule</groupId><artifactId>TBSchedule</artifactId><version>3.3.3.2</version></dependency>

? ? TBSchedule有三種引入方式：

通過ScheduleConsole引入

? ? TBSchedule隨著宿主調(diào)度應(yīng)用部署到服務(wù)器后，可以通過Web瀏覽器的方式訪問其提供監(jiān)控平臺。

? ? 第一步，初始化ZooKeeper

? ? 第二步，創(chuàng)建調(diào)度策略

? ? 第三步，創(chuàng)建調(diào)度任務(wù)

? ? 第四步，監(jiān)控調(diào)度任務(wù)

? ? 2、通過原生Java引入

// 初始化SpringApplicationContext ctx = new FileSystemXmlApplicationContext("spring-config.xml");// 初始化調(diào)度工廠TBScheduleManagerFactory scheduleManagerFactory = new TBScheduleManagerFactory();Properties p = new Properties();p.put("zkConnectString", "127.0.0.1:2181");p.put("rootPath", "/taobao-schedule/train_worker");p.put("zkSessionTimeout", "60000"); p.put("userName", "train_dev");p.put("password", " train_dev ");p.put("isCheckParentPath", "true");scheduleManagerFactory.setApplicationContext(ctx);scheduleManagerFactory.init(p); // 創(chuàng)建任務(wù)調(diào)度任務(wù)的基本信息 String baseTaskTypeName = "DemoTask";ScheduleTaskType baseTaskType = new ScheduleTaskType();baseTaskType.setBaseTaskType(baseTaskTypeName);baseTaskType.setDealBeanName("demoTaskBean");baseTaskType.setHeartBeatRate(10000);baseTaskType.setJudgeDeadInterval(100000);baseTaskType.setTaskParameter("AREA=BJ,YEAR>30");baseTaskType.setTaskItems(ScheduleTaskType.splitTaskItem("0:{TYPE=A,KIND=1},1:{TYPE=A,KIND=2},2:{TYPE=A,KIND=3},3:{TYPE=A,KIND=4}," +"4:{TYPE=A,KIND=5},5:{TYPE=A,KIND=6},6:{TYPE=A,KIND=7},7:{TYPE=A,KIND=8}," +"8:{TYPE=A,KIND=9},9:{TYPE=A,KIND=10}"));baseTaskType.setFetchDataNumber(500);baseTaskType.setThreadNumber(5);this.scheduleManagerFactory.getScheduleDataManager().createBaseTaskType(baseTaskType);log.info("創(chuàng)建調(diào)度任務(wù)成功:" + baseTaskType.toString());// 創(chuàng)建任務(wù)的調(diào)度策略String taskName = baseTaskTypeName;String strategyName =taskName +"-Strategy";try {this.scheduleManagerFactory.getScheduleStrategyManager().deleteMachineStrategy(strategyName,true);} catch (Exception e) {e.printStackTrace();}ScheduleStrategy strategy = new ScheduleStrategy();strategy.setStrategyName(strategyName);strategy.setKind(ScheduleStrategy.Kind.Schedule);strategy.setTaskName(taskName);strategy.setTaskParameter("china");strategy.setNumOfSingleServer(1);strategy.setAssignNum(10);strategy.setIPList("127.0.0.1".split(","));this.scheduleManagerFactory.getScheduleStrategyManager().createScheduleStrategy(strategy);log.info("創(chuàng)建調(diào)度策略成功:" + strategy.toString());

? ? 3、通過Spring容器引入

<!-- 初始化ZooKeeper --> <bean id="scheduleManagerFactory"class="xx.xx.TBScheduleManagerFactory"> <property name="zkConfig"> <map><entry key="zkConnectString" value="127.0.0.1:2181" /><entry key="rootPath" value="/taobao-schedule/train_worker" /><entry key="zkSessionTimeout" value="60000" /><entry key="userName" value="train_dev" /><entry key="password" value="train_dev" /><entry key="isCheckParentPath" value="true" /> </map> </property> </bean> <!-- 配置調(diào)度策略 凌晨1點到3點執(zhí)行 --> <bean id="abstractDemoScheduleTask" class="com.xx.core.TBSchedule.InitScheduleTask" abstract="true"> <property name="scheduleTaskType.heartBeatRate" value="10000" /> <property name="scheduleTaskType.judgeDeadInterval" value="100000" /> <property name="scheduleTaskType.permitRunStartTime" value="0 0 1 * * ?"/> <property name="scheduleTaskType.permitRunEndTime" value="0 0 3 * * ?"/> <property name="scheduleTaskType.taskParameter" value="AREA=BJ,YEAR>30" /> <property name="scheduleTaskType.sleepTimeNoData" value="60000"/> <property name="scheduleTaskType.sleepTimeInterval" value="60000"/> <property name="scheduleTaskType.fetchDataNumber" value="500" /> <property name="scheduleTaskType.executeNumber" value="1" /> <property name="scheduleTaskType.threadNumber" value="5" /> <property name="scheduleTaskType.taskItems"> <list><value>0:{TYPE=A,KIND=1}</value><value>1:{TYPE=A,KIND=2}</value><value>2:{TYPE=A,KIND=3}</value><value>3:{TYPE=A,KIND=4}</value><value>4:{TYPE=A,KIND=5}</value><value>5:{TYPE=A,KIND=6}</value><value>6:{TYPE=A,KIND=7}</value><value>7:{TYPE=A,KIND=8}</value><value>8:{TYPE=A,KIND=9}</value><value>9:{TYPE=A,KIND=10}</value></list> </property> <property name="scheduleStrategy.kind" value="Schedule" /> <property name="scheduleStrategy.numOfSingleServer" value="1" /> <property name="scheduleStrategy.assignNum" value="10" /> <property name="scheduleStrategy.iPList"><list><value>127.0.0.1</value></list></property></bean> <!-- 配置調(diào)度任務(wù) --> <bean id="demoTask" class="com.xx.worker.task.DemoTask" parent="abstractDemoScheduleTask"> <property name="scheduleTaskType.baseTaskType" value="demoTask" /> <property name="scheduleTaskType.dealBeanName" value="demoTaskBean" /> <property name="scheduleStrategy.strategyName" value="demoTaskBean-Strategy" /> <property name="scheduleStrategy.taskName" value="demoTaskBean" /> </bean> 調(diào)度任務(wù)具體實現(xiàn) DemoTask.java/*** DemoTask任務(wù)類*/ public class DemoTask mplementsIScheduleTaskDealSingle,TScheduleTaskDeal {/*** 數(shù)據(jù)采集* @param taskItemNum--分配的任務(wù)項 taskItemList--總?cè)蝿?wù)項 * eachFetchDataNum--采集任務(wù)數(shù)量*/@Overridepublic List<DemoTask> selectTasks(String taskParameter,String ownSign, int taskItemNum, List<TaskItemDefine> taskItemList,int eachFetchDataNum) throws Exception {List<DemoTask> taskList = new LinkedList<DemoTask>();//客戶端根據(jù)條件進行數(shù)據(jù)采集start//客戶端根據(jù)條件進行數(shù)據(jù)采集endreturn rt;}/*** 數(shù)據(jù)處理*/@Overridepublic boolean execute(DemoTask task, String ownSign)throws Exception {//客戶端pop任務(wù)進行處理start//客戶端pop任務(wù)進行處理endreturn true;} }

? ? 其實我們看對于TBSchedule客戶端的使用非常簡單，初始化ZooKeeper、配置調(diào)度策略、配置調(diào)度任務(wù)，對調(diào)度任務(wù)進行具體實現(xiàn)，就這幾個步驟?，F(xiàn)在可以慶祝下了，你又掌握了一個優(yōu)秀開源框架的設(shè)計思想和使用方式。

四、TBSchedule挑戰(zhàn)

? ? 任何事物都是沒有最好只有更好，TBSchedule也一樣，雖然它現(xiàn)在已經(jīng)很完美了，我們不能放棄對更完美的追求。阿里團隊可以在下面幾個方面進行優(yōu)化。

目前ScheduleConsole監(jiān)控頁面過于簡單，需完善UI設(shè)計，提高用戶體驗。

支持Zookeeper集群自動切換，避免ZooKeeper服務(wù)的集群單點故障。

原生ZooKeeper操作替換為Curator，Curator對ZooKeeper進行了一次包裝，對原生ZooKeeper的操作做了大量優(yōu)化，Client和Server之間的連接可能出現(xiàn)的問題處理等等，可以進一步提高TBSchedule的高可用。

TBSchedule的幫助文檔較少，網(wǎng)上的資料基本是千篇一律，希望有更多的愛好者加入進來。

? ? 至此，我們已經(jīng)完成了對TBSchedule的全部介紹，盡快使用起來吧！

總結(jié)

以上是生活随笔為你收集整理的【JEECG TBSchedule】详解应对平台高并发的分布式调度框架TBSchedule的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。