xxl-job任務(wù)調度中間件解決定時(shí)任務(wù)的調度問(wèn)題

xxl-job是一款非常優(yōu)秀的任務(wù)調度中間件，輕量級、使用簡(jiǎn)單、支持分布式等優(yōu)點(diǎn)，讓它廣泛應用在我們的項目中，解決了不少定時(shí)任務(wù)的調度問(wèn)題。

我們都知道，在使用過(guò)程中需要先到xxl-job的任務(wù)調度中心頁(yè)面上，配置執行器executor 和具體的任務(wù)job ，這一過(guò)程如果項目中的定時(shí)任務(wù)數量不多還好說(shuō)，如果任務(wù)多了的話(huà)還是挺費工夫的。

假設項目中有上百個(gè)這樣的定時(shí)任務(wù)，那么每個(gè)任務(wù)都需要走一遍綁定jobHander后端接口，填寫(xiě)cron表達式這個(gè)流程…

我就想問(wèn)問(wèn)，填多了誰(shuí)能不迷糊？

于是出于功能優(yōu)化（偷懶 ）這一動(dòng)機，前幾天我萌生了一個(gè)想法，有沒(méi)有什么方法能夠告別xxl-job的管理頁(yè)面，能夠讓我不再需要到頁(yè)面上去手動(dòng)注冊執行器和任務(wù)，實(shí)現讓它們自動(dòng)注冊到調度中心呢。

分析

分析一下，其實(shí)我們要做的很簡(jiǎn)單，只要在項目啟動(dòng)時(shí)主動(dòng)注冊executor和各個(gè)jobHandler到調度中心就可以了，流程如下：

有的小伙伴們可能要問(wèn)了，我在頁(yè)面上創(chuàng )建執行器 的時(shí)候，不是有一個(gè)選項叫做自動(dòng)注冊 嗎，為什么我們這里還要自己添加新執行器？

其實(shí)這里有個(gè)誤區，這里的自動(dòng)注冊指的是會(huì )根據項目中配置的xxl.job.executor.appname，將配置的機器地址自動(dòng)注冊到這個(gè)執行器的地址列表中。但是如果你之前沒(méi)有手動(dòng)創(chuàng )建過(guò)執行器，那么是不會(huì )給你自動(dòng)添加一個(gè)新執行器到調度中心的。

既然有了想法咱們就直接開(kāi)干，先到github上拉一份xxl-job的源碼下來(lái)

整個(gè)項目導入idea后，先看一下結構：

結合著(zhù)文檔和代碼，先梳理一下各個(gè)模塊都是干什么的：

xxl-job-admin：任務(wù)調度中心，啟動(dòng)后就可以訪(fǎng)問(wèn)管理頁(yè)面，進(jìn)行執行器和任務(wù)的注冊、以及任務(wù)調用等功能了

xxl-job-core：公共依賴(lài)，項目中使用到xxl-job時(shí)要引入的依賴(lài)包

xxl-job-executor-samples：執行示例，分別包含了springboot版本和不使用框架的版本

為了弄清楚注冊和查詢(xún)executor和jobHandler調用的是哪些接口，我們先從頁(yè)面上去抓一個(gè)請求看看：

好了，這樣就能定位到xxl-job-admin模塊中/jobgroup/save這個(gè)接口，接下來(lái)可以很容易地找到源碼位置：

按照這個(gè)思路，可以找到下面這幾個(gè)關(guān)鍵接口：

/jobgroup/pageList：執行器列表的條件查詢(xún)

/jobgroup/save：添加執行器

/jobinfo/pageList：任務(wù)列表的條件查詢(xún)

/jobinfo/add：添加任務(wù)

但是如果直接調用這些接口，那么就會(huì )發(fā)現它會(huì )跳轉到xxl-job-admin的的登錄頁(yè)面：

其實(shí)想想也明白，出于安全性考慮，調度中心的接口也不可能允許裸調的。那么再回頭看一下剛才頁(yè)面上的請求就會(huì )發(fā)現，它在Headers中添加了一條名為XXL_JOB_LOGIN_IDENTITY的cookie：

至于這條cookie，則是在通過(guò)用戶(hù)名和密碼調用調度中心的/login接口時(shí)返回的，在返回的response可以直接拿到。只要保存下來(lái)，并在之后每次請求時(shí)攜帶，就能夠正常訪(fǎng)問(wèn)其他接口了。

到這里，我們需要的5個(gè)接口就基本準備齊了，接下來(lái)準備開(kāi)始正式的改造工作。

基于 Spring Boot + MyBatis Plus + Vue & Element 實(shí)現的后臺管理系統 + 用戶(hù)小程序，支持 RBAC 動(dòng)態(tài)權限、多租戶(hù)、數據權限、工作流、三方登錄、支付、短信、商城等功能

改造

我們改造的目的是實(shí)現一個(gè)starter，以后只要引入這個(gè)starter就能實(shí)現executor和jobHandler的自動(dòng)注冊，要引入的關(guān)鍵依賴(lài)有下面兩個(gè)：

com.xuxueli
xxl-job-core
2.3.0



org.springframework.boot
spring-boot-autoconfigure

1、接口調用

在調用調度中心的接口前，先把xxl-job-admin模塊中的XxlJobInfo和XxlJobGroup這兩個(gè)類(lèi)拿到我們的starter項目中，用于接收接口調用的結果。

登錄接口

創(chuàng )建一個(gè)JobLoginService，在調用業(yè)務(wù)接口前，需要通過(guò)登錄接口獲取cookie，并在獲取到cookie后，緩存到本地的Map中。

privatefinalMaploginCookie=newHashMap<>();

publicvoidlogin(){
Stringurl=adminAddresses+"/login";
HttpResponseresponse=HttpRequest.post(url)
.form("userName",username)
.form("password",password)
.execute();
Listcookies=response.getCookies();
OptionalcookieOpt=cookies.stream()
.filter(cookie->cookie.getName().equals("XXL_JOB_LOGIN_IDENTITY")).findFirst();
if(!cookieOpt.isPresent())
thrownewRuntimeException("getxxl-jobcookieerror!");

Stringvalue=cookieOpt.get().getValue();
loginCookie.put("XXL_JOB_LOGIN_IDENTITY",value);
}

其他接口在調用時(shí)，直接從緩存中獲取cookie，如果緩存中不存在則調用/login接口，為了避免這一過(guò)程失敗，允許最多重試3次。

publicStringgetCookie(){
for(inti=0;i

	

	執行器接口

	創(chuàng  )建一個(gè)JobGroupService，根據appName和執行器名稱(chēng)title查詢(xún)執行器列表：

	
publicListgetJobGroup(){
Stringurl=adminAddresses+"/jobgroup/pageList";
HttpResponseresponse=HttpRequest.post(url)
.form("appname",appName)
.form("title",title)
.cookie(jobLoginService.getCookie())
.execute();

Stringbody=response.body();
JSONArrayarray=JSONUtil.parse(body).getByPath("data",JSONArray.class);
Listlist=array.stream()
.map(o->JSONUtil.toBean((JSONObject)o,XxlJobGroup.class))
.collect(Collectors.toList());
returnlist;
}


	

	我們在后面要根據配置文件中的appName和title判斷當前執行器是否已經(jīng)被注冊到調度中心過(guò)，如果已經(jīng)注冊過(guò)那么則跳過(guò)，而/jobgroup/pageList接口是一個(gè)模糊查詢(xún)接口，所以在查詢(xún)列表的結果列表中，還需要再進(jìn)行一次精確匹配。

	
publicbooleanpreciselyCheck(){
ListjobGroup=getJobGroup();
Optionalhas=jobGroup.stream()
.filter(xxlJobGroup->xxlJobGroup.getAppname().equals(appName)
&&xxlJobGroup.getTitle().equals(title))
.findAny();
returnhas.isPresent();
}


	

	注冊新executor到調度中心：

	
publicbooleanautoRegisterGroup(){
Stringurl=adminAddresses+"/jobgroup/save";
HttpResponseresponse=HttpRequest.post(url)
.form("appname",appName)
.form("title",title)
.cookie(jobLoginService.getCookie())
.execute();
Objectcode=JSONUtil.parse(response.body()).getByPath("code");
returncode.equals(200);
}


	

	任務(wù)接口

	創(chuàng  )建一個(gè)JobInfoService，根據執行器id，jobHandler名稱(chēng)查詢(xún)任務(wù)列表，和上面一樣，也是模糊查詢(xún)：

	
publicListgetJobInfo(IntegerjobGroupId,StringexecutorHandler){
Stringurl=adminAddresses+"/jobinfo/pageList";
HttpResponseresponse=HttpRequest.post(url)
.form("jobGroup",jobGroupId)
.form("executorHandler",executorHandler)
.form("triggerStatus",-1)
.cookie(jobLoginService.getCookie())
.execute();

Stringbody=response.body();
JSONArrayarray=JSONUtil.parse(body).getByPath("data",JSONArray.class);
Listlist=array.stream()
.map(o->JSONUtil.toBean((JSONObject)o,XxlJobInfo.class))
.collect(Collectors.toList());

returnlist;
}


	

	注冊一個(gè)新任務(wù)，最終返回創(chuàng  )建的新任務(wù)的id：

	
publicIntegeraddJobInfo(XxlJobInfoxxlJobInfo){
Stringurl=adminAddresses+"/jobinfo/add";
MapparamMap=BeanUtil.beanToMap(xxlJobInfo);
HttpResponseresponse=HttpRequest.post(url)
.form(paramMap)
.cookie(jobLoginService.getCookie())
.execute();

JSONjson=JSONUtil.parse(response.body());
Objectcode=json.getByPath("code");
if(code.equals(200)){
returnConvert.toInt(json.getByPath("content"));
}
thrownewRuntimeException("addjobInfoerror!");
}


	

	2、創(chuàng  )建新注解

	在創(chuàng  )建任務(wù)時(shí)，必填字段除了執行器和jobHandler之外，還有任務(wù)描述 、負責人 、Cron表達式 、調度類(lèi)型 、運行模式 。在這里，我們默認調度類(lèi)型為CRON、運行模式為BEAN，另外的3個(gè)字段的信息需要用戶(hù)指定。

	因此我們需要創(chuàng  )建一個(gè)新注解@XxlRegister，來(lái)配合原生的@XxlJob注解進(jìn)行使用，填寫(xiě)這幾個(gè)字段的信息：

	
@Target({ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
public@interfaceXxlRegister{
Stringcron();
StringjobDesc()default"defaultjobDesc";
Stringauthor()default"defaultAuthor";
inttriggerStatus()default0;
}


	

	最后，額外添加了一個(gè)triggerStatus屬性，表示任務(wù)的默認調度狀態(tài)，0為停止狀態(tài)，1為運行狀態(tài)。

	3、自動(dòng)注冊核心

	基本準備工作做完后，下面實(shí)現自動(dòng)注冊執行器和jobHandler的核心代碼。核心類(lèi)實(shí)現ApplicationListener接口，在接收到ApplicationReadyEvent事件后開(kāi)始執行自動(dòng)注冊邏輯。

	
@Component
publicclassXxlJobAutoRegisterimplementsApplicationListener,
ApplicationContextAware{
privatestaticfinalLoglog=LogFactory.get();
privateApplicationContextapplicationContext;
@Autowired
privateJobGroupServicejobGroupService;
@Autowired
privateJobInfoServicejobInfoService;

@Override
publicvoidsetApplicationContext(ApplicationContextapplicationContext)throwsBeansException{
this.applicationContext=applicationContext;
}

@Override
publicvoidonApplicationEvent(ApplicationReadyEventevent){
addJobGroup();//注冊執行器
addJobInfo();//注冊任務(wù)
}
}


	

	自動(dòng)注冊執行器的代碼非常簡(jiǎn)單，根據配置文件中的appName和title精確匹配查看調度中心是否已有執行器被注冊過(guò)了，如果存在則跳過(guò)，不存在則新注冊一個(gè)：

	
privatevoidaddJobGroup(){
if(jobGroupService.preciselyCheck())
return;

if(jobGroupService.autoRegisterGroup())
log.info("autoregisterxxl-jobgroupsuccess!");
}


	

	自動(dòng)注冊任務(wù)的邏輯則相對復雜一些，需要完成：

	通過(guò)applicationContext拿到spring容器中的所有bean，再拿到這些bean中所有添加了@XxlJob注解的方法

	對上面獲取到的方法進(jìn)行檢查，是否添加了我們自定義的@XxlRegister注解，如果沒(méi)有則跳過(guò)，不進(jìn)行自動(dòng)注冊

	對同時(shí)添加了@XxlJob和@XxlRegister的方法，通過(guò)執行器id和jobHandler的值判斷是否已經(jīng)在調度中心注冊過(guò)了，如果已存在則跳過(guò)

	對于滿(mǎn)足注解條件且沒(méi)有注冊過(guò)的jobHandler，調用接口注冊到調度中心

	具體代碼如下：

	
privatevoidaddJobInfo(){
ListjobGroups=jobGroupService.getJobGroup();
XxlJobGroupxxlJobGroup=jobGroups.get(0);

String[]beanDefinitionNames=applicationContext.getBeanNamesForType(Object.class,false,true);
for(StringbeanDefinitionName:beanDefinitionNames){
Objectbean=applicationContext.getBean(beanDefinitionName);

MapannotatedMethods=MethodIntrospector.selectMethods(bean.getClass(),
newMethodIntrospector.MetadataLookup(){
@Override
publicXxlJobinspect(Methodmethod){
returnAnnotatedElementUtils.findMergedAnnotation(method,XxlJob.class);
}
});
for(Map.EntrymethodXxlJobEntry:annotatedMethods.entrySet()){
MethodexecuteMethod=methodXxlJobEntry.getKey();
XxlJobxxlJob=methodXxlJobEntry.getValue();

//自動(dòng)注冊
if(executeMethod.isAnnotationPresent(XxlRegister.class)){
XxlRegisterxxlRegister=executeMethod.getAnnotation(XxlRegister.class);
ListjobInfo=jobInfoService.getJobInfo(xxlJobGroup.getId(),xxlJob.value());
if(!jobInfo.isEmpty()){
//因為是模糊查詢(xún)，需要再判斷一次
Optionalfirst=jobInfo.stream()
.filter(xxlJobInfo->xxlJobInfo.getExecutorHandler().equals(xxlJob.value()))
.findFirst();
if(first.isPresent())
continue;
}

XxlJobInfoxxlJobInfo=createXxlJobInfo(xxlJobGroup,xxlJob,xxlRegister);
IntegerjobInfoId=jobInfoService.addJobInfo(xxlJobInfo);
}
}
}
}


	

	4、自動(dòng)裝配

	創(chuàng  )建一個(gè)配置類(lèi)，用于掃描bean：

	
@Configuration
@ComponentScan(basePackages="com.xxl.job.plus.executor")
publicclassXxlJobPlusConfig{
}


	

	將它添加到META-INF/spring.factories文件：

	
org.springframework.boot.autoconfigure.EnableAutoConfiguration=
com.xxl.job.plus.executor.config.XxlJobPlusConfig


	

	到這里starter的編寫(xiě)就完成了，可以通過(guò)maven發(fā)布jar包到本地或者私服：

	
mvncleaninstall/deploy


	

	基于 Spring Cloud Alibaba + Gateway + Nacos + RocketMQ + Vue & Element 實(shí)現的后臺管理系統 + 用戶(hù)小程序，支持 RBAC 動(dòng)態(tài)權限、多租戶(hù)、數據權限、工作流、三方登錄、支付、短信、商城等功能

	項目地址：https://github.com/YunaiV/yudao-cloud

	視頻教程：https://doc.iocoder.cn/video/

	測試

	新建一個(gè)springboot項目，引入我們在上面打好的包：

	

com.cn.hydra
xxljob-autoregister-spring-boot-starter
0.0.1



	

	在application.properties中配置xxl-job的信息，首先是原生的配置內容：

	
xxl.job.admin.addresses=http://127.0.0.1:8080/xxl-job-admin
xxl.job.accessToken=default_token
xxl.job.executor.appname=xxl-job-executor-test
xxl.job.executor.address=
xxl.job.executor.ip=127.0.0.1
xxl.job.executor.port=9999
xxl.job.executor.logpath=/data/applogs/xxl-job/jobhandler
xxl.job.executor.logretentiondays=30


	

	此外還要額外添加我們自己的starter要求的新配置內容：

	
#admin用戶(hù)名
xxl.job.admin.username=admin
#admin密碼
xxl.job.admin.password=123456
#執行器名稱(chēng)
xxl.job.executor.title=test-title


	

	完成后在代碼中配置一下XxlJobSpringExecutor，然后在測試接口上添加原生@XxlJob注解和我們自定義的@XxlRegister注解：

	
@XxlJob(value="testJob")
@XxlRegister(cron="000**?*",
author="hydra",
jobDesc="測試job")
publicvoidtestJob(){
System.out.println("#碼農參上");
}


@XxlJob(value="testJob222")
@XxlRegister(cron="591-20**?",
triggerStatus=1)
publicvoidtestJob2(){
System.out.println("#作者：Hydra");
}

@XxlJob(value="testJob444")
@XxlRegister(cron="595923**?")
publicvoidtestJob4(){
System.out.println("helloxxljob");
}


	

	啟動(dòng)項目，可以看到執行器自動(dòng)注冊成功：

	

	再打開(kāi)調度中心的任務(wù)管理頁(yè)面，可以看到同時(shí)添加了兩個(gè)注解的任務(wù)也已經(jīng)自動(dòng)完成了注冊：

	

	從頁(yè)面上手動(dòng)執行任務(wù)進(jìn)行測試，可以執行成功：

	

	到這里，starter的編寫(xiě)和測試過(guò)程就算基本完成了，項目中引入后，以后也能省出更多的時(shí)間來(lái)摸魚(yú)學(xué)習了~

	






	審核編輯：劉清