XXL-JOB-01

2018/04/28 15:44
阅读数 533

xxx-job 源码解读(一)

2018年03月07日 14:43:56

阅读数:52

1.  调度中心启动源码分析

 首先从spring的配置看起, 从以下配置可以看出,xxl内部使用的是quartz

spring配置

<bean id="quartzScheduler" lazy-init="false" class="org.springframework.scheduling.quartz.SchedulerFactoryBean">

   <property name="dataSource" ref="dataSource" />

   <property name="autoStartup" value="true" />         <!--自动启动 -->

   <property name="startupDelay" value="20" />             <!--延时启动,应用启动成功后在启动 -->

   <property name="overwriteExistingJobs" value="true" /> <!--覆盖DB中JOB:true、以数据库中已经存在的为准:false -->

   <property name="applicationContextSchedulerContextKey"  value="applicationContextKey" />

   <property name="configLocation" value="classpath:quartz.properties"/>

</bean>

<!-- 这个调度中心,在启动的时候,会做很多初始化的工作 ,比如:执行器信息,注册机器列表等信息 -->

<bean id="xxlJobDynamicScheduler" class="com.xxl.job.admin.core.schedule.XxlJobDynamicScheduler" init-method="init" destroy-method="destroy" >

    <!-- 配置调度中心的名称 -->

   <property name="scheduler" ref="quartzScheduler"/>

    <!-- 用于调度中心和执行器之间通信的时候做数据加密 -->

   <property name="accessToken" value="${xxl.job.accessToken}" />

</bean>

com.xxl.job.admin.core.schedule.XxlJobDynamicScheduler  在启动的时候会做如下工作:

 

XxlJobDynamicScheduler

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

public void init() throws Exception {

    // 启动自动注册线程, 获取类型为自动注册的执行器信息,完成机器的自动注册与发现

    JobRegistryMonitorHelper.getInstance().start();

 

    // 启动失败日志监控线程

    JobFailMonitorHelper.getInstance().start();

 

    // admin-server(spring-mvc)

    NetComServerFactory.putService(AdminBiz.class, XxlJobDynamicScheduler.adminBiz);

    NetComServerFactory.setAccessToken(accessToken);

 

    // valid

    Assert.notNull(scheduler, "quartz scheduler is null");

    logger.info(">>>>>>>>> init xxl-job admin success.");

}

 

JobRegistryMonitorHelper.getInstance().start()   详细代码如下:

 

 

JobRegistryMonitorHelper

public void start(){

   //创建一个线程

   registryThread = new Thread(new Runnable() {

      @Override

      public void run() {

         // 当toStop 为false时进入该循环。

         while (!toStop) {

            try {

               // 获取类型为自动注册的执行器地址列表

               List<XxlJobGroup> groupList = XxlJobDynamicScheduler.xxlJobGroupDao.findByAddressType(0);

               if (CollectionUtils.isNotEmpty(groupList)) {

 

                  // 删除 90秒之内没有更新信息的注册机器, 90秒没有心跳信息返回,代表机器已经出现问题,故移除

                  XxlJobDynamicScheduler.xxlJobRegistryDao.removeDead(RegistryConfig.DEAD_TIMEOUT);

 

                  // fresh online address (admin/executor)

                  HashMap<String, List<String>> appAddressMap = new HashMap<String, List<String>>();

                  // 查询在90秒之内有过更新的机器列表

                  List<XxlJobRegistry> list = XxlJobDynamicScheduler.xxlJobRegistryDao.findAll(RegistryConfig.DEAD_TIMEOUT);

                  if (list != null) {

                     //循环注册机器列表,  根据执行器不同,将这些机器列表区分拿出来

                     for (XxlJobRegistry item: list) {

                        // 判断该机器注册信息RegistryGroup ,RegistType 是否是EXECUTOR , EXECUTOR 代表该机器是注册到执行器上面的

                        // RegistType  分为两种, ADMIN 和EXECUTOR

                        if (RegistryConfig.RegistType.EXECUTOR.name().equals(item.getRegistryGroup())) {

                           // 获取注册的执行器 KEY  (也就是执行器)

                           String appName = item.getRegistryKey();

                           List<String> registryList = appAddressMap.get(appName);

                           if (registryList == null) {

                              registryList = new ArrayList<String>();

                           }

 

                           if (!registryList.contains(item.getRegistryValue())) {

                              registryList.add(item.getRegistryValue());

                           }

                           // 收集 机器信息,根据执行器做区分

                           appAddressMap.put(appName, registryList);

                        }

                     }

                  }

 

                  //  遍历执行器列表

                  for (XxlJobGroup group: groupList) {

                     // 通过执行器的APP_NAME  拿出他下面的集群机器地址

                     List<String> registryList = appAddressMap.get(group.getAppName());

                     String addressListStr = null;

                     if (CollectionUtils.isNotEmpty(registryList)) {

                        Collections.sort(registryList);

                        // 转为为String, 通过逗号分隔

                        addressListStr = StringUtils.join(registryList, ",");

                     }

                     group.setAddressList(addressListStr);

                     // 将 这个执行器的 集群机器地址列表,写入到数据库

                     XxlJobDynamicScheduler.xxlJobGroupDao.update(group);

                  }

               }

            catch (Exception e) {

               logger.error("job registry instance error:{}", e);

            }

            try {

               TimeUnit.SECONDS.sleep(RegistryConfig.BEAT_TIMEOUT);

            catch (InterruptedException e) {

               logger.error("job registry instance error:{}", e);

            }

         }

      }

   });

   registryThread.setDaemon(true);

   //启动线程

   registryThread.start();

}

JobFailMonitorHelper.getInstance().start(); 详细代码如下:

JobFailMonitorHelper

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

public void start(){

   // 启动线程

 monitorThread = new Thread(new Runnable() {

 

 @Override

 public void run() {

         // monitor

 while (!toStop) {

            try {

               List<Integer> jobLogIdList = new ArrayList<Integer>();

               // 从队列中拿出所有可用的 jobLogIds

                int drainToNum = JobFailMonitorHelper.instance.queue.drainTo(jobLogIdList);

               if (CollectionUtils.isNotEmpty(jobLogIdList)) {

                  for (Integer jobLogId : jobLogIdList) {

                     if (jobLogId==null || jobLogId==0) {

                        continue;

                     }

                     //从数据库跟以前有日志信息

                    XxlJobLog log = XxlJobDynamicScheduler.xxlJobLogDao.load(jobLogId);

                     if (log == null) {

                        continue;

                     }

                     //任务触发成功, 但是JobHandle 还没有返回结果

                    if (IJobHandler.SUCCESS.getCode() == log.getTriggerCode() && log.getHandleCode() == 0) {

                        //将 JobLogId 放入队列 , 继续监控

                         JobFailMonitorHelper.monitor(jobLogId);

                        logger.info(">>>>>>>>>>> job monitor, job running, JobLogId:{}", jobLogId);

                     else if (IJobHandler.SUCCESS.getCode() == log.getHandleCode()) {

                        // job success, pass

                        logger.info(">>>>>>>>>>> job monitor, job success, JobLogId:{}", jobLogId);

                     else if (IJobHandler.FAIL.getCode() == log.getTriggerCode()

                           || IJobHandler.FAIL.getCode() == log.getHandleCode()

                           || IJobHandler.FAIL_RETRY.getCode() == log.getHandleCode() ) {

                        // 任务执行失败, 执行发送邮件等预警措施

                        failAlarm(log);

                        logger.info(">>>>>>>>>>> job monitor, job fail, JobLogId:{}", jobLogId);

                     else {

                        JobFailMonitorHelper.monitor(jobLogId);

                        logger.info(">>>>>>>>>>> job monitor, job status unknown, JobLogId:{}", jobLogId);

                     }

                  }

               }

               // 停顿一下

                TimeUnit.SECONDS.sleep(10);

            catch (Exception e) {

               logger.error("job monitor error:{}", e);

            }

         }

 

   });

   monitorThread.setDaemon(true);

   monitorThread.start();

}

以上 是xxl-job 在启动的时候做的操作,  主要是启动两个线程,  

  1. 用来监控自动注册上来的机器,达到自动注册的目的
  2. 监控任务的执行状态, 如若失败,则发送邮件预警
     

xxl-job 是基于quartz 进行的二次开发,在系统启动的时候,quartz框架会自动去数据库读取相关的配置信息,载入相关定时器信息

展开阅读全文
打赏
0
0 收藏
分享
加载中
更多评论
打赏
0 评论
0 收藏
0
分享
返回顶部
顶部