新一代分布式任務(wù)調(diào)度框架：當(dāng)當(dāng)elastic-job開源項目的10項特性

一、為什么需要作業(yè)（定時任務(wù)）？

作業(yè)即定時任務(wù)。一般來說，系統(tǒng)可使用消息傳遞代替部分使用作業(yè)的場景。兩者確有相似之處。可互相替換的場景，如隊列表。將待處理的數(shù)據(jù)放入隊列表，然后使用頻率極短的定時任務(wù)拉取隊列表的數(shù)據(jù)并處理。這種情況使用消息中間件的推送模式可更好的處理實時性數(shù)據(jù)。而且基于數(shù)據(jù)庫的消息存儲吞吐量遠(yuǎn)遠(yuǎn)小于基于文件的順序追加消息存儲。

但在某些場景下則不能互換：

• 時間驅(qū)動 OR 事件驅(qū)動：內(nèi)部系統(tǒng)一般可以通過事件來驅(qū)動，但涉及到外部系統(tǒng)，則只能使用時間驅(qū)動。如：抓取外部系統(tǒng)價格。每小時抓取，由于是外部系統(tǒng)，不能像內(nèi)部系統(tǒng)一樣發(fā)送事件觸發(fā)事件。
• 批量處理 OR 逐條處理：批量處理堆積的數(shù)據(jù)更加高效，在不需要實時性的情況下比消息中間件更有優(yōu)勢。而且有的業(yè)務(wù)邏輯只能批量處理，如：電商公司與快遞公司結(jié)算，一個月結(jié)算一次，并且根據(jù)送貨的數(shù)量有提成。比如，當(dāng)月送貨超過1000則額外給快遞公司多1%優(yōu)惠。
• 非實時性 OR 實時性：雖然消息中間件可以做到實時處理數(shù)據(jù)，但有的情況并不需要如的實時。如：VIP用戶降級，如果超過1年無購買行為，則自動降級。這類需求沒有強烈的時間要求，不需要按照時間精確的降級VIP用戶。

系統(tǒng)內(nèi)部 OR 系統(tǒng)解耦。作業(yè)一般封裝在系統(tǒng)內(nèi)部，而消息中間件可用于系統(tǒng)間解耦。

二、當(dāng)當(dāng)之前在使用什么作業(yè)系統(tǒng)？

當(dāng)當(dāng)之前使用的作業(yè)系統(tǒng)比較散亂，各自為戰(zhàn)，大致分為以下4種：

• Quartz：Java事實上的定時任務(wù)標(biāo)準(zhǔn)。但Quartz關(guān)注點在于定時任務(wù)而非數(shù)據(jù)，并無一套根據(jù)數(shù)據(jù)處理而定制化的流程。雖然Quartz可以基于數(shù)據(jù)庫實現(xiàn)作業(yè)的高可用，但缺少分布式并行執(zhí)行作業(yè)的功能。
• TBSchedule：阿里早期開源的分布式任務(wù)調(diào)度系統(tǒng)。代碼略陳舊，使用timer而非線程池執(zhí)行任務(wù)調(diào)度。眾所周知，timer在處理異常狀況時是有缺陷的。而且TBSchedule作業(yè)類型較為單一，只能是獲取/處理數(shù)據(jù)一種模式。還有就是文檔缺失比較嚴(yán)重。
• Crontab：Linux系統(tǒng)級的定時任務(wù)執(zhí)行器。缺乏分布式和集中管理功能。
• Perl：遺留系統(tǒng)使用，目前已不符合公司的Java化戰(zhàn)略。

三、elastic-job的來歷

elastic-job原本是當(dāng)當(dāng)Java應(yīng)用框架ddframe的一部分，本名dd-job。

ddframe包括編碼規(guī)范，開發(fā)框架，技術(shù)規(guī)范，監(jiān)控以及分布式組件。ddframe規(guī)劃分為4個演進階段，目前處于第2階段。3、4階段涉及的技術(shù)組件不代表當(dāng)當(dāng)沒有使用，只是ddframe還未統(tǒng)一規(guī)劃。

ddframe由各種模塊組成，均已dd-開頭，如dd-container，dd-soa，dd-rdb，dd-job等。當(dāng)當(dāng)希望將ddframe的各個模塊與公司環(huán)境解耦并開源以反饋社區(qū)。之前開源的Dubbo擴展版本DubboX即是dd-soa的核心模塊。而本次介紹的elastic-job則是dd-job的開源部分，其中監(jiān)控（但開源了監(jiān)控方法）和ddframe核心接入等部分并未開源。

四、elastic-job包含的功能

• 分布式：最重要的功能，如果任務(wù)不能在分布式的環(huán)境下執(zhí)行，那么直接使用Quartz就可以了。
• 任務(wù)分片：是elastic-job中最重要也是最難理解的概念。任務(wù)的分布式執(zhí)行，需要將一個任務(wù)拆分為n個獨立的任務(wù)項，然后由分布式的服務(wù)器分別執(zhí)行某一個或幾個分片項。
• 彈性擴容縮容：將任務(wù)拆分為n個任務(wù)項后，各個服務(wù)器分別執(zhí)行各自分配到的任務(wù)項。一旦有新的服務(wù)器加入集群，或現(xiàn)有服務(wù)器下線，elastic-job將在保留本次任務(wù)執(zhí)行不變的情況下，下次任務(wù)開始前觸發(fā)任務(wù)重分片。舉例說明：有3臺服務(wù)器，分為10個片。則分片項分配如下：{server1: [0,1,2], server2: [3,4,5], server3: [6,7,8,9]}。如果一臺服務(wù)器崩潰，則分片項分配如下：{server1: [0,1,2,3,4], server2: [5,6,7,8,9]}。如果新增一臺服務(wù)器，則分片項分配如下：{server1: [0,1], server2: [2,3] , server3: [4,5,6] , server4: [7,8,9]}。
• 穩(wěn)定性：在服務(wù)器無波動的情況下，并不會重新分片；即使服務(wù)器有波動，下次分片的結(jié)果也會根據(jù)服務(wù)器IP和作業(yè)名稱哈希值算出穩(wěn)定的分片順序，盡量不做大的變動。
• 高性能：elastic-job會將作業(yè)運行狀態(tài)的必要信息更新到注冊中心，但為了考慮性能問題，可以犧牲一些功能，而換取性能的提升。
• 冪等性：elastic-job可犧牲部分性能用以保證同一分片項不會同時在兩個服務(wù)器上運行。
• 失效轉(zhuǎn)移：彈性擴容縮容在下次作業(yè)運行前重分片，但本次作業(yè)執(zhí)行的過程中，下線的服務(wù)器所分配的作業(yè)將不會重新被分配。失效轉(zhuǎn)移功能可以在本次作業(yè)運行中用空閑服務(wù)器抓取孤兒作業(yè)分片執(zhí)行。同樣失效轉(zhuǎn)移功能也會犧牲部分性能。
• 狀態(tài)監(jiān)控：監(jiān)控作業(yè)的運行狀態(tài)，可以監(jiān)控數(shù)據(jù)處理功能和失敗次數(shù)，作業(yè)運行時間等。是冪等性，失效轉(zhuǎn)移必須的功能。
• 多作業(yè)模式：作業(yè)可分為簡單和數(shù)據(jù)流處理兩種模式，數(shù)據(jù)流又分為高吞吐處理模式和順序性處理模式，其中高吞吐處理模式可以開啟足夠多的線程快速的處理數(shù)據(jù)，而順序性處理模式將每個分片項分配到一個獨立線程，用于保證同一分片的順序性，這點類似于kafka的分區(qū)順序性。

其他一些功能，如錯過任務(wù)重執(zhí)行，單機并行處理，容錯處理，Spring命名空間支持，運維平臺等。

五、elastic-job的部署和使用

將使用elastic-job框架的jar/war連接同一個基于Zookeeper的注冊中心即可。

作業(yè)框架執(zhí)行數(shù)據(jù)并不限于數(shù)據(jù)庫，且作業(yè)框架本身是不對數(shù)據(jù)進行關(guān)聯(lián)的。作業(yè)可以用于處理數(shù)據(jù)，文件，API等任何操作。

使用elastic-job所需要關(guān)注的僅僅是將業(yè)務(wù)處理邏輯和框架所分配的分片項匹配并處理，如：如果分片項是1，則獲取id以1結(jié)尾的數(shù)據(jù)處理。所以如果是處理數(shù)據(jù)的話，最佳實踐是將作業(yè)分片項規(guī)則和數(shù)據(jù)中間層規(guī)則對應(yīng)。

通過上面的部署圖可以看出來，作業(yè)分片只是個邏輯概念，分片和實際數(shù)據(jù)其實框架是不做任何匹配關(guān)系的。而根據(jù)分片項和實際業(yè)務(wù)如何關(guān)聯(lián)，是成功使用elastic-job的關(guān)鍵所在。為了不讓代碼寫起來很無聊，看起來像if(shardingItem == 1) {do xxx} else if (shardingItem == 2) {do xxx}，elastic-job提供了自定義參數(shù)，可將分片項序號和實際業(yè)務(wù)做映射。比如設(shè)置為1=北京，2=上海。那么代碼中可以通過北京或是上海的枚舉，從業(yè)務(wù)中的北京倉庫或上海倉庫取數(shù)據(jù)。elastic-job更多的還是關(guān)注作業(yè)調(diào)度和分布式分配，處理數(shù)據(jù)還是交由數(shù)據(jù)中間層更好些。

誠如剛才所說，最佳實踐是將作業(yè)分片項規(guī)則和數(shù)據(jù)中間層規(guī)則對應(yīng)，省去作業(yè)分片時，再次適配數(shù)據(jù)中間層的分片邏輯。

六、對開源產(chǎn)品的開發(fā)理念

為了讓感興趣的人放心使用，我想分享一下我們對開源產(chǎn)品的開發(fā)理念。

用心寫代碼。代碼是項目的唯一核心和產(chǎn)出，任何一行的代碼都需要用心思考優(yōu)雅性，可讀性，合理性。優(yōu)雅性看似簡單的幾個字，其實實現(xiàn)的難度非常大。每個人心中都有自己對代碼的理解，而elastic-job也好，ddframe也好，都不是出自一人之手。對代碼優(yōu)雅性的權(quán)衡，是比較難把控的。后面幾項，可以理解為對第一項的補充，或具體的實現(xiàn)思路。

代碼整潔干凈到極致。簡單點說就是重度代碼潔癖患者。只有代碼漂亮整潔，其他開源愛好者才愿意閱讀代碼，進而找出項目中的bug和貢獻高質(zhì)量代碼。

極簡代碼, 高度復(fù)用，無重復(fù)代碼和配置。Java生態(tài)圈的特點是高質(zhì)量的開源產(chǎn)品極多。我們盡量考慮復(fù)用輪子，比如項目中大量用到lombok簡化代碼；但也不會無原則的使用開源產(chǎn)品，我們傾向于把開源產(chǎn)品分為積木類和大廈類。項目中一般只考慮使用積木類搭建屬于我們自己的大廈，而不會直接用其他已成型的大廈。java系的公司有兩種不同的聲音，擁抱開源，或完全不使用開源。我們的看法是既然選擇使用java，就應(yīng)該遵循java的理念，去擁抱java這些年累積的成熟東西。java相比其他新興語言，在語法上可能沒什么優(yōu)勢，但在廣度上還是少有其他生態(tài)圈可比擬。

單一需求可不考慮擴展性；兩個類似需求時再提煉。為了不盲目追求所謂的極致，我們用這條規(guī)則，盡量提升交付的速度。

模塊抽象劃分合理。這點也很難用標(biāo)準(zhǔn)衡量。以elastic-job舉例：elastic-job核心代碼分為4塊，core，spring，console和example；分別用于放置核心，spring支持，控制臺和代碼示例。在項目級別上做拆分。而core中將包分為api，exception，plugin和internal。用于放置對外發(fā)布的接口、異常，向最終用戶提供的可擴展插件以及封裝好的內(nèi)部實現(xiàn)。內(nèi)部實現(xiàn)的任何改動，都不會影響對外接口的變動，用戶自定義的插件，也不會影響內(nèi)部代碼的穩(wěn)定性。

如無特殊理由, 測試需全覆蓋。elastic-job核心模塊的測試覆蓋率是95%以上。雖然單元測試覆蓋率在分布式的復(fù)雜環(huán)境中并無太大說服力，但至少證明項目中很少出現(xiàn)低級邏輯錯誤。

對質(zhì)量的定義。代碼可讀性 > 代碼可測性 > 模塊解耦設(shè)計 > 功能正確性 > 性能 > 功能可擴展性。只有代碼可讀，可測試，可100%掌控，項目才可持續(xù)發(fā)展。功能有缺陷可以修復(fù)，性能不夠可以優(yōu)化，而代碼不清晰則項目會漸漸變?yōu)楹诤?。所以對于框架類產(chǎn)品，我們認(rèn)為質(zhì)量 > 時間 > 成本。

文檔清晰。

七、未來展望

監(jiān)控體系有待提高，目前只能通過注冊中心做簡單的存活和數(shù)據(jù)積壓監(jiān)控。未來需要做的監(jiān)控部分有：

1. 增加可監(jiān)控維度，如作業(yè)運行時間等。
2. 基于JMX的內(nèi)部狀態(tài)監(jiān)控。
3. 基于歷史的全量數(shù)據(jù)監(jiān)控，將所有監(jiān)控數(shù)據(jù)通過flume等形式發(fā)到外部監(jiān)控中心，提供實時分析功能。

增加任務(wù)工作流，如任務(wù)依賴，初始化任務(wù)，清理任務(wù)等。

失效轉(zhuǎn)移功能的實時性提升。

更多作業(yè)類型支持，如文件，MQ等類型作業(yè)的支持。

更多分片策略支持。

Q&A

Q1：請問失效轉(zhuǎn)移中如何判斷失效？對任務(wù)本身實現(xiàn)有什么限制？

失效轉(zhuǎn)移目前通過Zookeeper監(jiān)聽分片項臨時節(jié)點判斷。elastic-job會經(jīng)過注冊中心會話過期時間才能感知任務(wù)掛掉。失效轉(zhuǎn)移有兩種形式：1、任務(wù)掛掉，elastic-job會找空閑的作業(yè)服務(wù)器（可能是未分配任務(wù)的，也可能是完成執(zhí)行本次任務(wù)執(zhí)行的）執(zhí)行。2、如果當(dāng)時沒有空閑服務(wù)器，則將在某服務(wù)器完成分配的任務(wù)時抓取未分配的分片項。

Q2：Zookeeper的作用是保存任務(wù)信息嗎，如果Zookeeper掛了會影響任務(wù)執(zhí)行嗎？

Zookeeper目前的znode分四類，config，servers，execution，leader。config用于保存分布式作業(yè)的全局控制，如，分多少片，要不要執(zhí)行misfire，cron表達(dá)式。servers用于注冊作業(yè)服務(wù)器狀態(tài)和分片信息。execution以分片的維度存儲作業(yè)運行時狀態(tài)。leader用于存儲主節(jié)點。elastic-job作業(yè)執(zhí)行是無中心化的，但主節(jié)點起到協(xié)調(diào)的作用，如：重分片、清理上次運行時信息等。

Q3：在任務(wù)處理上可以與spring batch集成嗎？

spring batch之前關(guān)注過，但目前elastic-job還沒有集成。elastic-job的spring支持是自定義了job的命名空間，更簡化了基于spring的配置，并且可以使用spring注入的bean。spring batch也是很好的作業(yè)框架，包括spring-quartz也很不錯，但分布式功能并不成熟。所以在這之上改動難度比較大，而且elastic-job更希望做一個不依賴于spring，而是能融入spring的綠色產(chǎn)品。

Q4：針對簡單和數(shù)據(jù)流，能夠說說具體分片是怎么處理的嗎？

簡單的作業(yè)就是未經(jīng)過任何業(yè)務(wù)邏輯的封裝，只是提供了一個execute方法，定時觸發(fā)，但是增加了分布式分片功能?？梢院唵卫斫鉃閝uartz的分布式版本。quartz雖然可以支持基于數(shù)據(jù)庫的分布式高可用，但不能分片。也就是說，兩臺服務(wù)器，只能一主一備，不能同時負(fù)載均衡的運行。數(shù)據(jù)流類型作業(yè)參照了阿里之前開源的TBSchedule，將數(shù)據(jù)處理分為fetchData和processData。先將數(shù)據(jù)從數(shù)據(jù)庫，文件系統(tǒng)，或其他數(shù)據(jù)源取出來，然后processData集中處理，可以逐條處理，可以批量處理（這塊未來將加上）。processData是多線程執(zhí)行的，數(shù)據(jù)流類型作業(yè)可再細(xì)分為兩種，一種是高吞吐，一種是順序性。高吞吐可以開啟任意多的線程并行執(zhí)行數(shù)據(jù)處理，而順序執(zhí)行會根據(jù)每個分片項一個線程，保證分片項之中的數(shù)據(jù)有序，這點參照了kafka的實現(xiàn)。數(shù)據(jù)流類型作業(yè)有isStreaming這個參數(shù)，用于控制是否流式不停歇的處理數(shù)據(jù)，類似永動機，只要有數(shù)據(jù)，則一直處理。但這種作業(yè)不適合每次fetchData都對數(shù)據(jù)庫造成壓力很大的場景。

Q5：請問如何實現(xiàn)一個任務(wù)僅僅只在一個節(jié)點執(zhí)行一次？

目前的冪等性，是在execution的znode中增加了對分片項狀態(tài)的注冊，如果狀態(tài)是運行中，即使有別的服務(wù)器要運行這個分片項，elastic-job也會拒絕運行，而是等待這個狀態(tài)變?yōu)榉沁\行的狀態(tài)。每個作業(yè)分片項啟動時會更新狀態(tài)。服務(wù)器沒有波動的情況下，是不存在一個分片被分到兩個服務(wù)器的情況。但一旦服務(wù)器波動，在分片的瞬間有可能出現(xiàn)這種情況。關(guān)于分片，其實是比較復(fù)雜的實現(xiàn)。目前分片是發(fā)現(xiàn)服務(wù)器波動，或修改分片總數(shù)，將記錄一個狀態(tài)，而非直接分片。分片將在下次作業(yè)觸發(fā)時執(zhí)行，只有主節(jié)點可以分片，分片中從節(jié)點都將阻塞。無調(diào)度中心式分布式作業(yè)最大的一個問題是，無法保證主節(jié)點作業(yè)一定先于其他從節(jié)點觸發(fā)。所以很有可能從節(jié)點先觸發(fā)執(zhí)行，而使用舊分片；然后主節(jié)點才重新分片，將造成這次作業(yè)分片可能不一致。這就需要execution節(jié)點來保證冪等性。下次執(zhí)行時，只要無服務(wù)器波動，之前錯誤的分片自然會修正。

Q6：如果Zookeeper掛了，是否全部的任務(wù)都掛了不能運行包括已經(jīng)運行過一次的，如果又恢復(fù)了，任務(wù)能正常運行嗎，還是業(yè)務(wù)應(yīng)用服務(wù)也要重新啟動？

其實Zookeeper是不太容易掛的。畢竟Zookeeper是分布式高可用，一般不會是單臺。目前elastic-job做到的容錯是，連不上Zookeeper的作業(yè)服務(wù)器將立刻停止執(zhí)行作業(yè)，防止主節(jié)點已重新分片，而腦裂的服務(wù)器還在執(zhí)行。也就是說，Zookeeper掛掉，所有作業(yè)都將停止。而作業(yè)服務(wù)器一旦與Zookeeper恢復(fù)連接，作業(yè)也將恢復(fù)運行。所以Zookeeper掛掉不會影響數(shù)據(jù)，而Zookeeper恢復(fù)，作業(yè)會繼續(xù)跑，不用重啟。

Q7：可以具體到業(yè)務(wù)層面嗎？比如有個任務(wù)，是一樣發(fā)送100w的用戶郵件，這時候應(yīng)該怎么分片？針對分布式數(shù)據(jù)庫的分頁在咱們這里又是怎么處理的？

100W用戶的郵件，個人認(rèn)為可以按照用戶id取模，比如分成100個分片，將整個userid % 100，然后每個分片發(fā)送userid結(jié)尾是取摸結(jié)果的郵件。詳細(xì)來說：分片1發(fā)送以01結(jié)尾的userid的郵件，…，分片99發(fā)送以99結(jié)尾的userid的郵件。分布式數(shù)據(jù)庫的分頁，理論上來說，不是作業(yè)框架處理的范疇，應(yīng)由數(shù)據(jù)中間層處理。順便說下，ddframe的數(shù)據(jù)中間層部分，sharding-JDBC將于明年初開源。通過修改JDBC驅(qū)動實現(xiàn)分庫分表。非MyCat或cobar這種中間件方式；也非基于hibernate或mybatis這種ORM方式。sharding-JDBC相對輕量級，也更加容易適配各種數(shù)據(jù)庫和ORM

Q8：ddframe是由很多組件組成？支持多語言嗎？

ddframe是很多組件的總稱。分為核心模塊，分布式組件模塊和監(jiān)控對接模塊等。核心模塊可以理解為spring-boot這種可快速啟動，快速搭建項目的東西。

分布式組件包括SOA調(diào)用的Dubbox，基于分布式作業(yè)的elastic-job，還有剛才提到的sharding-JDBC，以及近期暫無開源計劃的緩存、MQ、NoSQL等模塊。

監(jiān)控模塊估計以后也不會開源，和公司本身的業(yè)務(wù)場景綁定太緊，不是不想開源，是無法開源。主要分為日志中心，流量分析和系統(tǒng)關(guān)系調(diào)用圖。監(jiān)控部分目前也還在做，不是很強大。

多語言方面，SOA模塊支持，Dubbox的REST擴展就是為了支持其他語言的調(diào)用。剩下的暫時不行。比如sharing-JDBC，主要是基于java的JDBC，如果多語言，中間層是個更好的方法。

ddframe的模塊名字都是dd-*，dd-soa，dd-rdb，dd-job，dd-log之類。elastic-job，sharding-JDBC等，是為開源而從ddframe抽離并重新起的名字。

【愛碼仕i】：專注于Java開發(fā)技術(shù)的研究與知識分享！

————END————

• 點贊（感謝）
• ...
• 轉(zhuǎn)發(fā)（感謝）
• ...
• 關(guān)注（感謝）
• ...