文档

2025-12-24 03:42:07 +08:00 · 2022-09-02 17:19:44 +08:00
parent 639b1f8336
commit fcefe7ac38
1 changed files with 250 additions and 0 deletions
--- a/docs/dev_guide/KnowStreaming
+++ b/docs/dev_guide/KnowStreaming
@@ -0,0 +1,250 @@
+# KnowStreaming Task模块简介
+
+## 1、Task简介
+
+在 KnowStreaming 中(下面简称KS)，Task模块主要是用于执行一些周期任务，包括Cluster、Broker、Topic等指标的定时采集，集群元数据定时更新至DB，集群状态的健康巡检等。在KS中，与Task模块相关的代码，我们都统一存放在km-task模块中。
+
+Task模块是基于 LogiCommon 中的Logi-Job组件实现的任务周期执行，Logi-Job 的功能类似 XXX-Job，它是 XXX-Job 在 KnowStreaming 的内嵌实现，主要用于简化 KnowStreaming 的部署。
+Logi-Job 的任务总共有两种执行模式，分别是：
+
+ 广播模式：同一KS集群下，同一任务周期中，所有KS主机都会执行该定时任务。
+ 抢占模式：同一KS集群下，同一任务周期中，仅有某一台KS主机会执行该任务。
+
+KS集群范围定义：连接同一个DB，且application.yml中的spring.logi-job.app-name的名称一样的KS主机为同一KS集群。
+
+## 2、使用指南
+
+Task模块基于Logi-Job的广播模式与抢占模式，分别实现了任务的抢占执行、重复执行以及均衡执行，他们之间的差别是：
+
+ 抢占执行：同一个KS集群，同一个任务执行周期中，仅有一台KS主机执行该任务；
+ 重复执行：同一个KS集群，同一个任务执行周期中，所有KS主机都执行该任务。比如3台KS主机，3个Kafka集群，此时每台KS主机都会去采集这3个Kafka集群的指标；
+ 均衡执行：同一个KS集群，同一个任务执行周期中，每台KS主机仅执行该任务的一部分，所有的KS主机共同协作完成了任务。比如3台KS主机，3个Kafka集群，稳定运行情况下，每台KS主机将仅采集1个Kafka集群的指标，3台KS主机共同完成3个Kafka集群指标的采集。
+
+下面我们看一下具体例子。
+
+### 2.1、抢占模式——抢占执行
+
+功能说明：
+
+ 同一个KS集群，同一个任务执行周期中，仅有一台KS主机执行该任务。
+
+代码例子：
+
+```java
+// 1、实现Job接口，重写excute方法； 
+// 2、在类上添加@Task注解，并且配置好信息，指定为随机抢占模式； 
+// 效果：KS集群中，每5秒，会有一台KS主机输出 "测试定时任务运行中"；
+@Task(name = "TestJob",
+      description = "测试定时任务",
+      cron = "*/5 * * * * ?",
+      autoRegister = true,
+      consensual = ConsensualEnum.RANDOM, // 这里一定要设置为RANDOM        
+      timeout = 6 * 60) 
+public class TestJob implements Job {   
+  @Override    
+  public TaskResult execute(JobContext jobContext) throws Exception {        
+    System.out.println("测试定时任务运行中");        
+    return new TaskResult();    
+  } 
+}
+```
+
+
+
+### 2.2、广播模式——重复执行
+
+功能说明：
+
+ 同一个KS集群，同一个任务执行周期中，所有KS主机都执行该任务。比如3台KS主机，3个Kafka集群，此时每台KS主机都会去重复采集这3个Kafka集群的指标。 
+
+代码例子：
+
+```java
+// 1、实现Job接口，重写excute方法； 
+// 2、在类上添加@Task注解，并且配置好信息，指定为广播抢占模式； 
+// 效果：KS集群中，每5秒，每台KS主机都会输出 "测试定时任务运行中"； 
+@Task(name = "TestJob",        
+			description = "测试定时任务",
+      cron = "*/5 * * * * ?",
+      autoRegister = true,
+      consensual = ConsensualEnum.BROADCAST, // 这里一定要设置为BROADCAST
+      timeout = 6 * 60) 
+public class TestJob implements Job {    
+  @Override    
+  public TaskResult execute(JobContext jobContext) throws Exception {        
+    System.out.println("测试定时任务运行中");        
+    return new TaskResult();    
+    } 
+}
+```
+
+
+
+### 2.3、广播模式——均衡执行
+
+功能说明：
+
+ 同一个KS集群，同一个任务执行周期中，每台KS主机仅执行该任务的一部分，所有的KS主机共同协作完成了任务。比如3台KS主机，3个Kafka集群，稳定运行情况下，每台KS主机将仅采集1个Kafka集群的指标，3台KS主机共同完成3个Kafka集群指标的采集。
+
+代码例子：
+
+ 该模式有点特殊，是KS基于Logi-Job的广播模式，做的一个扩展，以下为一个使用例子：
+
+```java
+// 1、继承AbstractClusterPhyDispatchTask，实现processSubTask方法； 
+// 2、在类上添加@Task注解，并且配置好信息，指定为广播模式； 
+// 效果：在本样例中，每隔1分钟ks会将所有的kafka集群列表在ks集群主机内均衡拆分，每台主机会将分发到自身的Kafka集群依次执行processSubTask方法,实现KS集群的任务协同处理。 
+@Task(name = "kmJobTask",
+			description = "km job 模块调度执行任务",
+      cron = "0 0/1 * * * ? *",
+      autoRegister = true,
+      consensual = ConsensualEnum.BROADCAST,
+      timeout = 6 * 60) 
+public class KMJobTask extends AbstractClusterPhyDispatchTask {   
+  
+    @Autowired    
+    private JobService jobService;     
+  
+    @Override    
+    protected TaskResult processSubTask(ClusterPhy clusterPhy, long triggerTimeUnitMs) throws Exception {
+          jobService.scheduleJobByClusterId(clusterPhy.getId());        
+          return TaskResult.SUCCESS;    
+    }
+}
+```
+
+
+
+## 3、原理简介
+
+### 3.1、Task注解说明
+
+```java
+public @interface Task {
+    String name() default ""; //任务名称
+    String description() default ""; //任务描述
+    String owner() default "system";  //拥有者     
+    String cron() default ""; //定时执行的时间策略     
+    int retryTimes() default 0; //失败以后所能重试的最大次数 
+    long timeout() default 0; //在超时时间里重试  
+    //是否自动注册任务到数据库中  
+    //如果设置为false，需要手动去数据库km_task表注册定时任务信息。数据库记录和@Task注解缺一不可
+    boolean autoRegister() default false;   
+    //执行模式：广播、随机抢占  
+    //广播模式：同一集群下的所有服务器都会执行该定时任务  
+    //随机抢占模式：同一集群下随机一台服务器执行该任务   
+    ConsensualEnum consensual() default ConsensualEnum.RANDOM; 
+  }
+```
+
+### 3.2、数据库表介绍
+
+ logi_task：记录项目中的定时任务信息，一个定时任务对应一条记录。
+ logi_job：具体任务执行信息。
+ logi_job_log：定时任务的执行日志。
+ logi_worker：记录机器信息，实现集群控制。
+
+### 3.3、均衡执行简介
+
+#### 3.3.1、类关系图
+
+这里以KMJobTask为例，简单介绍KM中的定时任务实现逻辑。
+
+        ![img](https://cooper.didichuxing.com/cooper_gateway/cn/shimo-images/fXd8Bfj4tPEcdOcT/image.png)      
+
+ Job：使用logi组件实现定时任务，必须实现该接口。
+ Comparable & EntufyIdInterface：比较接口，实现任务的排序逻辑。
+ AbstractDispatchTask：实现广播模式下，任务的均衡分发。
+ AbstractClusterPhyDispatchTask：对分发到当前服务器的集群列表进行枚举。
+ KMJobTask:实现对单个集群的定时任务处理。
+
+#### 3.3.2、关键类代码
+
+ **AbstractDispatchTask类**
+
+```java
+// 实现Job接口的抽象类，进行任务的负载均衡执行 
+public abstract class AbstractDispatchTask<E extends Comparable & EntifyIdInterface> implements Job { 
+  
+  // 罗列所有的任务    
+  protected abstract List<E> listAllTasks(); 
+  
+  // 执行被分配给该KS主机的任务    
+  protected abstract TaskResult processTask(List<E> subTaskList, long triggerTimeUnitMs);   
+  
+  // 被Logi-Job触发执行该方法    
+  // 该方法进行任务的分配    
+  @Override    
+  public TaskResult execute(JobContext jobContext) {        
+    try {            
+      
+      long triggerTimeUnitMs = System.currentTimeMillis();            
+      // 获取所有的任务            
+      List<E> allTaskList = this.listAllTasks();
+      
+      // 计算当前KS机器需要执行的任务            
+      List<E> subTaskList = this.selectTask(allTaskList, jobContext.getAllWorkerCodes(), jobContext.getCurrentWorkerCode()); 
+      
+      // 进行任务处理           
+      return this.processTask(subTaskList, triggerTimeUnitMs);        
+    } catch (Exception e) {           
+      // ...       
+    }  
+  }
+}
+```
+
+ **AbstractClusterPhyDispatchTask类**
+
+```java
+// 继承AbstractDispatchTask的抽象类，对Kafka集群进行负载均衡执行 
+public abstract class AbstractClusterPhyDispatchTask extends AbstractDispatchTask<ClusterPhy> {   
+  // 执行被分配的任务，具体由子类实现    
+  protected abstract TaskResult processSubTask(ClusterPhy clusterPhy, long triggerTimeUnitMs) throws Exception;     	// 返回所有的Kafka集群   
+  @Override    
+  public List<ClusterPhy> listAllTasks() {        
+    return clusterPhyService.listAllClusters();   
+  }     
+  // 执行被分配给该KS主机的Kafka集群任务   
+  @Override   
+  public TaskResult processTask(List<ClusterPhy> subTaskList, long triggerTimeUnitMs) {        // ...     }
+}   
+```
+
+ **KMJobTask类**
+
+```java
+// 加上@Task注解，并配置任务执行信息 
+@Task(name = "kmJobTask",   
+			description = "km job 模块调度执行任务", 
+      cron = "0 0/1 * * * ? *",   
+      autoRegister = true,    
+      consensual = ConsensualEnum.BROADCAST,  
+      timeout = 6 * 60)
+// 继承AbstractClusterPhyDispatchTask类 
+public class KMJobTask extends AbstractClusterPhyDispatchTask {  
+
+				@Autowired   
+        private JobService jobService;  
+        
+        // 执行该Kafka集群的Job模块的任务  
+        @Override  
+        protected TaskResult processSubTask(ClusterPhy clusterPhy, long triggerTimeUnitMs) throws Exception {        
+          jobService.scheduleJobByClusterId(clusterPhy.getId());   
+          return TaskResult.SUCCESS;  
+        }
+}
+```
+
+#### 3.3.3、均衡执行总结
+
+均衡执行的实现原理总结起来就是以下几点：
+
+ Logi-Job设置为广播模式，触发所有的KS主机执行任务；
+ 每台KS主机，被触发执行后，按照统一的规则，对任务列表，KS集群主机列表进行排序。然后按照顺序将任务列表均衡的分配给排序后的KS集群主机。KS集群稳定运行情况下，这一步保证了每台KS主机之间分配到的任务列表不重复，不丢失。
+ 最后每台KS主机，执行被分配到的任务。
+
+## 4、注意事项
+
+ 不能100%保证任务在一个周期内，且仅且执行一次，可能出现重复执行或丢失的情况，所以必须严格是且仅且执行一次的任务，不建议基于Logi-Job进行任务控制。
+ 尽量让Logi-Job仅负责任务的触发，后续的执行建议放到自己创建的线程池中进行。