在學習完powerjob并了解基本工作原理之后,基于業(yè)務需要基礎的BasicProcessor(單機處理器)不能完美的契合業(yè)務需求,BasicProcessor如果一次任務中處理設備量過大,那么則會造成單個任務實例負載過高,并且系統(tǒng)壓力過大,因此選用了支持將單個任務中全部設備,通過將任務分批拆分為多個包含少量設備的子任務的MapProcessor。
MapProcessor介紹
MapProcessor接口繼承了BasicProcessor接口并且對其進行再次增強,使其可以通過子任務的方式將任務拆分執(zhí)行,對執(zhí)行任務的節(jié)點更加友好,在MapProcessor中可以通過MapProcessor接口的isRoot()方法判斷該次執(zhí)行的任務是否為根任務,調用該方法會返回一個布爾值,為根任務,則可以構建子任務,需手動創(chuàng)建子任務實體類,用于存入子任務執(zhí)行時所需要的參數,此次在項目中使用則是解決設備功能調用中同時給大批量設備同時下發(fā)功能調用指令.考慮到后期設備可能過多,所以選擇MapProcessor處理器通過構建子任務的方式將,任務拆分,每個子任務攜帶部分設備ID分別執(zhí)行,減輕單次執(zhí)行任務時的壓力。
具體實現(xiàn)
1. 定義子任務類
用于在構建子任務時攜帶執(zhí)行所需參數。
2.創(chuàng)建執(zhí)行執(zhí)行器并實現(xiàn)MapProcessor接口
在實現(xiàn)接口時需重寫process()方法即執(zhí)行的任務邏輯
3. isRoot()
通過MapProcessor執(zhí)行任務,無論是根任務還是子任務都是執(zhí)行的同一個process()方法,所以MapProcessor接口提供了isRoot()方法用來判斷,當前執(zhí)行的任務是根任務還是子任務。
4.判斷為根任務
當isRoot()方法返回true時,則表明當前任務為根任務,此時需構建子任務集合,例如此任務需操作1000個設備,那么則可以構建5個子任務將1000個設備ID分別保存到5個子任務對象中,然后將五個子任務對象放入到集合中,再通過調用map()方法,將子任務集合及任務名,傳入到taskContext,任務上下文中,以便后續(xù)的調用。
5.判斷不為根任務
本文僅以簡單的二層任務結構距離,如果判斷不為根任務則此時一定是子任務,只需要通過taskContext調用getSubTask()方法,將我們創(chuàng)建的子任務取出。
并根據先前在創(chuàng)建子任務時傳入的執(zhí)行所需參數,進行對應的業(yè)務處理。
創(chuàng)建任務
在創(chuàng)建需要使用MapProcessor處理器處理的任務時,其他的參數與我們之前創(chuàng)建演示Demo無異,只是需要在執(zhí)行配置處,選擇Map執(zhí)行,并將我們所編寫的處理器全類名填寫正確.如果沒有選擇Map執(zhí)行,任務實例日志中會顯示。
此時則需要檢查,創(chuàng)建任務的參數是否填選正確。
后續(xù)的任務執(zhí)行則無特別操作,只需要在任務管理頁面,通過滑塊來確定是否需要執(zhí)行該任務.并在運行實例中可以查看每次任務執(zhí)行的日志。
拓展: MapReduceProcessor
此處理器則是在MapProcessor處理器基礎上,添加了reduce()方法,在實現(xiàn)該接口時除了需要重寫process方法外還需要重寫reduce方法.reduce方法則是根據process()方法的返回值,對子任務的執(zhí)行狀態(tài)進行統(tǒng)計,方便后續(xù)對子任務的計算操作,單此處理器在執(zhí)行計算統(tǒng)計時,需要對數據庫進行全量掃描,對節(jié)點內存壓力過大,謹慎使用,如果只是簡單的對任務進行拆分,不需要后續(xù)再對子任務處理結果進行計算操作,MapProcessor即可進行處理。
