沉淀、分享、成長，讓自己和他人都能有所收獲！?

前言

@SpringBootApplication

@EnableScheduling

public class Application{

public static void mian(String[] args){

SpringApplication.run(Application.class,args);

}

@Scheduled(cron = "0/3 * * * * *")

public void demoTask() {

//...

}

}

咔咔，上面這段代碼很熟悉吧，他就是SpringBoot的Schedule定時任務，簡單易用。在我們開發中如果需要做一些定時或指定時刻循環執行邏輯時候，基本都會使用到Schedule。

但是，如果我們的任務是比較大型的，比如；定時跑批T+1結算、商品秒殺前狀態變更、刷新數據預熱到緩存等等，這些定時任務都相同的特點；作業量大、實時性強、可用率高。而這時候如果只是單純使用Schedule就顯得不足以控制。

那么，我們產品需求就出來了，分布式DcsSchedule任務；

多機器部署任務

統一控制中心啟停

宕機災備，自動啟動執行

實時檢測任務執行信息：部署數量、任務總量、成功次數、失敗次數、執行耗時等

嗯？有人憋半天了想說可以用Quertz，嗯可以的，但這不是本篇文章的重點。難道你不想看看一個自言開源中間件是怎么誕生的嗎，怎么推到中心Maven倉的嗎？比如下圖；真香不！

首頁監控

任務列表

?好了，接下來開始介紹這個中間件如何使用和怎么開發的了！

中間件使用

1. 版本記錄

版本

發布日期

備注

1

1.0.0-RELEASE

2019-12-07

基本功能實現；任務接入、分布式啟停

2

1.0.1-RELEASE

2019-12-07

上傳測試版本

2. 環境準備

jdk1.8

StringBoot 2.x

配置中心zookeeper 3.4.14 {準備好zookeeper服務，如果windows調試可以從這里下載：https://www-eu.apache.org/dis...}

下載后解壓，在bin同級路徑創建文件夾data、logs

修改conf/zoo.cfg，修改配置如下；

dataDir=D:\\Program Files\\apache-zookeeper-3.4.14\\data

dataLogDir=D:\\Program Files\\apache-zookeeper-3.4.14\\logs

3. 配置POM

org.itstack.middleware

schedule-spring-boot-starter

1.0.0-RELEASE

4. 引入分布式任務DcsSchedule @EnableDcsScheduling

與SpringBoot的Sceduling非常像，他的注解是；@EnableScheduling，盡可能降低使用難度

這個注解主要方便給我們自己的中間件一個入口，也是?扒拉源碼發現的可以這么干{我一直說好的代碼都很騷氣}

@SpringBootApplication

@EnableDcsScheduling

public class HelloWorldApplication {

public static void main(String[] args) {

SpringApplication.run(HelloWorldApplication.class, args);

}

}

5. 在任務方法上添加注解

這個注解也和SpringBoot的Schedule很像，但是多了desc描述和啟停初始化控制

cron：執行計劃

desc：任務描述

autoStartup：默認啟動狀態

如果你的任務需要參數可以通過引入service去調用獲取等方式都可以

@Component("demoTaskThree")

public class DemoTaskThree {

@DcsScheduled(cron = "0 0 9,13 * * *", desc = "03定時任務執行測試：taskMethod01", autoStartup = false)

public void taskMethod01() {

System.out.println("03定時任務執行測試：taskMethod01");

}

@DcsScheduled(cron = "0 0/30 8-10 * * *", desc = "03定時任務執行測試：taskMethod02", autoStartup = false)

public void taskMethod02() {

System.out.println("03定時任務執行測試：taskMethod02");

}

}

6. 啟動驗證

啟動SpringBoot工程即可，autoStartup = true的會自動啟動任務(任務是多線程并行執行的)

啟動控制平臺：itstack-middleware-control，訪問：http://localhost:7397/ 成功界面如下；可以開啟/關閉驗證了！{功能還在完善}

中間件開發

以SpringBoot為基礎開發一款中間件我也是第一次，因為接觸SpringBoot也剛剛1個月左右。雖然SpringBoot已經出來挺久的了，但由于我們項目開發并不使用SpringBoot的一套東西，所以一直依賴沒有接觸。直到上個月開始考慮領域驅動設計才接觸，嗯！真的不錯，那么就開始了夯實技能、學習思想用到項目里。

按照我的產品需求，開發這么一款分布式任務的中間件，我腦袋中的模型已經存在了。另外就是需要開發過程中去探索我需要的知識工具，簡單包括；

讀取Yml自定義配置

使用zookeeper作為配置中心，這樣如果有機器宕機了就可以通過臨時節點監聽知道

通過Spring類；ApplicationContextAware, BeanPostProcessor, ApplicationListener，執行服務啟動、注解掃描、節點掛在

分布式任務統一控制臺，來管理任務

1. 工程模型

schedule-spring-boot-starter

└── src

├── main

│ ├── java

│ │ └── org.itstack.middleware.schedule

│ │ ├── annotation

│ │ │ ├── DcsScheduled.java

│ │ │ └── EnableDcsScheduling.java

│ │ ├── annotation

│ │ │ └── InstructStatus.java

│ │ ├── config

│ │ │ ├── DcsSchedulingConfiguration.java

│ │ │ ├── StarterAutoConfig.java

│ │ │ └── StarterServiceProperties.java

│ │ ├── domain

│ │ │ ├── DataCollect.java

│ │ │ ├── DcsScheduleInfo.java

│ │ │ ├── DcsServerNode.java

│ │ │ ├── ExecOrder.java

│ │ │ └── Instruct.java

│ │ ├── export

│ │ │ └── DcsScheduleResource.java

│ │ ├── service

│ │ │ ├── HeartbeatService.java

│ │ │ └── ZkCuratorServer.java

│ │ ├── task

│ │ │ ├── TaskScheduler.java

│ │ │ ├── ScheduledTask.java

│ │ │ ├── SchedulingConfig.java

│ │ │ └── SchedulingRunnable.java

│ │ ├── util

│ │ │ └── StrUtil.java

│ │ └── DoJoinPoint.java

│ └── resources

│ └── META_INF

│ └── spring.factories

└── test

└── java

└── org.itstack.demo.test

└── ApiTest.java

2. 代碼講解

篇幅較長，只講解部分重點代碼塊，如果你愿意參與到開源編寫，可以和我申請

我說過好的代碼都很騷氣，那么就從這部分入手吧

2.1 自定義注解

annotation/EnableDcsScheduling.java & 自定義注解

這個注解一堆的圈A，這些配置都是為了開始啟動執行我們的中間件；

Target 標識需要放到類上執行

Retention 注釋將由編譯器記錄在類文件中，并且在運行時由VM保留，因此可以反射地讀取它們

Import 引入入口資源，在程序啟動時會執行到自己定義的類中，以方便我們；初始化配置/服務、啟動任務、掛在節點

ComponentScan 告訴程序掃描位置

@Target({ElementType.TYPE})

@Retention(RetentionPolicy.RUNTIME)

@Import({DcsSchedulingConfiguration.class})

@ImportAutoConfiguration({SchedulingConfig.class, CronTaskRegister.class, DoJoinPoint.class})

@ComponentScan("org.itstack.middleware.*")

public @interface EnableDcsScheduling {

}

2.2 掃描自定義注解、初始化配置/服務、啟動任務、掛在節點

config/DcsSchedulingConfiguration.java & 初始化配置/服務、啟動任務、掛在節點

寫到這的時候，我們的自定義注解有了，已經寫到方法上了，那么我們怎么拿到呢？

需要通過實現BeanPostProcessor.postProcessAfterInitialization，在每個bean實例化的時候進行掃描

這里遇到一個有趣的問題，一個方法會得到兩次，因為有一個CGLIB給代理的，像真假美猴王一樣，幾乎一毛一樣。?扒了源碼才看到，生命注解批注沒有。好那就可以判斷了！method.getDeclaredAnnotations()

我們將掃描下來的任務信息匯總到Map中，當Spring初始化完成后，在執行我們中間件內容。{太早執行有點喧賓奪主了！主要人家也不讓呀，給你拋異常?。}

@Override

public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {

Class> targetClass = AopProxyUtils.ultimateTargetClass(bean);

if (this.nonAnnotatedClasses.contains(targetClass)) return bean;

Method[] methods = ReflectionUtils.getAllDeclaredMethods(bean.getClass());

if (methods == null) return bean;

for (Method method : methods) {

DcsScheduled dcsScheduled = AnnotationUtils.findAnnotation(method, DcsScheduled.class);

if (null == dcsScheduled || 0 == method.getDeclaredAnnotations().length) continue;

List execOrderList = Constants.execOrderMap.computeIfAbsent(beanName, k -> new ArrayList<>());

ExecOrder execOrder = new ExecOrder();

execOrder.setBean(bean);

execOrder.setBeanName(beanName);

execOrder.setMethodName(method.getName());

execOrder.setDesc(dcsScheduled.desc());

execOrder.setCron(dcsScheduled.cron());

execOrder.setAutoStartup(dcsScheduled.autoStartup());

execOrderList.add(execOrder);

this.nonAnnotatedClasses.add(targetClass);

}

return bean;

}

初始化服務連接zookeeper配置中心

連接后將創建我們的節點以及添加監聽，這個監聽主要負責分布式消息通知，收到通知負責控制任務啟停

這里包括了循環創建節點以及批量節點刪除，似乎！面試題會問?

private void init_server(ApplicationContext applicationContext) {

try {

//獲取zk連接

CuratorFramework client = ZkCuratorServer.getClient(Constants.Global.zkAddress);

//節點組裝

path_root_server = StrUtil.joinStr(path_root, LINE, "server", LINE, schedulerServerId);

path_root_server_ip = StrUtil.joinStr(path_root_server, LINE, "ip", LINE, Constants.Global.ip);

//創建節點&遞歸刪除本服務IP下的舊內容

ZkCuratorServer.deletingChildrenIfNeeded(client, path_root_server_ip);

ZkCuratorServer.createNode(client, path_root_server_ip);

ZkCuratorServer.setData(client, path_root_server, schedulerServerName);

//添加節點&監聽

ZkCuratorServer.createNodeSimple(client, Constants.Global.path_root_exec);

ZkCuratorServer.addTreeCacheListener(applicationContext, client, Constants.Global.path_root_exec);

} catch (Exception e) {

logger.error("itstack middleware schedule init server error！", e);

throw new RuntimeException(e);

}

}

啟動標記了True的Schedule任務

Scheduled默認是單線程執行的，這里擴展為多線程并行執行

private void init_task(ApplicationContext applicationContext) {

CronTaskRegister cronTaskRegistrar = applicationContext.getBean("itstack-middlware-schedule-cronTaskRegister", CronTaskRegister.class);

Set beanNames = Constants.execOrderMap.keySet();

for (String beanName : beanNames) {

List execOrderList = Constants.execOrderMap.get(beanName);

for (ExecOrder execOrder : execOrderList) {

if (!execOrder.getAutoStartup()) continue;

SchedulingRunnable task = new SchedulingRunnable(execOrder.getBean(), execOrder.getBeanName(), execOrder.getMethodName());

cronTaskRegistrar.addCronTask(task, execOrder.getCron());

}

}

}

掛在任務節點到zookeeper掛在

按照不同的場景，有些內容是掛在到虛擬機節點。{?又來個面試題，虛擬節點數據怎么掛在，創建的是永久節點，那么虛擬值怎么加？}

path_root_server_ip_clazz_method；這個結構是：根目錄、服務、IP、類、方法

private void init_node() throws Exception {

Set beanNames = Constants.execOrderMap.keySet();

for (String beanName : beanNames) {

List execOrderList = Constants.execOrderMap.get(beanName);

for (ExecOrder execOrder : execOrderList) {

String path_root_server_ip_clazz = StrUtil.joinStr(path_root_server_ip, LINE, "clazz", LINE, execOrder.getBeanName());

String path_root_server_ip_clazz_method = StrUtil.joinStr(path_root_server_ip_clazz, LINE, "method", LINE, execOrder.getMethodName());

String path_root_server_ip_clazz_method_status = StrUtil.joinStr(path_root_server_ip_clazz, LINE, "method", LINE, execOrder.getMethodName(), "/status");

//添加節點

ZkCuratorServer.createNodeSimple(client, path_root_server_ip_clazz);

ZkCuratorServer.createNodeSimple(client, path_root_server_ip_clazz_method);

ZkCuratorServer.createNodeSimple(client, path_root_server_ip_clazz_method_status);

//添加節點數據[臨時]

ZkCuratorServer.appendPersistentData(client, path_root_server_ip_clazz_method + "/value", JSON.toJSONString(execOrder));

//添加節點數據[永久]

ZkCuratorServer.setData(client, path_root_server_ip_clazz_method_status, execOrder.getAutoStartup() ? "1" : "0");

}

}

}

2.3 zookeeper控制服務

service/ZkCuratorServer.java & zk服務

這里提供一個zk的方法集合，其中比較重要的方法添加監聽

zookeeper有一個特性是對這個監聽后，當節點內容發生變化時會收到通知，當然宕機也是收得到的，這個也就是我們后面開發災備的核心觸發點

public static void addTreeCacheListener(final ApplicationContext applicationContext, final CuratorFramework client, String path) throws Exception {

TreeCache treeCache = new TreeCache(client, path);

treeCache.start();

treeCache.getListenable().addListener((curatorFramework, event) -> {

//...

switch (event.getType()) {

case NODE_ADDED:

case NODE_UPDATED:

if (Constants.Global.ip.equals(instruct.getIp()) && Constants.Global.schedulerServerId.equals(instruct.getSchedulerServerId())) {

//執行命令

Integer status = instruct.getStatus();

switch (status) {

case 0: //停止任務

cronTaskRegistrar.removeCronTask(instruct.getBeanName() + "_" + instruct.getMethodName());

setData(client, path_root_server_ip_clazz_method_status, "0");

logger.info("itstack middleware schedule task stop {} {}", instruct.getBeanName(), instruct.getMethodName());

break;

case 1: //啟動任務

cronTaskRegistrar.addCronTask(new SchedulingRunnable(scheduleBean, instruct.getBeanName(), instruct.getMethodName()), instruct.getCron());

setData(client, path_root_server_ip_clazz_method_status, "1");

logger.info("itstack middleware schedule task start {} {}", instruct.getBeanName(), instruct.getMethodName());

break;

case 2: //刷新任務

cronTaskRegistrar.removeCronTask(instruct.getBeanName() + "_" + instruct.getMethodName());

cronTaskRegistrar.addCronTask(new SchedulingRunnable(scheduleBean, instruct.getBeanName(), instruct.getMethodName()), instruct.getCron());

setData(client, path_root_server_ip_clazz_method_status, "1");

logger.info("itstack middleware schedule task refresh {} {}", instruct.getBeanName(), instruct.getMethodName());

break;

}

}

break;

case NODE_REMOVED:

break;

default:

break;

}

});

}

2.4 并行任務注冊

由于默認的SpringBoot是單線程的，所以這里改造了下，可以支持多線程并行執行

包括了添加任務和刪除任務，也就是執行取消future.cancel(true)

public void addCronTask(SchedulingRunnable task, String cronExpression) {

if (null != Constants.scheduledTasks.get(task.taskId())) {

removeCronTask(task.taskId());

}

CronTask cronTask = new CronTask(task, cronExpression);

Constants.scheduledTasks.put(task.taskId(), scheduleCronTask(cronTask));

}

public void removeCronTask(String taskId) {

ScheduledTask scheduledTask = Constants.scheduledTasks.remove(taskId);

if (scheduledTask == null) return;

scheduledTask.cancel();

}

2.5 待擴展的自定義AOP

我們最開始配置的掃描@ComponentScan("org.itstack.middleware.*")，主要用到這里的自定義注解，否則是掃描不到的，也就是你自定義切面失效的效果

目前這里的功能并沒有擴展，基本只是打印執行耗時，后續完善的任務執行耗時監聽等，就需要這里來完善

@Pointcut("@annotation(org.itstack.middleware.schedule.annotation.DcsScheduled)")

public void aopPoint() {

}

@Around("aopPoint()")

public Object doRouter(ProceedingJoinPoint jp) throws Throwable {

long begin = System.currentTimeMillis();

Method method = getMethod(jp);

try {

return jp.proceed();

} finally {

long end = System.currentTimeMillis();

logger.info("\nitstack middleware schedule method：{}.{} take time(m)：{}", jp.getTarget().getClass().getSimpleName(), method.getName(), (end - begin));

}

}

3. Jar包發布

開發完成后還是需要將Jar包發布到manven中心倉庫的，這個過程較長單獨寫了博客；發布Jar包到Maven中央倉庫(為開發開源中間件做準備)

綜上總結

要開發要實現的還很多，一個周末也干不完所有的！而且需要有想法的小猿/媛伴一起加入！? ? ?

這里沒有講解分布式任務中間件控制平臺itstack-middleware-control，因為比較簡單只是使用了中間件的zk功能接口做展示和操作。

中間件開發是一件非常有意思的事情，不同于業務它更像易筋經，寺廟老僧，劍走偏鋒，馳騁縱橫，騷招滿屏。

推薦閱讀

《新程序員》：云原生和全面數字化實踐50位技術專家共同創作，文字、視頻、音頻交互閱讀

總結

以上是生活随笔為你收集整理的怎样用springboot开发cs_SpringBoot分布式任务中间件开发 附视频讲解 (手把手教你开发和使用中间件)...的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。