功能:
定期去检查action的状态,包含coord的 和wf的;
/**
* {@link ActionCheckRunnable} is the runnable which is scheduled to run and
* queue Action checks.
*/
static class ActionCheckRunnable implements Runnable {
private int actionCheckDelay;
private List<XCallable<Void>> callables;
private StringBuilder msg = null;
public ActionCheckRunnable(int actionCheckDelay) {
this.actionCheckDelay = actionCheckDelay;
}
public void run() {
XLog.Info.get().clear();
XLog LOG = XLog.getLog(getClass());
msg = new StringBuilder();
try {
runWFActionCheck();
runCoordActionCheck();
}
catch (CommandException ce) {
LOG.error("Unable to run action checks, ", ce);
}
LOG.debug("QUEUING [{0}] for potential checking", msg.toString());
if (null != callables) {
boolean ret = Services.get().get(CallableQueueService.class).queueSerial(callables);
if (ret == false) {
LOG.warn("Unable to queue the callables commands for CheckerService. "
+ "Most possibly command queue is full. Queue size is :" + Services.get().get(CallableQueueService.class).queueSize());
}
callables = null;
}
}
/**
* check workflow actions
*
* @throws CommandException
*/
private void runWFActionCheck() throws CommandException {
JPAService jpaService = Services.get().get(JPAService.class);
if (jpaService == null) {
throw new CommandException(ErrorCode.E0610);
}
List<WorkflowActionBean> actions;
try {
actions = WorkflowActionQueryExecutor.getInstance().getList(WorkflowActionQuery.GET_RUNNING_ACTIONS,
actionCheckDelay);
}
catch (JPAExecutorException je) {
throw new CommandException(je);
}
if (actions == null || actions.isEmpty()) {
return;
}
List<String> actionIds = toIds(actions);
try {
actionIds = Services.get().get(JobsConcurrencyService.class).getJobIdsForThisServer(actionIds);
}
catch (Exception ex) {
throw new CommandException(ErrorCode.E1700, ex.getMessage(), ex);
}
msg.append(" WF_ACTIONS : ").append(actionIds.size());
for (String actionId : actionIds) {
Services.get().get(InstrumentationService.class).get().incr(INSTRUMENTATION_GROUP,
INSTR_CHECK_ACTIONS_COUNTER, 1);
queueCallable(new ActionCheckXCommand(actionId));
}
}
/**
* check coordinator actions
*
* @throws CommandException
*/
private void runCoordActionCheck() throws CommandException {
JPAService jpaService = Services.get().get(JPAService.class);
if (jpaService == null) {
throw new CommandException(ErrorCode.E0610);
}
List<String> cactionIds;
try {
cactionIds = jpaService.execute(new CoordActionsRunningGetJPAExecutor(
actionCheckDelay));
}
catch (JPAExecutorException je) {
throw new CommandException(je);
}
if (cactionIds == null || cactionIds.isEmpty()) {
return;
}
try {
cactionIds = Services.get().get(JobsConcurrencyService.class).getJobIdsForThisServer(cactionIds);
}
catch (Exception ex) {
throw new CommandException(ErrorCode.E1700, ex.getMessage(), ex);
}
msg.append(" COORD_ACTIONS : ").append(cactionIds.size());
for (String coordActionId : cactionIds) {
Services.get().get(InstrumentationService.class).get().incr(INSTRUMENTATION_GROUP,
INSTR_CHECK_COORD_ACTIONS_COUNTER, 1);
queueCallable(new CoordActionCheckXCommand(coordActionId, actionCheckDelay));
}
}
/**
* Adds callables to a list. If the number of callables in the list
* reaches {@link ActionCheckerService#CONF_CALLABLE_BATCH_SIZE}, the
* entire batch is queued and the callables list is reset.
*
* @param callable the callable to queue.
*/
private void queueCallable(XCallable<Void> callable) {
if (callables == null) {
callables = new ArrayList<XCallable<Void>>();
}
callables.add(callable);
if (callables.size() == ConfigurationService.getInt(CONF_CALLABLE_BATCH_SIZE)) {
boolean ret = Services.get().get(CallableQueueService.class).queueSerial(callables);
if (ret == false) {
XLog.getLog(getClass()).warn(
"Unable to queue the callables commands for CheckerService. "
+ "Most possibly command queue is full. Queue size is :"
+ Services.get().get(CallableQueueService.class).queueSize());
}
callables = new ArrayList<XCallable<Void>>();
}
}
private List<String> toIds(List<WorkflowActionBean> actions) {
List<String> ids = new ArrayList<String>(actions.size());
for (WorkflowActionBean action : actions) {
ids.add(action.getId());
}
return ids;
}
}
定时调度的主要工作就是对于处于running状态超过一个时间的任务,不管是wf的还是coord的任务,异步发起action check的命令,来检查任务的状态。