功能完善:

1、jetty关闭优化(来自osc上好友QQ2575029833的pr);
2、任务终止时回调优化,执行队列中的调度进行回调;
v1.3
xueli.xue 9 years ago
parent 37f98c9c1d
commit 168050dcda

@ -71,6 +71,13 @@ git.osc地址http://git.oschina.net/xuxueli0323/xxl-job
2、执行器异步回调执行日志 2、执行器异步回调执行日志
3、【重要】在 “调度中心” 支持HA的基础上扩展执行器的Failover支持支持配置多执行期地址 3、【重要】在 “调度中心” 支持HA的基础上扩展执行器的Failover支持支持配置多执行期地址
# 规划中
1、任务终止时任务队列中调度回调通过被终止的接口
2、任务执行规则自定义假如前一个任务正在执行后续调度执行规则支持自定义
串行(默认,当前逻辑):后续调度入调度队列;
并行:后续调度并行执行;
Pass后续调度被Pass
# 源码目录说明 # 源码目录说明
/xxl-job-admin 【调度中心】:负责管理调度信息,按照调度配置发出调度请求; /xxl-job-admin 【调度中心】:负责管理调度信息,按照调度配置发出调度请求;
/xxl-job-core 公共依赖 /xxl-job-core 公共依赖
@ -87,3 +94,4 @@ git.osc地址http://git.oschina.net/xuxueli0323/xxl-job
4、人人聚财金服 4、人人聚财金服
5、…… 5、……
更多接入公司欢迎在https://github.com/xuxueli/xxl-job/issues/1 登记。 更多接入公司欢迎在https://github.com/xuxueli/xxl-job/issues/1 登记。

@ -22,6 +22,7 @@ public class XxlJobLogCallbackServer {
return trigger_log_address; return trigger_log_address;
} }
Server server = null;
public void start(int callBackPort) throws Exception { public void start(int callBackPort) throws Exception {
// init address // init address
@ -32,7 +33,7 @@ public class XxlJobLogCallbackServer {
new Thread(new Runnable() { new Thread(new Runnable() {
@Override @Override
public void run() { public void run() {
Server server = new Server(); server = new Server();
server.setThreadPool(new ExecutorThreadPool(200, 200, 30000)); // 非阻塞 server.setThreadPool(new ExecutorThreadPool(200, 200, 30000)); // 非阻塞
// connector // connector
@ -59,4 +60,14 @@ public class XxlJobLogCallbackServer {
} }
public void destroy() {
if (server!=null) {
try {
server.stop();
} catch (Exception e) {
e.printStackTrace();
}
}
}
} }

@ -55,15 +55,24 @@ public final class DynamicSchedulerUtil implements ApplicationContextAware, Init
} }
// init // init
XxlJobLogCallbackServer xxlJobLogCallbackServer = null;
public void init(){ public void init(){
try { try {
// start callback server // start callback server
new XxlJobLogCallbackServer().start(callBackPort); xxlJobLogCallbackServer = new XxlJobLogCallbackServer();
xxlJobLogCallbackServer.start(callBackPort);
} catch (Exception e) { } catch (Exception e) {
e.printStackTrace(); e.printStackTrace();
} }
} }
// destroy
public void destroy(){
if (xxlJobLogCallbackServer!=null) {
xxlJobLogCallbackServer.destroy();
}
}
// xxlJobLogDao、xxlJobInfoDao // xxlJobLogDao、xxlJobInfoDao
public static IXxlJobLogDao xxlJobLogDao; public static IXxlJobLogDao xxlJobLogDao;
public static IXxlJobInfoDao xxlJobInfoDao; public static IXxlJobInfoDao xxlJobInfoDao;

@ -18,7 +18,7 @@
</bean> </bean>
<!-- 协同-调度器 --> <!-- 协同-调度器 -->
<bean id="dynamicSchedulerUtil" class="com.xxl.job.admin.core.util.DynamicSchedulerUtil" init-method="init"> <bean id="dynamicSchedulerUtil" class="com.xxl.job.admin.core.util.DynamicSchedulerUtil" init-method="init" destroy-method="destroy" >
<!-- (轻易不要变更“调度器名称”, 任务创建时会绑定该“调度器名称”) --> <!-- (轻易不要变更“调度器名称”, 任务创建时会绑定该“调度器名称”) -->
<property name="scheduler" ref="quartzScheduler"/> <property name="scheduler" ref="quartzScheduler"/>
<property name="callBackPort" value="8888"/> <property name="callBackPort" value="8888"/>

@ -29,12 +29,14 @@ public class XxlJobExecutor implements ApplicationContextAware {
this.port = port; this.port = port;
} }
// ---------------------------------- job server ------------------------------------
Server server = null;
public void start() throws Exception { public void start() throws Exception {
new Thread(new Runnable() { new Thread(new Runnable() {
@Override @Override
public void run() { public void run() {
Server server = new Server(); server = new Server();
server.setThreadPool(new ExecutorThreadPool(200, 200, 30000)); // 非阻塞 server.setThreadPool(new ExecutorThreadPool(200, 200, 30000)); // 非阻塞
// connector // connector
@ -61,6 +63,17 @@ public class XxlJobExecutor implements ApplicationContextAware {
} }
public void destroy(){
if (server!=null) {
try {
server.stop();
} catch (Exception e) {
e.printStackTrace();
}
}
}
// ---------------------------------- init job handler ------------------------------------
public static ApplicationContext applicationContext; public static ApplicationContext applicationContext;
@Override @Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {

@ -96,6 +96,12 @@ public class HandlerThread extends Thread{
params.put("status", _status.name()); params.put("status", _status.name());
params.put("msg", _msg); params.put("msg", _msg);
HandlerRepository.pushCallBack(HttpUtil.addressToUrl(log_address), params); HandlerRepository.pushCallBack(HttpUtil.addressToUrl(log_address), params);
} else {
HashMap<String, String> params = new HashMap<String, String>();
params.put("log_id", log_id);
params.put("status", JobHandleStatus.FAIL.name());
params.put("msg", "人工手动终止[业务运行中,被强制终止]");
HandlerRepository.pushCallBack(HttpUtil.addressToUrl(log_address), params);
} }
} else { } else {
i++; i++;
@ -113,6 +119,22 @@ public class HandlerThread extends Thread{
logger.info("HandlerThread Exception:", e); logger.info("HandlerThread Exception:", e);
} }
} }
// callback trigger request in queue
while(handlerDataQueue!=null && handlerDataQueue.size()>0){
Map<String, String> handlerData = handlerDataQueue.poll();
if (handlerData!=null) {
String log_address = handlerData.get(HandlerParamEnum.LOG_ADDRESS.name());
String log_id = handlerData.get(HandlerParamEnum.LOG_ID.name());
HashMap<String, String> params = new HashMap<String, String>();
params.put("log_id", log_id);
params.put("status", JobHandleStatus.FAIL.name());
params.put("msg", "人工手动终止[任务尚未执行,在调度队列中被终止]");
HandlerRepository.pushCallBack(HttpUtil.addressToUrl(log_address), params);
}
}
logger.info(">>>>>>>>>>>> xxl-job handlerThrad stoped, hashCode:{}", Thread.currentThread()); logger.info(">>>>>>>>>>>> xxl-job handlerThrad stoped, hashCode:{}", Thread.currentThread());
} }
} }

@ -12,7 +12,7 @@
<context:component-scan base-package="com.xxl.job.executor" /> <context:component-scan base-package="com.xxl.job.executor" />
<!-- 执行器 --> <!-- 执行器 -->
<bean id="xxlJobExecutor" class="com.xxl.job.core.executor.jetty.XxlJobExecutor" init-method="start"> <bean id="xxlJobExecutor" class="com.xxl.job.core.executor.jetty.XxlJobExecutor" init-method="start" destroy-method="destroy" >
<property name="port" value="9999" /> <property name="port" value="9999" />
</bean> </bean>

Loading…
Cancel
Save