分布式定时任务系列5：XXL-job中blockingQueue的应用-平芜编程栈

传送门

分布式定时任务系列1：XXL-job安装

分布式定时任务系列2：XXL-job使用

分布式定时任务系列3：任务执行引擎设计

分布式定时任务系列4：任务执行引擎设计续

分布式定时任务系列6：XXL-job触发日志过大引发的CPU告警

分布式定时任务系列7：XXL-job源码分析之任务触发

分布式定时任务系列8：XXL-job源码分析之远程调用

分布式定时任务系列9：XXL-job路由策略

分布式定时任务系列10：XXL-job源码分析之路由策略

分布式定时任务系列12：XXL-job的任务触发为什么是死循环？

分布式定时任务系列13：死循环是任务触发的银弹？

分布式定时任务系列14：XXL-job的注册模型

Java并发编程实战1：java中的阻塞队列

引子

这篇文章的主要目不是讨论XXL-job的使用，而是要通过它的任务线程实现机制来分析java中阻塞队列的应用！

而这一切要从上周某天，公司一个普通的下午说起。

当时一个同事要添加任务，就随口问了一句：“阻塞处理策略选啥？”，我心里一惊，以前没注意过这个地方，每次都是默认的，也就是单机串行。

就是下图这个：

这里面有3个选项：

众人先是侃侃讨论了半天，后面再去翻了源码来查看，发现了下面的blockingQueue

阻塞队列

什么是阻塞队列

关于java阻塞队列，可以看看java中的阻塞队列。里面例子用到的是有界队列ArrayBlockingQueue，XXL-job里面用的是无界队列LinkedBlockingQueue。不论它们的相似及区别点，最重要的都是拥有阻塞特性。

要分析XXL-job是怎么使用阻塞队列的，就从XXL-job的调度触发机制入手！

什么是XXL-job任务

前面在分布式定时任务系列2：XXL-job使用里面介绍过XXL-job的使用，注册一个任务也很简单，只要在业务代码方法中打上@XxlJob注解即可！下面是官方提供的demo：

/** * 1、简单任务示例（Bean模式） */ @XxlJob("demoJobHandler") public void demoJobHandler() throws Exception { XxlJobHelper.log("XXL-JOB, Hello World."); for (int i = 0; i < 5; i++) { XxlJobHelper.log("beat at:" + i); TimeUnit.SECONDS.sleep(2); } // default success }

开发步骤

1、任务开发：在Spring Bean实例中，开发Job方法；
2、注解配置：为Job方法添加注解 "@XxlJob(value="自定义jobhandler名称", init = "JobHandler初始化方法", destroy = "JobHandler销毁方法")"，注解value值对应的是调度中心新建任务的JobHandler属性的值。
3、执行日志：需要通过 "XxlJobHelper.log" 打印执行日志；
4、任务结果：默认任务结果为 "成功" 状态，不需要主动设置；如有诉求，比如设置任务结果为失败，可以通过 "XxlJobHelper.handleFail/handleSuccess" 自主设置任务结果；

任务注册

对于XXL-job来说，是怎么注册并管理这些任务的呢？

在程序启动的时候，xxl-job-core包中的XxlJobSpringExecutor类由于实现了接口SmartInitializingSingleton，在Bean实例化完成时会调用afterSingletonsInstantiated()方法：

@Override public void afterSingletonsInstantiated() { // init JobHandler Repository (for method) initJobHandlerMethodRepository(applicationContext); // refresh GlueFactory GlueFactory.refreshInstance(1); // super start try { super.start(); } catch (Exception e) { throw new RuntimeException(e); } }

在initJobHandlerMethodRepository(applicationContext)方法中，现在看一下这个方法的代码：该方法会提取所有SpringBean的实例中方法上添加了XxlJob注解的方法，并进行注册！

private void initJobHandlerMethodRepository(ApplicationContext applicationContext) { if (applicationContext == null) { return; } // init job handler from method String[] beanDefinitionNames = applicationContext.getBeanNamesForType(Object.class, false, true); for (String beanDefinitionName : beanDefinitionNames) { Object bean = applicationContext.getBean(beanDefinitionName); Map<Method, XxlJob> annotatedMethods = null; // referred to ：org.springframework.context.event.EventListenerMethodProcessor.processBean try { annotatedMethods = MethodIntrospector.selectMethods(bean.getClass(), new MethodIntrospector.MetadataLookup<XxlJob>() { @Override public XxlJob inspect(Method method) { return AnnotatedElementUtils.findMergedAnnotation(method, XxlJob.class); } }); } catch (Throwable ex) { logger.error("xxl-job method-jobhandler resolve error for bean[" + beanDefinitionName + "].", ex); } if (annotatedMethods==null || annotatedMethods.isEmpty()) { continue; } for (Map.Entry<Method, XxlJob> methodXxlJobEntry : annotatedMethods.entrySet()) { Method executeMethod = methodXxlJobEntry.getKey(); XxlJob xxlJob = methodXxlJobEntry.getValue(); // regist registJobHandler(xxlJob, bean, executeMethod); } } // ---------------------- job handler repository ---------------------- private static ConcurrentMap<String, IJobHandler> jobHandlerRepository = new ConcurrentHashMap<String, IJobHandler>(); public static IJobHandler loadJobHandler(String name){ return jobHandlerRepository.get(name); } public static IJobHandler registJobHandler(String name, IJobHandler jobHandler){ logger.info(">>>>>>>>>>> xxl-job register jobhandler success, name:{}, jobHandler:{}", name, jobHandler); return jobHandlerRepository.put(name, jobHandler); }

注册的方法就是registJobHandler(name, new MethodJobHandler(bean, executeMethod, initMethod, destroyMethod))，会将XxlJob注解标记的方法，提取出对应的方法名，通过反射得到Method信息并实例为对应的MethodJobHandler，最后通过ConcurrentMap来管理，其中key是注任务名称，value为JobHandler实例。

看到这里，不知你有没有一个疑问：这里的JobHandler里面并没有阻塞队列BlockingQueue，它们是怎么关联起来的呢？所以这里就要讨论一下XXL-job的任务执行!

XXL-job的任务执行

在后台管理手动触发一下测试任务，并把代码打上debug断点：

首先会发送后台请求：

一路跟踪下去，会进入到XxlJobExecutor.runExecutor(TriggerParam triggerParam, String address)方法，最终是到ExecutorBizImpl.run(TriggerParam triggerParam)方法：

public ReturnT<String> run(TriggerParam triggerParam) { // load old：jobHandler + jobThread JobThread jobThread = XxlJobExecutor.loadJobThread(triggerParam.getJobId()); IJobHandler jobHandler = jobThread!=null?jobThread.getHandler():null; String removeOldReason = null; // valid：jobHandler + jobThread GlueTypeEnum glueTypeEnum = GlueTypeEnum.match(triggerParam.getGlueType()); if (GlueTypeEnum.BEAN == glueTypeEnum) { // new jobhandler IJobHandler newJobHandler = XxlJobExecutor.loadJobHandler(triggerParam.getExecutorHandler()); // valid old jobThread if (jobThread!=null && jobHandler != newJobHandler) { // change handler, need kill old thread removeOldReason = "change jobhandler or glue type, and terminate the old job thread."; jobThread = null; jobHandler = null; } // valid handler if (jobHandler == null) { jobHandler = newJobHandler; if (jobHandler == null) { return new ReturnT<String>(ReturnT.FAIL_CODE, "job handler [" + triggerParam.getExecutorHandler() + "] not found."); } } } else if (GlueTypeEnum.GLUE_GROOVY == glueTypeEnum) { // valid old jobThread if (jobThread != null && !(jobThread.getHandler() instanceof GlueJobHandler && ((GlueJobHandler) jobThread.getHandler()).getGlueUpdatetime()==triggerParam.getGlueUpdatetime() )) { // change handler or gluesource updated, need kill old thread removeOldReason = "change job source or glue type, and terminate the old job thread."; jobThread = null; jobHandler = null; } // valid handler if (jobHandler == null) { try { IJobHandler originJobHandler = GlueFactory.getInstance().loadNewInstance(triggerParam.getGlueSource()); jobHandler = new GlueJobHandler(originJobHandler, triggerParam.getGlueUpdatetime()); } catch (Exception e) { logger.error(e.getMessage(), e); return new ReturnT<String>(ReturnT.FAIL_CODE, e.getMessage()); } } } else if (glueTypeEnum!=null && glueTypeEnum.isScript()) { // valid old jobThread if (jobThread != null && !(jobThread.getHandler() instanceof ScriptJobHandler && ((ScriptJobHandler) jobThread.getHandler()).getGlueUpdatetime()==triggerParam.getGlueUpdatetime() )) { // change script or gluesource updated, need kill old thread removeOldReason = "change job source or glue type, and terminate the old job thread."; jobThread = null; jobHandler = null; } // valid handler if (jobHandler == null) { jobHandler = new ScriptJobHandler(triggerParam.getJobId(), triggerParam.getGlueUpdatetime(), triggerParam.getGlueSource(), GlueTypeEnum.match(triggerParam.getGlueType())); } } else { return new ReturnT<String>(ReturnT.FAIL_CODE, "glueType[" + triggerParam.getGlueType() + "] is not valid."); } // executor block strategy if (jobThread != null) { ExecutorBlockStrategyEnum blockStrategy = ExecutorBlockStrategyEnum.match(triggerParam.getExecutorBlockStrategy(), null); if (ExecutorBlockStrategyEnum.DISCARD_LATER == blockStrategy) { // discard when running if (jobThread.isRunningOrHasQueue()) { return new ReturnT<String>(ReturnT.FAIL_CODE, "block strategy effect："+ExecutorBlockStrategyEnum.DISCARD_LATER.getTitle()); } } else if (ExecutorBlockStrategyEnum.COVER_EARLY == blockStrategy) { // kill running jobThread if (jobThread.isRunningOrHasQueue()) { removeOldReason = "block strategy effect：" + ExecutorBlockStrategyEnum.COVER_EARLY.getTitle(); jobThread = null; } } else { // just queue trigger } } // replace thread (new or exists invalid) if (jobThread == null) { jobThread = XxlJobExecutor.registJobThread(triggerParam.getJobId(), jobHandler, removeOldReason); } // push data to queue ReturnT<String> pushResult = jobThread.pushTriggerQueue(triggerParam); return pushResult; }

任务执行过程中，如果发现XXL-job对应的执行线程不存在，会创建一个新new Thread实例，并绑定前面的JobHandler实例，这样对于每一个任务都有一个单独的线程，也就将执行线程JobThread跟任务关联起来了：

public static JobThread registJobThread(int jobId, IJobHandler handler, String removeOldReason){ JobThread newJobThread = new JobThread(jobId, handler); newJobThread.start(); logger.info(">>>>>>>>>>> xxl-job regist JobThread success, jobId:{}, handler:{}", new Object[]{jobId, handler}); JobThread oldJobThread = jobThreadRepository.put(jobId, newJobThread); // putIfAbsent | oh my god, map's put method return the old value!!! if (oldJobThread != null) { oldJobThread.toStop(removeOldReason); oldJobThread.interrupt(); } return newJobThread; } public JobThread(int jobId, IJobHandler handler) { this.jobId = jobId; this.handler = handler; // 初始化阻塞队列 this.triggerQueue = new LinkedBlockingQueue<TriggerParam>(); this.triggerLogIdSet = Collections.synchronizedSet(new HashSet<Long>()); // assign job thread name this.setName("xxl-job, JobThread-"+jobId+"-"+System.currentTimeMillis()); }

而执行任务的最后一步，就是将当前任务入队到刚才的执行线程的BlockingQueue中：

// push data to queue ReturnT<String> pushResult = jobThread.pushTriggerQueue(triggerParam);

而这也就跟文章最前面的代码关联起来了：

/** * new trigger to queue * * @param triggerParam * @return */ public ReturnT<String> pushTriggerQueue(TriggerParam triggerParam) { // avoid repeat if (triggerLogIdSet.contains(triggerParam.getLogId())) { logger.info(">>>>>>>>>>> repeate trigger job, logId:{}", triggerParam.getLogId()); return new ReturnT<String>(ReturnT.FAIL_CODE, "repeate trigger job, logId:" + triggerParam.getLogId()); } triggerLogIdSet.add(triggerParam.getLogId()); triggerQueue.add(triggerParam); return ReturnT.SUCCESS; }