TBSchedule源码阅读2-TBScheduleManagerFactory 定时任务ManagerFactoryTimerTask

定时任务 ：

主要功能：监听zookeeper状态，正常则this.factory.refresh()，异常则this.factory.reStart();

1 正常情况this.factory.refresh()

异常则停止所有调度任务
正常则重新分配调度器

1.1分配调度器

    public void reRegisterManagerFactory() throws Exception{
        //重新分配调度器
        List<String> stopList = this.getScheduleStrategyManager().registerManagerFactory(this);
        for (String strategyName : stopList) {
            this.stopServer(strategyName);
        }
        this.assignScheduleServer();
        this.reRunScheduleServer();
    }

1.1.1 this.assignScheduleServer();

    /**
     * 根据策略重新分配调度任务的机器
     * @throws Exception
     */
    public void assignScheduleServer() throws Exception{
        for(ScheduleStrategyRunntime run: this.scheduleStrategyManager.loadAllScheduleStrategyRunntimeByUUID(this.uuid)){
            List<ScheduleStrategyRunntime> factoryList = this.scheduleStrategyManager.loadAllScheduleStrategyRunntimeByTaskType(run.getStrategyName());
            if(factoryList.size() == 0 || this.isLeader(this.uuid, factoryList) ==false){
                continue;
            }
            ScheduleStrategy scheduleStrategy =this.scheduleStrategyManager.loadStrategy(run.getStrategyName());
            
            int[] nums =  ScheduleUtil.assignTaskNumber(factoryList.size(), scheduleStrategy.getAssignNum(), scheduleStrategy.getNumOfSingleServer());
            for(int i=0;i<factoryList.size();i++){
                ScheduleStrategyRunntime factory =  factoryList.get(i);
                //更新请求的服务器数量
                this.scheduleStrategyManager.updateStrategyRunntimeReqestNum(run.getStrategyName(), 
                        factory.getUuid(),nums[i]);
            }
        }
    }

• 从策略目录获取所有策略信息
  策略目录/tbSchedule/zmlTbScheduleTest/strategy
  策略信息

[ScheduleStrategyRunntime [strategyName=DBDemoSingle-stratrery,
  uuid=192.168.255.1$zhuml$69F86F4CFF4E4DA397880E19411DF068$0000000005, 
  ip=null, kind=null, taskName=null, taskParameter=null, requestNum=0, currentNum=0, message=]]

• 再根据策略名字DBDemoSingle-stratrery，获取策略详细信息
  目录路径：/tbSchedule/zmlTbScheduleTest/strategy/DBDemoSingle-stratrery

  {"strategyName":"DBDemoSingle-stratrery","IPList":["127.0.0.1"],"numOfSingleServer":0,"assignNum":4,
    "kind":"Schedule","taskName":"DBDemoSingle_task","taskParameter":"","sts":"resume"}

• 然后任务分片，并把分片信息写到zookeeper上
  路径 /tbSchedule/zmlTbScheduleTest/strategy/DBDemoSingle-stratrery/192.168.10.249$zhuml$43AB1F13800E4469BBB2DD385C6A0193$0000000006
  data:

  [{"strategyName":"DBDemoSingle-stratrery","uuid":"192.168.10.249$zhuml$43AB1F13800E4469BBB2DD385C6A0193$0000000006","requestNum":4,"currentNum":0,"message":""}]

1.1.2 运行 this.assignScheduleServer();

    public void reRunScheduleServer() throws Exception{
        for (ScheduleStrategyRunntime run : this.scheduleStrategyManager.loadAllScheduleStrategyRunntimeByUUID(this.uuid)) {
            List<IStrategyTask> list = this.managerMap.get(run.getStrategyName());
            if(list == null){
                list = new ArrayList<IStrategyTask>();
                this.managerMap.put(run.getStrategyName(),list);
            }
            while(list.size() > run.getRequestNum() && list.size() >0){
                IStrategyTask task  =  list.remove(list.size() - 1);
                    try {
                        task.stop(run.getStrategyName());
                    } catch (Throwable e) {
                        logger.error("注销任务错误：strategyName=" + run.getStrategyName(), e);
                    }
                }
           //不足，增加调度器
           ScheduleStrategy strategy = this.scheduleStrategyManager.loadStrategy(run.getStrategyName());
           while(list.size() < run.getRequestNum()){
               IStrategyTask result = this.createStrategyTask(strategy);
               if(null==result){
                   logger.error("strategy 对应的配置有问题。strategy name="+strategy.getStrategyName());
               }
               list.add(result);
            }
        }
    }

• 获取策略信息，根据分片数量判断是否停止任务或者创建任务
  停止任务stop就ok了
  创建任务 createStrategyTask

    /**
     * 创建调度服务器
     * @param baseTaskType
     * @param ownSign
     * @return
     * @throws Exception
     */
    public IStrategyTask createStrategyTask(ScheduleStrategy strategy)
            throws Exception {
        IStrategyTask result = null;
        try{
            if(ScheduleStrategy.Kind.Schedule == strategy.getKind()){
                String baseTaskType = ScheduleUtil.splitBaseTaskTypeFromTaskType(strategy.getTaskName());
                String ownSign =ScheduleUtil.splitOwnsignFromTaskType(strategy.getTaskName());
                result = new TBScheduleManagerStatic(this,baseTaskType,ownSign,scheduleDataManager);
            }else if(ScheduleStrategy.Kind.Java == strategy.getKind()){
                result=(IStrategyTask)Class.forName(strategy.getTaskName()).newInstance();
                result.initialTaskParameter(strategy.getStrategyName(),strategy.getTaskParameter());
            }else if(ScheduleStrategy.Kind.Bean == strategy.getKind()){
                result=(IStrategyTask)this.getBean(strategy.getTaskName());
                result.initialTaskParameter(strategy.getStrategyName(),strategy.getTaskParameter());
            }
        }catch(Exception e ){
            logger.error("strategy 获取对应的java or bean 出错,schedule并没有加载该任务,请确认" +strategy.getStrategyName(),e);
        }
        return result;
    }

创建任务有三种类型 public enum Kind{Schedule,Java,Bean}
这里只介绍Schedule：result = new TBScheduleManagerStatic(this,baseTaskType,ownSign,scheduleDataManager);