调度中心admin

1. 调度中心admin启动过程

  1. // 1. 加载 XxlJobAdminConfig
  3. //2. 启动过程  XxlJobScheduler
  4. public void init() throws Exception {
  5. // init i18n
  6. initI18n();
  8. // admin registry monitor run 注册监听器线程的启动  ***
  9. JobRegistryMonitorHelper.getInstance().start();
  11. // admin fail-monitor run  失败任务监听线程启动  ***  每10s去数据库轮询失败的日志任务，然后发送消息
  12. JobFailMonitorHelper.getInstance().start();
  14. // admin lose-monitor run  丢失任务监听线程启动 ***                         
  15. // 任务结果丢失处理：调度记录停留在 "运行中" 状态超过10min，且对应执行器心跳注册失败不在线，则将本地调度主动标记失败；
  16. JobLosedMonitorHelper.getInstance().start();
  18. // admin trigger pool start 触发器池线程的启动 *****  
  19. JobTriggerPoolHelper.toStart();
  21. // admin log report start  日志报告线程的启动 *** 每1s统计一次任务执行器执行情况
  22. JobLogReportHelper.getInstance().start();
  24. // start-schedule  调度线程的启动维护 *****
  25. JobScheduleHelper.getInstance().start();
  27. }

1.1 注册监听器线程的启动

每30s维护一次自动注册的在线执行器。

  1. //任务注册Beat周期默认30s; 执行器以一倍Beat进行执行器注册, 调度中心以一倍Beat进行动态任务发现; 注册信息的失效时间为三倍Beat; 
  3. public void start(){
  4.         registryThread = new Thread(new Runnable() {
  5.             @Override
  6.             public void run() {
  7.                 while (!toStop) {
  8.                     try {
  9.                         // 查询出自动注册类型的执行器
  10.                         List<XxlJobGroup> groupList = XxlJobAdminConfig.getAdminConfig().getXxlJobGroupDao().findByAddressType(0);
  12.                         if (groupList!=null && !groupList.isEmpty()) {
  13.                             // 删除无效的注册地址   BEAT_TIMEOUT=30s,DEAD_TIMEOUT = BEAT_TIMEOUT * 3;
  14.               //找到 xxl_job_registry表中 updateTime < 当前时间-90s的地址,判定这些地址无效,数据删除
  15.                             List<Integer> ids = XxlJobAdminConfig.getAdminConfig().getXxlJobRegistryDao().findDead(RegistryConfig.DEAD_TIMEOUT, new Date());
  16.                             if (ids!=null && ids.size()>0) {
  17.                                 XxlJobAdminConfig.getAdminConfig().getXxlJobRegistryDao().removeDead(ids);
  18.                             }
  20.               // 找到 xxl_job_registry表中 updateTime > 当前时间-90s 的有效地址,更新xxl_job_group的address_list字段
  21.                             // fresh online address (admin/executor)
  22.                             HashMap<String, List<String>> appAddressMap = new HashMap<String, List<String>>();
  23.                             List<XxlJobRegistry> list = XxlJobAdminConfig.getAdminConfig().getXxlJobRegistryDao().findAll(RegistryConfig.DEAD_TIMEOUT, new Date());
  24.                             if (list != null) {
  25.                                 for (XxlJobRegistry item: list) {
  26.                                     if (RegistryConfig.RegistType.EXECUTOR.name().equals(item.getRegistryGroup())) {
  27.                                         String appname = item.getRegistryKey();
  28.                                         List<String> registryList = appAddressMap.get(appname);
  29.                                         if (registryList == null) {
  30.                                             registryList = new ArrayList<String>();
  31.                                         }
  33.                                         if (!registryList.contains(item.getRegistryValue())) {
  34.                                             registryList.add(item.getRegistryValue());
  35.                                         }
  36.                                         appAddressMap.put(appname, registryList);
  37.                                     }
  38.                                 }
  39.                             }
  41.                             // fresh group address
  42.                             for (XxlJobGroup group: groupList) {
  43.                                 List<String> registryList = appAddressMap.get(group.getAppname());
  44.                                 String addressListStr = null;
  45.                                 if (registryList!=null && !registryList.isEmpty()) {
  46.                                     Collections.sort(registryList);
  47.                                     StringBuilder addressListSB = new StringBuilder();
  48.                                     for (String item:registryList) {
  49.                                         addressListSB.append(item).append(",");
  50.                                     }
  51.                                     addressListStr = addressListSB.toString();
  52.                                     addressListStr = addressListStr.substring(0, addressListStr.length()-1);
  53.                                 }
  54.                                 group.setAddressList(addressListStr);
  55.                                 XxlJobAdminConfig.getAdminConfig().getXxlJobGroupDao().update(group);
  56.                             }
  57.                         }
  58.                     } catch (Exception e) {
  59.                         if (!toStop) {
  60.                             logger.error(">>>>>>>>>>> xxl-job, job registry monitor thread error:{}", e);
  61.                         }
  62.                     }
  63.                     try {
  64.             //频率为30s一次
  65.                         TimeUnit.SECONDS.sleep(RegistryConfig.BEAT_TIMEOUT);
  66.                     } catch (InterruptedException e) {
  67.                         if (!toStop) {
  68.                             logger.error(">>>>>>>>>>> xxl-job, job registry monitor thread error:{}", e);
  69.                         }
  70.                     }
  71.                 }
  72.                 logger.info(">>>>>>>>>>> xxl-job, job registry monitor thread stop");
  73.             }
  74.         });
  75.     //守护线程
  76.         registryThread.setDaemon(true);
  77.         registryThread.setName("xxl-job, admin JobRegistryMonitorHelper");
  78.         registryThread.start();
  79.     }

1.2 调度线程的启动维护
  1. 1. 计算单线程1s处理任务数,假设处理一个任务需要50ms,那么1s能够处理20个任务
  2. 2. 计算出1s最大处理任务量 = 最大线程数 * 20
  3. 3. 死循环轮询加载需要调度的任务
  4.      3.1 开启事务，使用 select * from xxl_job_lock where lock_name = 'schedule_lock' for update 加行级锁，保证单线程处理
  5.      3.2 得到未来5s的任务, triggerNextTime < nowTime()+5s, limit 最大任务处理量
  6.      3.3 判断任务情况
  7.          if (nowTime > triggerNextTime +5 s), 不再执行该任务(misfire)，只更新下次执行时间
  8.          else if (nowTime > getTriggerNextTime)
  9.             , 超时在5s内的，立即trigger任务，然后更新下次执行时间
  10.             if(jobInfo.getTriggerStatus() == 1 && nowTime + PRE_READ_MS > jobInfo.getTriggerNextTime())
  11.              成功触发任务,如果下次执行时间在5秒内，扔到ringData中等待处理，更新下次执行时间
  12.        else 还未达到触发的时间点，扔到ringData中等待处理，更新下次执行时间
  13. 4. 提交事务,释放锁
  14. 5. 判断此次时间是否超过了1s,小于1s,需等待到1s，TimeUnit.MILLISECONDS.sleep((preReadSuc ? 1000 : PRE_READ_MS) - System.currentTimeMillis() % 1000);
  15.    否则,直接下次循环     
  17. ## 调度的准时处理
  18.     秒级的准时处理,3.3 中Map<Integer, List<Integer>> ringData 的key使用的是 int ringSecond = (int) ((jobInfo.getTriggerNextTime() / 1000) % 60);这样保证了一个jobId不会出现重复添加的情况
  19.      // 避免处理耗时太长，跨过刻度，向前校验一个刻度；
  20.      for (int i = 0; i < 2; i++) {
  21.        List<Integer> tmpData = ringData.remove((nowSecond + 60 - i) % 60);
  22.        if (tmpData != null) {
  23.            ringItemData.addAll(tmpData);
  24.        }
  25.      }

1.3 触发器池线程的启动

1.4 触发的路由策略

  • 一致性hash策略

    1. 分组下机器地址相同，不同JOB均匀散列在不同机器上，保证分组下机器分配JOB平均；且每个JOB固定调度其中一台机器；
    2. a、virtual node：解决不均衡问题
    3. b、hash method replace hashCode：String的hashCode可能重复，需要进一步扩大hashCode的取值范围
    4. 参考文章: https://juejin.im/post/6873407119421341704

  • 最不经常使用原则 LFU

    1. // load least userd count address
    2.       List<Map.Entry<String, Integer>> lfuItemList = new ArrayList<>(lfuItemMap.entrySet());
    3. //按照使用次数排序    
    4. Collections.sort(lfuItemList, Comparator.comparing(Map.Entry::getValue));


  • LRU

    1. /**
    2. * LinkedHashMap
    3. *      a、accessOrder：true=访问顺序排序（get/put时排序）；false=插入顺序排期；
    4. *      b、removeEldestEntry：新增元素时将会调用，返回true时会删除最老元素；可封装LinkedHashMap并重写该方法，比如定义最大容量，超出是返回true即可实现固定长度的LRU算法；
    5. */
    6. 巧用 new LinkedHashMap<>(16, 0.75f, true);
    7. 或者 new LinkedHashSet<>(16, 0.75f, true);

  • https://juejin.im/post/6844904053000912903

  • 故障转移
    做一次心跳检测，执行心跳检测成功的服务
  • 忙碌转移
    判断当前job对应的线程是否正在执行,以非执行中的去执行
  • 轮询
    按照一定的规则取模