Elastic Job 配置管理源码分析

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();
        }
    }
}