Azkaban的定时调度任务分析34:定时执行下文
2016-04-12 00:00
1361 查看
下面说线程azkaban.trigger.TriggerManager.TriggerScannerThread是如何运作的
1)初始化
stop in azkaban.trigger.TriggerManager$TriggerScannerThread.xxx
run
初始化的上下文是:
java.lang.Exception
at azkaban.trigger.TriggerManager$TriggerScannerThread.xxx(TriggerManager.java:187)
at azkaban.trigger.TriggerManager$TriggerScannerThread.<init>(Trigg
Step completed: erManager.java:191)
at azkaban.trigger.TriggerManager.<init"thread=main", azkaban.trigger.TriggerManager$TriggerScannerThread.xxx(), line=188 bci=10
>(Tri188 }
ggerManager.java:67)
at azkaban.wemain[1] bapp.AzkabanWebServer.loadTriggerManager(AzkabanWebServer.java:275)
at azkaban.webapp.AzkabanWebServer.<init>(AzkabanWebServer.java:199)
at azkaban.webapp.AzkabanWebServer.main(AzkabanWebServer.java:723)
=======================================================================================
2)线程的具体运行
stop in azkaban.trigger.TriggerManager$TriggerScannerThread.run
stop in azkaban.trigger.TriggerManager$TriggerScannerThread.checkAllTriggers
stop in azkaban.trigger.TriggerManager$TriggerScannerThread.onTriggerTrigger
---另外要注意:
long scannerInterval = props.getLong("trigger.scan.interval", DEFAULT_SCANNER_INTERVAL_MS);
runnerThread = new TriggerScannerThread(scannerInterval);
设置了查看周期,默认值竟然是
public static final long DEFAULT_SCANNER_INTERVAL_MS = 60000;
竟然有1分钟。。。
果然修改此值
然后继续研究,发现
private void onTriggerTrigger(Trigger t) throws TriggerManagerException {
List<TriggerAction> actions = t.getTriggerActions();
for (TriggerAction action : actions) {
try {
logger.info("Doing trigger actions");
System.out.println("action*************************************:" + action);
action.doAction();//重点来了!!!
} catch (Exception e) {
logger.error("Failed to do action " + action.getDescription(), e);
} catch (Throwable th) {
logger.error("Failed to do action " + action.getDescription(), th);
}
}
if (t.isResetOnTrigger()) {
t.resetTriggerConditions();
t.resetExpireCondition();
} else {
t.setStatus(TriggerStatus.EXPIRED);
}
try {
triggerLoader.updateTrigger(t);
} catch (TriggerLoaderException e) {
throw new TriggerManagerException(e);
}
}
如果屏蔽这行,exec就不会执行任务!
action的代码是azkaban.trigger.builtin.ExecuteFlowAction
============================================================
通过这里加代码观察
然后观察打印出来的日志:
正好验证了我的想法!因为我的任务也是6秒为周期!
==================================================================================
stop in azkaban.trigger.builtin.ExecuteFlowAction.doAction
run
==================================================================================
总结:定时把一个任务放入到web的一个queue中,而一次性任务则是放置1次!
1)初始化
stop in azkaban.trigger.TriggerManager$TriggerScannerThread.xxx
run
初始化的上下文是:
java.lang.Exception
at azkaban.trigger.TriggerManager$TriggerScannerThread.xxx(TriggerManager.java:187)
at azkaban.trigger.TriggerManager$TriggerScannerThread.<init>(Trigg
Step completed: erManager.java:191)
at azkaban.trigger.TriggerManager.<init"thread=main", azkaban.trigger.TriggerManager$TriggerScannerThread.xxx(), line=188 bci=10
>(Tri188 }
ggerManager.java:67)
at azkaban.wemain[1] bapp.AzkabanWebServer.loadTriggerManager(AzkabanWebServer.java:275)
at azkaban.webapp.AzkabanWebServer.<init>(AzkabanWebServer.java:199)
at azkaban.webapp.AzkabanWebServer.main(AzkabanWebServer.java:723)
=======================================================================================
2)线程的具体运行
stop in azkaban.trigger.TriggerManager$TriggerScannerThread.run
stop in azkaban.trigger.TriggerManager$TriggerScannerThread.checkAllTriggers
stop in azkaban.trigger.TriggerManager$TriggerScannerThread.onTriggerTrigger
---另外要注意:
long scannerInterval = props.getLong("trigger.scan.interval", DEFAULT_SCANNER_INTERVAL_MS);
runnerThread = new TriggerScannerThread(scannerInterval);
设置了查看周期,默认值竟然是
public static final long DEFAULT_SCANNER_INTERVAL_MS = 60000;
竟然有1分钟。。。
果然修改此值
然后继续研究,发现
private void onTriggerTrigger(Trigger t) throws TriggerManagerException {
List<TriggerAction> actions = t.getTriggerActions();
for (TriggerAction action : actions) {
try {
logger.info("Doing trigger actions");
System.out.println("action*************************************:" + action);
action.doAction();//重点来了!!!
} catch (Exception e) {
logger.error("Failed to do action " + action.getDescription(), e);
} catch (Throwable th) {
logger.error("Failed to do action " + action.getDescription(), th);
}
}
if (t.isResetOnTrigger()) {
t.resetTriggerConditions();
t.resetExpireCondition();
} else {
t.setStatus(TriggerStatus.EXPIRED);
}
try {
triggerLoader.updateTrigger(t);
} catch (TriggerLoaderException e) {
throw new TriggerManagerException(e);
}
}
如果屏蔽这行,exec就不会执行任务!
action的代码是azkaban.trigger.builtin.ExecuteFlowAction
============================================================
通过这里加代码观察
然后观察打印出来的日志:
正好验证了我的想法!因为我的任务也是6秒为周期!
==================================================================================
stop in azkaban.trigger.builtin.ExecuteFlowAction.doAction
run
==================================================================================
总结:定时把一个任务放入到web的一个queue中,而一次性任务则是放置1次!
相关文章推荐
- azkaban简介
- Azkaban-任务调度管理器
- Azkaban的Web Server源码探究系列6: alerters及插件机制分析
- Azkaban的Web Server源码探究系列7: ExecutorManager的初始化
- Azkaban的Web Server源码探究系列8: 水文一篇
- Azkaban的Web Server源码探究系列9: Servlet引擎初始化
- Azkaban的Web Server源码探究系列10 : /对应的servlet解析
- Azkaban的Web Server源码探究系列11: 登陆对应的servlet解析
- Azkaban的Web Server源码探究系列12: 首页之前的跳转
- Azkaban的Web Server源码探究系列13:首页/index的内容获取
- Azkaban的Web Server源码探究系列14:创建Project
- Azkaban的Web Server源码探究系列15:使用过程中几个需注意的配置&3.0中丢失的文件
- Azkaban的Web Server源码探究系列16:跳转Project
- Azkaban的Web Server源码探究系列17:Creating Flows
- Azkaban的Web Server源码探究系列19:loadProjectFromDir&Chek
- Azkaban的Web Server源码探究系列20:resolve&buildFlow
- Azkaban的Web Server源码探究系列21: Flow执行前的准备工作
- Azkaban的Web Server源码探究系列22: 一次性执行execute的提交准备
- Azkaban的Web Server源码探究系列23: 一次性执行execute的正式提交
- Azkaban的Web Server源码探究系列24: 一次性执行execute任务取出处理