如果我們的任務是比較大型的，比如；定時跑批T+1結算、商品秒殺前狀態變更、刷新數據預熱到緩存等等，這些定時任務都相同的特點；作業量大、實時性強、可用率高。而這時候如果只是單純使用Schedule就顯得不足以控制。
前言@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/dist/zookeeper}
- 下載后解壓，在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 - 打包部署控制平臺
- 下載地址：https://git***.com/fuzhengwei/itstack-middleware-control.git
- 部署訪問：http://localhost:7397
- 3. 配置POM
<dependency>
<groupId>org.itstack.middleware</groupId>
<artifactId>schedule-spring-boot-starter</artifactId>
<version>1.0.0-RELEASE</version>
</dependency>
- 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<ExecOrder> 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<String> beanNames = Constants.execOrderMap.keySet();
for (String beanName : beanNames) {
List<ExecOrder> 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<String> beanNames = Constants.execOrderMap.keySet();
for (String beanName : beanNames) {
List<ExecOrder> 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功能接口做展示和操作。
- 中間件開發是一件非常有意思的事情，不同于業務它更像易筋經，寺廟老僧，劍走偏鋒，馳騁縱橫，騷招滿屏。
