Elastic Job 有自己 console 控制台， 可以对某些配置进行修改 ，控制中心修改对应代码部分
@PUT
@Consumes(MediaType.APPLICATION_JSON)
public void updateJobSettings(final JobSettings jobSettings) {
jobAPIService.getJobSettingsAPI().updateJobSettings(jobSettings);
}
@Override
public void updateJobSettings(final JobSettings jobSettings) {
Preconditions.checkArgument(!Strings.isNullOrEmpty(jobSettings.getJobName()), "jobName can not be empty.");
Preconditions.checkArgument(!Strings.isNullOrEmpty(jobSettings.getCron()), "cron can not be empty.");
Preconditions.checkArgument(jobSettings.getShardingTotalCount() > 0, "shardingTotalCount should larger than zero.");
JobNodePath jobNodePath = new JobNodePath(jobSettings.getJobName());
regCenter.update(jobNodePath.getConfigNodePath(), LiteJobConfigurationGsonFactory.toJsonForObject(jobSettings));
}
简单说 ，用新的配置 更新zk config 节点 value
跑job的那些jvm 实例是怎么知道的呢
CronSettingAndJobEventChangedJobListener 通过监听config 节点是否有变更的方式 更新 quartz 的trigger 调度周期 。
class CronSettingAndJobEventChangedJobListener extends AbstractJobListener {
@Override
protected void dataChanged(final String path, final Type eventType, final String data) {
if (configNode.isConfigPath(path) && Type.NODE_UPDATED == eventType && !JobRegistry.getInstance().isShutdown(jobName)) {
JobRegistry.getInstance().getJobScheduleController(jobName).rescheduleJob(LiteJobConfigurationGsonFactory.fromJson(data).getTypeConfig().getCoreConfig().getCron());
}
}
}
ShardingTotalCountChangedJobListener 通过监听config 节点是否有变更的方式 更新分片总数
设定 重分片flag
MonitorExecutionSettingsChangedJobListener 通过监听config 节点是否有变更的方式 如果 monitorExecution 为false
就会删除 zk sharding/id/running 节点
FailoverSettingsChangedJobListener 通过监听config 节点是否有变更的方式 如果 failover ＝false 就会删除 zk sharding/id/failover 节点
除了这4个, 还有 misfire 等 通过 直接从 treenode cache 中读取 如下
@Override
public boolean isExecuteMisfired(final Collection<Integer> shardingItems) {
return isEligibleForJobRunning() && configService.load(true).getTypeConfig().getCoreConfig().isMisfire() && !executionService.getMisfiredJobItems(shardingItems).isEmpty();
}
这里有个疑问 配置更新 时 ， 是否回对正在运行job 会有影响呢 ， misfire ，failover ，总分片数 ，monitorExecution 都会影响正在运行的 分片 。
但是 cron trigger 的变更 会？ 这个就要看 QuartzScheduler 的rescheduleJob 以及 CronTriggerImpl 的 computeFirstFireTime 方法
public Date rescheduleJob(TriggerKey triggerKey,
Trigger newTrigger) throws SchedulerException {
validateState();
if (triggerKey == null) {
throw new IllegalArgumentException("triggerKey cannot be null");
}
if (newTrigger == null) {
throw new IllegalArgumentException("newTrigger cannot be null");
}
OperableTrigger trig = (OperableTrigger)newTrigger;
Trigger oldTrigger = getTrigger(triggerKey);
if (oldTrigger == null) {
return null;
} else {
trig.setJobKey(oldTrigger.getJobKey());
}
trig.validate();
Calendar cal = null;
if (newTrigger.getCalendarName() != null) {
cal = resources.getJobStore().retrieveCalendar(
newTrigger.getCalendarName());
}
Date ft = trig.computeFirstFireTime(cal);
if (ft == null) {
throw new SchedulerException(
"Based on configured schedule, the given trigger will never fire.");
}
if (resources.getJobStore().replaceTrigger(triggerKey, trig)) {
notifySchedulerThread(newTrigger.getNextFireTime().getTime());
notifySchedulerListenersUnscheduled(triggerKey);
notifySchedulerListenersSchduled(newTrigger);
} else {
return null;
}
return ft;
}
public Date computeFirstFireTime(org.quartz.Calendar calendar) {
nextFireTime = getFireTimeAfter(new Date(getStartTime().getTime() - 1000l));
while (nextFireTime != null && calendar != null
&& !calendar.isTimeIncluded(nextFireTime.getTime())) {
nextFireTime = getFireTimeAfter(nextFireTime);
}
return nextFireTime;
}
简单看 时间的话 不会
小疑问 1 ， 1个 jvm 实例 ， 实例 在启动过程中，在运行分片前 ， disable 掉这个实例 ， 这时候 这个实例 会干嘛 。 简单回答的话 最大开始时间超时时间过去后 会报超时异常。 这个job 不会调用 doBeforeJobExecutedAtLastStarted，然后 会报 最大结束时间超时异常 不会调用doAfterJobExecutedAtLastCompleted
还有一点 不要最大开始时间超时时间 ，结束时间 设置为负数 ， 这样 线程 会被阻塞很长时间 ,Long.MAX_VALUE
this.startedTimeoutMilliseconds = Long.MAX_VALUE;
点击console 触发 在 作业运行时 ，没有任何作用
class JobTriggerStatusJobListener extends AbstractJobListener {
@Override
protected void dataChanged(final String path, final Type eventType, final String data) {
if (!InstanceOperation.TRIGGER.name().equals(data) || !instanceNode.isLocalInstancePath(path) || Type.NODE_UPDATED != eventType) {
return;
}
instanceService.clearTriggerFlag();
if (!JobRegistry.getInstance().isShutdown(jobName) && !JobRegistry.getInstance().isJobRunning(jobName)) {
// TODO 目前是作业运行时不能触发, 未来改为堆积式触发
JobRegistry.getInstance().getJobScheduleController(jobName).triggerJob();
}
}
}
