Task模块简介

// 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();    
    
  } 
  
}

// 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();    
    
    } 
    
}

// 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;    
    }
}

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; 
  }

// 实现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) {           
      // ...       
    }  
  }
}

// 继承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) {        // ...     }
  
}   

// 加上@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;  
        }
}