From 628d42194199fb1b5e832c217b2891c0d73eb7a4 Mon Sep 17 00:00:00 2001 From: "xueli.xue" Date: Thu, 21 Jul 2016 23:28:07 +0800 Subject: [PATCH] fresh --- .../admin/core/jobbean/RemoteHttpJobBean.java | 2 + .../xxl/job/admin/core/model/XxlJobLog.java | 7 ++ .../admin/core/thread/JobMonitorHelper.java | 12 +++- .../mybatis-mapper/XxlJobLogMapper.xml | 5 ++ .../template/jobinfo/jobinfo.index.ftl | 6 +- .../WEB-INF/template/joblog/joblog.index.ftl | 5 +- .../main/webapp/static/js/joblog.index.1.js | 3 +- .../core/executor/jetty/XxlJobExecutor.java | 20 ++---- .../job/core/handler/HandlerRepository.java | 66 ++++++++++++++----- .../service/jobhandler/DemoJobHandler.java | 2 +- 10 files changed, 88 insertions(+), 40 deletions(-) diff --git a/xxl-job-admin/src/main/java/com/xxl/job/admin/core/jobbean/RemoteHttpJobBean.java b/xxl-job-admin/src/main/java/com/xxl/job/admin/core/jobbean/RemoteHttpJobBean.java index 6b92877d..300a4d1a 100644 --- a/xxl-job-admin/src/main/java/com/xxl/job/admin/core/jobbean/RemoteHttpJobBean.java +++ b/xxl-job-admin/src/main/java/com/xxl/job/admin/core/jobbean/RemoteHttpJobBean.java @@ -52,12 +52,14 @@ public class RemoteHttpJobBean extends QuartzJobBean { params.put(HandlerParamEnum.JOB_GROUP.name(), jobInfo.getJobGroup()); params.put(HandlerParamEnum.JOB_NAME.name(), jobInfo.getJobName()); + params.put(HandlerParamEnum.EXECUTOR_HANDLER.name(), jobInfo.getExecutorHandler()); params.put(HandlerParamEnum.EXECUTOR_PARAMS.name(), jobInfo.getExecutorParam()); params.put(HandlerParamEnum.GLUE_SWITCH.name(), String.valueOf(jobInfo.getGlueSwitch())); // failover trigger RemoteCallBack callback = failoverTrigger(jobInfo.getExecutorAddress(), params, jobLog); + jobLog.setExecutorHandler(jobInfo.getExecutorHandler()); jobLog.setExecutorParam(jobInfo.getExecutorParam()); logger.info(">>>>>>>>>>> xxl-job failoverTrigger response, jobId:{}, callback:{}", jobLog.getId(), callback); diff --git a/xxl-job-admin/src/main/java/com/xxl/job/admin/core/model/XxlJobLog.java b/xxl-job-admin/src/main/java/com/xxl/job/admin/core/model/XxlJobLog.java index 70c52ab1..5e58923f 100644 --- a/xxl-job-admin/src/main/java/com/xxl/job/admin/core/model/XxlJobLog.java +++ b/xxl-job-admin/src/main/java/com/xxl/job/admin/core/model/XxlJobLog.java @@ -15,6 +15,7 @@ public class XxlJobLog { private String jobName; private String executorAddress; // 执行器地址，有多个则逗号分隔 + private String executorHandler; // 执行器Handler private String executorParam; // 执行器，任务参数 // trigger info @@ -51,6 +52,12 @@ public class XxlJobLog { public void setExecutorAddress(String executorAddress) { this.executorAddress = executorAddress; } + public String getExecutorHandler() { + return executorHandler; + } + public void setExecutorHandler(String executorHandler) { + this.executorHandler = executorHandler; + } public String getExecutorParam() { return executorParam; } diff --git a/xxl-job-admin/src/main/java/com/xxl/job/admin/core/thread/JobMonitorHelper.java b/xxl-job-admin/src/main/java/com/xxl/job/admin/core/thread/JobMonitorHelper.java index c033f115..3e9cb479 100644 --- a/xxl-job-admin/src/main/java/com/xxl/job/admin/core/thread/JobMonitorHelper.java +++ b/xxl-job-admin/src/main/java/com/xxl/job/admin/core/thread/JobMonitorHelper.java @@ -10,6 +10,9 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.text.MessageFormat; +import java.util.Arrays; +import java.util.HashSet; +import java.util.Set; import java.util.concurrent.*; /** @@ -51,8 +54,13 @@ public class JobMonitorHelper { if (RemoteCallBack.FAIL.equals(log.getTriggerStatus()) || RemoteCallBack.FAIL.equals(log.getHandleStatus())) { XxlJobInfo info = DynamicSchedulerUtil.xxlJobInfoDao.load(log.getJobGroup(), log.getJobName()); if (info!=null && info.getAlarmEmail()!=null && info.getAlarmEmail().trim().length()>0) { - MailUtil.sendMail(info.getAlarmEmail(), "《调度监控报警-调度平台平台XXL-JOB》", - MessageFormat.format("任务调度失败, 任务组:{0}, 任务描述:{1}.", info.getJobGroup(), info.getJobDesc()), false, null); + + Set emailSet = new HashSet(Arrays.asList(info.getAlarmEmail().split(","))); + for (String email: emailSet) { + String title = "《调度监控报警-调度平台平台XXL-JOB》"; + String content = MessageFormat.format("任务调度失败, 任务组:{0}, 任务描述:{1}.", info.getJobGroup(), info.getJobDesc()); + MailUtil.sendMail(email, title, content, false, null); + } } } } diff --git a/xxl-job-admin/src/main/resources/mybatis-mapper/XxlJobLogMapper.xml b/xxl-job-admin/src/main/resources/mybatis-mapper/XxlJobLogMapper.xml index ed4d4d6e..d924fb01 100644 --- a/xxl-job-admin/src/main/resources/mybatis-mapper/XxlJobLogMapper.xml +++ b/xxl-job-admin/src/main/resources/mybatis-mapper/XxlJobLogMapper.xml @@ -10,6 +10,7 @@ + @@ -27,6 +28,7 @@ t.job_group, t.job_name, t.executor_address, + t.executor_handler, t.executor_param, t.trigger_time, t.trigger_status, @@ -94,11 +96,13 @@ `job_group`, `job_name`, `executor_address`, + `executor_handler`, `executor_param` ) VALUES ( #{jobGroup}, #{jobName}, #{executorAddress}, + #{executorHandler}, #{executorParam} ); @@ -113,6 +117,7 @@ `trigger_status`= #{triggerStatus}, `trigger_msg`= #{triggerMsg}, `executor_address`= #{executorAddress}, + `executor_handler`=#{executorHandler}, `executor_param`= #{executorParam} WHERE `id`= #{id} diff --git a/xxl-job-admin/src/main/webapp/WEB-INF/template/jobinfo/jobinfo.index.ftl b/xxl-job-admin/src/main/webapp/WEB-INF/template/jobinfo/jobinfo.index.ftl index 9ce3a372..61f583b7 100644 --- a/xxl-job-admin/src/main/webapp/WEB-INF/template/jobinfo/jobinfo.index.ftl +++ b/xxl-job-admin/src/main/webapp/WEB-INF/template/jobinfo/jobinfo.index.ftl @@ -40,7 +40,7 @@
- 分组 + 任务组 <#list JobGroupList as group> @@ -194,7 +194,7 @@ public class DemoJobHandler extends IJobHandler {
- +
diff --git a/xxl-job-admin/src/main/webapp/WEB-INF/template/joblog/joblog.index.ftl b/xxl-job-admin/src/main/webapp/WEB-INF/template/joblog/joblog.index.ftl index 8f13bf39..610c874c 100644 --- a/xxl-job-admin/src/main/webapp/WEB-INF/template/joblog/joblog.index.ftl +++ b/xxl-job-admin/src/main/webapp/WEB-INF/template/joblog/joblog.index.ftl @@ -34,7 +34,7 @@
- 分组 + 任务组 @@ -79,6 +79,7 @@ 任务组 任务名 执行器地址 + JobHandler 任务参数 调度时间 调度结果 diff --git a/xxl-job-admin/src/main/webapp/static/js/joblog.index.1.js b/xxl-job-admin/src/main/webapp/static/js/joblog.index.1.js index 9ad3c6eb..ef89a948 100644 --- a/xxl-job-admin/src/main/webapp/static/js/joblog.index.1.js +++ b/xxl-job-admin/src/main/webapp/static/js/joblog.index.1.js @@ -1,6 +1,6 @@ $(function() { - // 分组列表选中, 任务列表初始化和选中 + // 任务组列表选中, 任务列表初始化和选中 var ifParam = true; $("#jobGroup").on("change", function () { var jobGroup = $(this).children('option:selected').val(); @@ -97,6 +97,7 @@ $(function() { }, { "data": 'jobName', "visible" : false}, { "data": 'executorAddress', "visible" : true}, + { "data": 'executorHandler', "visible" : true}, { "data": 'executorParam', "visible" : true}, { "data": 'triggerTime', diff --git a/xxl-job-core/src/main/java/com/xxl/job/core/executor/jetty/XxlJobExecutor.java b/xxl-job-core/src/main/java/com/xxl/job/core/executor/jetty/XxlJobExecutor.java index b1438de0..20fc5bc4 100644 --- a/xxl-job-core/src/main/java/com/xxl/job/core/executor/jetty/XxlJobExecutor.java +++ b/xxl-job-core/src/main/java/com/xxl/job/core/executor/jetty/XxlJobExecutor.java @@ -1,7 +1,8 @@ package com.xxl.job.core.executor.jetty; -import java.util.*; - +import com.xxl.job.core.handler.HandlerRepository; +import com.xxl.job.core.handler.IJobHandler; +import com.xxl.job.core.handler.annotation.JobHander; import org.eclipse.jetty.server.Connector; import org.eclipse.jetty.server.Handler; import org.eclipse.jetty.server.Server; @@ -14,9 +15,7 @@ import org.springframework.beans.BeansException; import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationContextAware; -import com.xxl.job.core.handler.HandlerRepository; -import com.xxl.job.core.handler.IJobHandler; -import com.xxl.job.core.handler.annotation.JobHander; +import java.util.Map; /** * Created by xuxueli on 2016/3/2 21:14. @@ -89,14 +88,9 @@ public class XxlJobExecutor implements ApplicationContextAware { if (serviceBeanMap!=null && serviceBeanMap.size()>0) { for (Object serviceBean : serviceBeanMap.values()) { if (serviceBean instanceof IJobHandler){ - String jobKeys = serviceBean.getClass().getAnnotation(JobHander.class).value(); - if (jobKeys!=null && jobKeys.trim().length()>0) { - Set jobKeySet = new HashSet(Arrays.asList(jobKeys.split(","))); - for (String jobKey : jobKeySet) { - IJobHandler handler = (IJobHandler) serviceBean; - HandlerRepository.regist(jobKey, handler); - } - } + String name = serviceBean.getClass().getAnnotation(JobHander.class).value(); + IJobHandler handler = (IJobHandler) serviceBean; + HandlerRepository.registJobHandler(name, handler); } } } diff --git a/xxl-job-core/src/main/java/com/xxl/job/core/handler/HandlerRepository.java b/xxl-job-core/src/main/java/com/xxl/job/core/handler/HandlerRepository.java index 4d162db2..67dff129 100644 --- a/xxl-job-core/src/main/java/com/xxl/job/core/handler/HandlerRepository.java +++ b/xxl-job-core/src/main/java/com/xxl/job/core/handler/HandlerRepository.java @@ -38,6 +38,10 @@ public class HandlerRepository { * job name */ JOB_NAME, + /** + * params of jobhandler + */ + EXECUTOR_HANDLER, /** * params of jobhandler */ @@ -60,21 +64,27 @@ public class HandlerRepository { LOG_DATE } public enum ActionEnum{RUN, KILL, LOG, BEAT} - + + // jobhandler repository + private static ConcurrentHashMap handlerRepository = new ConcurrentHashMap(); + public static void registJobHandler(String name, IJobHandler jobHandler){ + handlerRepository.put(name, jobHandler); + logger.info("xxl-job register jobhandler success, name:{}, jobHandler:{}", name, jobHandler); + } + + // thread repository of jobhandler public static ConcurrentHashMap handlerTreadMap = new ConcurrentHashMap(); - - // regist handler - public static void regist(String handleName, IJobHandler handler){ + public static HandlerThread registJobHandlerThread(String jobkey, IJobHandler handler){ HandlerThread handlerThread = new HandlerThread(handler); handlerThread.start(); - handlerTreadMap.put(handleName, handlerThread); // putIfAbsent - logger.info(">>>>>>>>>>> xxl-job regist handler success, handleName:{}, handler:{}", new Object[]{handleName, handler}); + logger.info(">>>>>>>>>>> xxl-job regist handler success, jobkey:{}, handler:{}", new Object[]{jobkey, handler}); + return handlerTreadMap.put(jobkey, handlerThread); // putIfAbsent } - + // handler push to queue public static String service(Map _param) { logger.debug(">>>>>>>>>>> xxl-job service start, _param:{}", new Object[]{_param}); - + // callback RemoteCallBack callback = new RemoteCallBack(); callback.setStatus(RemoteCallBack.FAIL); @@ -91,7 +101,7 @@ public class HandlerRepository { callback.setMsg("Timestamp check failed."); return JacksonUtil.writeValueAsString(callback); } - + // parse namespace if (namespace.equals(ActionEnum.RUN.name())) { @@ -110,20 +120,40 @@ public class HandlerRepository { return JacksonUtil.writeValueAsString(callback); } + // load old thread String jobKey = job_group.concat("_").concat(job_name); HandlerThread handlerThread = handlerTreadMap.get(jobKey); + if ("0".equals(handler_glue_switch)) { // bean model - if (handlerThread == null) { - callback.setMsg("handler for jobKey=[" + jobKey + "] not found."); + + // handler name + String executor_handler = _param.get(HandlerParamEnum.EXECUTOR_HANDLER.name()); + if (executor_handler==null || executor_handler.trim().length()==0){ + callback.setMsg("EXECUTOR_HANDLER is null."); return JacksonUtil.writeValueAsString(callback); } + + // handler instance + IJobHandler jobHandler = handlerRepository.get(executor_handler); + + if (handlerThread == null) { + // jobhandler match + if (jobHandler==null) { + callback.setMsg("handler for jobKey=[" + jobKey + "] not found."); + return JacksonUtil.writeValueAsString(callback); + } + handlerThread = HandlerRepository.registJobHandlerThread(jobKey, jobHandler); + } else { + if (handlerThread.getHandler() != jobHandler) { + handlerThread = HandlerRepository.registJobHandlerThread(jobKey, jobHandler); + } + } } else { // glue - if (handlerThread==null) { - HandlerRepository.regist(jobKey, new GlueJobHandler(job_group, job_name)); + if (handlerThread == null) { + handlerThread = HandlerRepository.registJobHandlerThread(jobKey, new GlueJobHandler(job_group, job_name)); } - handlerThread = handlerTreadMap.get(jobKey); } // push data to queue @@ -145,7 +175,7 @@ public class HandlerRepository { IJobHandler handler = handlerThread.getHandler(); handlerThread.toStop(); handlerThread.interrupt(); - regist(jobKey, handler); + HandlerRepository.registJobHandlerThread(jobKey, handler); callback.setStatus(RemoteCallBack.SUCCESS); } else { callback.setMsg("handler for jobKey=[" + jobKey + "] not found."); @@ -179,11 +209,11 @@ public class HandlerRepository { callback.setMsg("param[Action] is not valid."); return JacksonUtil.writeValueAsString(callback); } - + logger.debug(">>>>>>>>>>> xxl-job service end, triggerData:{}"); - return JacksonUtil.writeValueAsString(callback); + return JacksonUtil.writeValueAsString(callback); } - + // ----------------------- for callback log ----------------------- private static LinkedBlockingQueue> callBackQueue = new LinkedBlockingQueue>(); static { diff --git a/xxl-job-executor-example/src/main/java/com/xxl/job/executor/service/jobhandler/DemoJobHandler.java b/xxl-job-executor-example/src/main/java/com/xxl/job/executor/service/jobhandler/DemoJobHandler.java index d348671d..ff6d7e52 100644 --- a/xxl-job-executor-example/src/main/java/com/xxl/job/executor/service/jobhandler/DemoJobHandler.java +++ b/xxl-job-executor-example/src/main/java/com/xxl/job/executor/service/jobhandler/DemoJobHandler.java @@ -20,7 +20,7 @@ import com.xxl.job.core.handler.annotation.JobHander; * * @author xuxueli 2015-12-19 19:43:36 */ -@JobHander(value="waimai_201607202316260736,waimai_201607202316260736") +@JobHander(value="demoJobHandler") @Service public class DemoJobHandler extends IJobHandler { private static transient Logger logger = LoggerFactory.getLogger(DemoJobHandler.class);