ElasticJob源码学习

1. spring命名空间 NamespaceHandlerSupport

 <reg:zookeeper namespace="${elastic.jobNamespace}" />

RegNamespaceHandler->ZookeeperBeanDefinitionParser.parseInternal
->ZookeeperRegistryCenter.init

<job:simple id="elBatchInitJob" class="com.xxx.clearingcenter.jobs.batch.BatchInitJob" />

JobNamespaceHandler->SimpleJobBeanDefinitionParser->AbstractJobBeanDefinitionParser.parseInternal->SpringJobScheduler.init->JobScheduler.init

2. ZookeeperRegistryCenter.init

CuratorFramework client = builder.build();  // 构建Zookeeper客户端
client.start();   // 启动，以便后续注册节点变化监听

3. JobScheduler对象

从spring配置即，命名空间的AbstractJobBeanDefinitionParser.parseInternal的调用来看，每一个job对应一个JobScheduler对象。
对象属性：

SchedulerFacade schedulerFacade
- ConfigurationService configService; 【config】弹性化分布式作业配置服务.
  主要功能:持久化分布式作业配置、读取作业配置。
  读取作业配置时，可通过从本地缓存TreeMap中获取，或直接client.getData()从zk上获取。
- LeaderService leaderService; 【leader/election/latch】【leader/election/instance】主节点服务
  主要功能：选举主节点electLeader、删除主节点removeLeader、判断是否主节点isLeaderUntilBlock,isLeader、判断是否已经有主节点hasLeader
- ServerService serverService;【servers】作业服务器服务
  persistOnline将本机IP持久化到作业服务器信息中。
- InstanceService instanceService; 【instances】作业运行实例服务,同一台服务器可能有多个运行实例的情况
- ShardingService shardingService; 【sharding/{item}/running】分片服务
- ExecutionService executionService; 【sharding/{item}/running】执行作业服务，主要注册作业启动、作业完成信息。
- MonitorService monitorService;
- ReconcileService reconcileService; 调解分布式作业不一致状态服务
- ListenerManager listenerManager; 作业注册中心的监听器管理者,包涵选主、分片、失效转移等监听器
JobFacade jobFacade
- ConfigurationService configService
- ShardingService shardingService
- ExecutionContextService executionContextService
- ExecutionService executionService
- FailoverService failoverService
- List<ElasticJobListener> elasticJobListeners
- JobEventBus jobEventBus
LiteJobConfiguration liteJobConfig Lite作业配置 job的配置信息
CoordinatorRegistryCenter regCenter 用于协调分布式服务的注册中心.

4.JobScheduler.init

        //@1
        LiteJobConfiguration liteJobConfigFromRegCenter = schedulerFacade.updateJobConfiguration(liteJobConfig);
        //@2
        JobRegistry.getInstance().setCurrentShardingTotalCount(liteJobConfigFromRegCenter.getJobName(), liteJobConfigFromRegCenter.getTypeConfig().getCoreConfig().getShardingTotalCount());
        //@3
        JobScheduleController jobScheduleController = new JobScheduleController(
                createScheduler(), createJobDetail(liteJobConfigFromRegCenter.getTypeConfig().getJobClass()), liteJobConfigFromRegCenter.getJobName());
        JobRegistry.getInstance().registerJob(liteJobConfigFromRegCenter.getJobName(), jobScheduleController, regCenter);
        //@4
        schedulerFacade.registerStartUpInfo(!liteJobConfigFromRegCenter.isDisabled());
        //@5
        jobScheduleController.scheduleJob(liteJobConfigFromRegCenter.getTypeConfig().getCoreConfig().getCron());

@1:持久化作业配置信息。

// liteJobConfig以json格式，持久化保存在/{jobNamespace}/{jobName}/config
 jobNodeStorage.replaceJobNode(ConfigurationNode.ROOT, LiteJobConfigurationGsonFactory.toJson(liteJobConfig));

@2:在JobRegistry中设置当前分片总数

currentShardingTotalCountMap.put(jobName, currentShardingTotalCount);

@3:创建quartz的调度器

1.createScheduler:StdSchedulerFactory.getScheduler->initialize创建QuartzScheduler

factory.initialize(getBaseQuartzProperties());

2.createJobDetail：jobDetail 对应 LiteJob,再看下LiteJob的execute方法！

 JobDetail result = JobBuilder.newJob(LiteJob.class).withIdentity(liteJobConfig.getJobName()).build();

@4: registerStartUpInfo 作业快要启动了！！

        // 1.启动作业监听服务
        listenerManager.startAllListeners();
        // 2.选举主节点
        leaderService.electLeader();
        // 3.持久化服务器上线服务
        serverService.persistOnline(enabled);
        // 4.持久化作业运行实例上线相关信息
        instanceService.persistOnline();
        // 5.重新分片
        shardingService.setReshardingFlag();
        monitorService.listen();
        if (!reconcileService.isRunning()) {
            reconcileService.startAsync();
        }

1.startAllListeners，启动监听器，监听zk上节点的变动，可能触发重新选主、重新分片

   /**
     * 开启所有监听器.
     */
    public void startAllListeners() {
         // LeaderElectionJobListener 选主事件监听器，监听节点主节点LeaderNode.INSTANCE (主节点与zk失联)
        // LeaderAbdicationJobListener 主退位监听器(主节点被配置成禁用的情况)
        electionListenerManager.start();  
        // 
        shardingListenerManager.start();
        failoverListenerManager.start();
        monitorExecutionListenerManager.start();
        shutdownListenerManager.start();
        triggerListenerManager.start();
        rescheduleListenerManager.start();
        guaranteeListenerManager.start();
        jobNodeStorage.addConnectionStateListener(regCenterConnectionStateListener);
    }

2.选举主节点过程

/**
     * 在主节点执行操作.
     * 
     * @param latchNode 分布式锁使用的作业节点名称
     * @param callback 执行操作的回调
     */
    public void executeInLeader(final String latchNode, final LeaderExecutionCallback callback) {
        try (LeaderLatch latch = new LeaderLatch(getClient(), jobNodePath.getFullPath(latchNode))) {
            latch.start();
            latch.await();
            callback.execute();
        //CHECKSTYLE:OFF
        } catch (final Exception ex) {
        //CHECKSTYLE:ON
            handleException(ex);
        }
    }

使用cautor开源框架提供的实现类org.apache.curator.framework.recipes.leader.LeaderLatch。
LeaderLatch需要传入两个参数：
CuratorFramework client：curator框架客户端。
latchPath：锁节点路径：/{jobNamespace}/{jobname}/leader/election/latch。
LeaderLatch.start，其主要过程就是去锁路径下创建一个临时排序节点，如果创建的节点序号最小，await方法将返回，否则在前一个节点监听该节点事件，并阻塞,如何获得分布式锁后，执行callback回调方法。
LeaderService$LeaderElectionExecutionCallback

@RequiredArgsConstructor
    class LeaderElectionExecutionCallback implements LeaderExecutionCallback {
        @Override
        public void execute() {
            if (!hasLeader()) {
                jobNodeStorage.fillEphemeralJobNode(LeaderNode.INSTANCE, JobRegistry.getInstance().getJobInstance(jobName).getJobInstanceId());
            }
        }
    }

成功获取选主的分布式锁后，如果{jobNamespace}/{jobname}/leader/election/instance节点不存在，则创建该临时节点，节点存储的内容为IP地址@-@进程ID,
其代码见JobInstance构造方法：jobInstanceId = IpUtils.getIp() + “@-@”+ ManagementFactory.getRuntimeMXBean().getName().split(“@”)[0];

3.持久化服务器上线服务 {jobNamespace}/{jobname}/servers/IP地址
4.持久化作业运行实例上线相关信息 {jobNamespace}/{jobname}/instances/IP地址@-@进程ID
5.设置需要重新分片 {jobNamespace}/{jobname}/leader/sharding/necessary

@5:启动调度作业

  scheduler.start();

5.作业执行Job.execute

前面已经分析在jobSchedule的初始化过程中，quartz的jobDetail使用的是LiteJob.

@Override
    public void execute(final JobExecutionContext context) throws JobExecutionException {
        JobExecutorFactory.getJobExecutor(elasticJob, jobFacade).execute();
    }

/**
     * 获取作业执行器.
     *
     * @param elasticJob 分布式弹性作业
     * @param jobFacade 作业内部服务门面服务
     * @return 作业执行器
     */
    @SuppressWarnings("unchecked")
    public static AbstractElasticJobExecutor getJobExecutor(final ElasticJob elasticJob, final JobFacade jobFacade) {
        if (null == elasticJob) {
            return new ScriptJobExecutor(jobFacade);
        }
        if (elasticJob instanceof SimpleJob) {
            return new SimpleJobExecutor((SimpleJob) elasticJob, jobFacade);
        }
        if (elasticJob instanceof DataflowJob) {
            return new DataflowJobExecutor((DataflowJob) elasticJob, jobFacade);
        }
        throw new JobConfigurationException("Cannot support job type '%s'", elasticJob.getClass().getCanonicalName());
    }

在项目中spring配置的是simple,调用SimpleJobExecutor.execute()->AbstractElasticJobExecutor.execute()

/**
     * 执行作业.
     */
    public final void execute() {
        // 1. 检查环境配置
        try {
            jobFacade.checkJobExecutionEnvironment();
        } catch (final JobExecutionEnvironmentException cause) {
            jobExceptionHandler.handleException(jobName, cause);
        }
     //  2.获取当前作业服务器的分片信息
        ShardingContexts shardingContexts = jobFacade.getShardingContexts();
        if (shardingContexts.isAllowSendJobEvent()) {
            // 发布作业状态追踪事件.
            jobFacade.postJobStatusTraceEvent(shardingContexts.getTaskId(), State.TASK_STAGING, String.format("Job '%s' execute begin.", jobName));
        }
        // 设置任务被错过执行的标记.
        if (jobFacade.misfireIfRunning(shardingContexts.getShardingItemParameters().keySet())) {
            if (shardingContexts.isAllowSendJobEvent()) {
                // 发布作业状态追踪事件.
                jobFacade.postJobStatusTraceEvent(shardingContexts.getTaskId(), State.TASK_FINISHED, String.format(
                        "Previous job '%s' - shardingItems '%s' is still running, misfired job will start after previous job completed.", jobName, 
                        shardingContexts.getShardingItemParameters().keySet()));
            }
            return;
        }

        // 执行作业前置监听
        try {
            jobFacade.beforeJobExecuted(shardingContexts);
            //CHECKSTYLE:OFF
        } catch (final Throwable cause) {
            //CHECKSTYLE:ON
            jobExceptionHandler.handleException(jobName, cause);
        }
        // 3.执行作业
        execute(shardingContexts, JobExecutionEvent.ExecutionSource.NORMAL_TRIGGER);
        while (jobFacade.isExecuteMisfired(shardingContexts.getShardingItemParameters().keySet())) {
            jobFacade.clearMisfire(shardingContexts.getShardingItemParameters().keySet());
            execute(shardingContexts, JobExecutionEvent.ExecutionSource.MISFIRE);
        }
        // 如果需要失效转移, 则执行作业失效转移
        jobFacade.failoverIfNecessary();
        try {
            jobFacade.afterJobExecuted(shardingContexts);
            //CHECKSTYLE:OFF
        } catch (final Throwable cause) {
            //CHECKSTYLE:ON
            jobExceptionHandler.handleException(jobName, cause);
        }
    }

1. 检查环境配置

 /**
     * 检查本机与注册中心的时间误差秒数是否在允许范围.
     * 
     * @throws JobExecutionEnvironmentException 本机与注册中心的时间误差秒数不在允许范围所抛出的异常
     */
    public void checkMaxTimeDiffSecondsTolerable() throws JobExecutionEnvironmentException {
        int maxTimeDiffSeconds =  load(true).getMaxTimeDiffSeconds();
        if (-1  == maxTimeDiffSeconds) {
            return;
        }
        long timeDiff = Math.abs(timeService.getCurrentMillis() - jobNodeStorage.getRegistryCenterTime());
        if (timeDiff > maxTimeDiffSeconds * 1000L) {
            throw new JobExecutionEnvironmentException(
                    "Time different between job server and register center exceed '%s' seconds, max time different is '%s' seconds.", timeDiff / 1000, maxTimeDiffSeconds);
        }
    }

2.获取当前作业服务器的分片信息

@Override
    public ShardingContexts getShardingContexts() {
        boolean isFailover = configService.load(true).isFailover();
        if (isFailover) {
            List<Integer> failoverShardingItems = failoverService.getLocalFailoverItems();
            if (!failoverShardingItems.isEmpty()) {
                return executionContextService.getJobShardingContext(failoverShardingItems);
            }
        }
        // 进行分片
        shardingService.shardingIfNecessary();
        List<Integer> shardingItems = shardingService.getLocalShardingItems();
        if (isFailover) {
            shardingItems.removeAll(failoverService.getLocalTakeOffItems());
        }
        shardingItems.removeAll(executionService.getDisabledItems(shardingItems));
        return executionContextService.getJobShardingContext(shardingItems);
    }

分片处理：

public void shardingIfNecessary() {
        // 获取当前可用的作业实例,client实时读取zk-instances节点
        List<JobInstance> availableJobInstances = instanceService.getAvailableJobInstances();
        // 判断是否需要分片 {jobNamespace}/{jobname}/leader/sharding/necessary 是否存在，前面启动过程中，已经添加了该节点
        // 且可用的作业实例不能为空
        if (!isNeedSharding() || availableJobInstances.isEmpty()) {
            return;
        }
        // 如果不是主节点，阻塞直到分片完成,阻塞检查间隔是100毫秒
        if (!leaderService.isLeaderUntilBlock()) {
            blockUntilShardingCompleted();
            return;
        }
        // 进入到这，说明当前节点已经成为主节点
        // 检查并等待 是否还有执行中的作业：遍历每个item ，查看是否存在节点/sharding/{item}/running  会不会执行一半就挂了，导致running没有删除
        waitingOtherShardingItemCompleted();
        // 读取作业配置 这里的fromCache=false,即不从缓存中获取，直接获取zk上最新的config配置信息
        LiteJobConfiguration liteJobConfig = configService.load(false);
        int shardingTotalCount = liteJobConfig.getTypeConfig().getCoreConfig().getShardingTotalCount();
        log.debug("Job '{}' sharding begin.", jobName);
        // 填充临时节点 /leader/sharding/processing 代表当前正在分片中
        jobNodeStorage.fillEphemeralJobNode(ShardingNode.PROCESSING, "");
        // 重新调整sharding下的item节点，删除item下的instance,并调整item,总分片可能增加或者减少
        resetShardingInfo(shardingTotalCount);
        // 获取作业分片策略实例 这里用的是AverageAllocationJobShardingStrategy 平均分片策略
        /* 基于平均分配算法的分片策略.
         * 
         * 
         * 如果分片不能整除, 则不能整除的多余分片将依次追加到序号小的服务器.
         * 如: 
         * 1. 如果有3台服务器, 分成9片, 则每台服务器分到的分片是: 1=[0,1,2], 2=[3,4,5], 3=[6,7,8].
         * 2. 如果有3台服务器, 分成8片, 则每台服务器分到的分片是: 1=[0,1,6], 2=[2,3,7], 3=[4,5].
         * 3. 如果有3台服务器, 分成10片, 则每台服务器分到的分片是: 1=[0,1,2,9], 2=[3,4,5], 3=[6,7,8].
         */
        JobShardingStrategy jobShardingStrategy = JobShardingStrategyFactory.getStrategy(liteJobConfig.getJobShardingStrategyClass());
        // 执行作业分片
        //并将分片信息持久化到zk上，删除/leader/sharding/processing和/leader/sharding/necessary (PersistShardingInfoTransactionExecutionCallback.execute)
        // 提交事务 curatorTransactionFinal.commit();
        jobNodeStorage.executeInTransaction(new PersistShardingInfoTransactionExecutionCallback(jobShardingStrategy.sharding(availableJobInstances, jobName, shardingTotalCount)));
        log.debug("Job '{}' sharding complete.", jobName);
    }

3.执行作业

// process-1
process(shardingContexts, executionSource);

process-1

private void process(final ShardingContexts shardingContexts, final JobExecutionEvent.ExecutionSource executionSource) {
        Collection<Integer> items = shardingContexts.getShardingItemParameters().keySet();
        // 只有一个分片
        if (1 == items.size()) {
            int item = shardingContexts.getShardingItemParameters().keySet().iterator().next();
            JobExecutionEvent jobExecutionEvent =  new JobExecutionEvent(shardingContexts.getTaskId(), jobName, executionSource, item);
            // process-2立刻执行
            process(shardingContexts, item, jobExecutionEvent);
            return;
        }
        final CountDownLatch latch = new CountDownLatch(items.size());
        for (final int each : items) {
            final JobExecutionEvent jobExecutionEvent = new JobExecutionEvent(shardingContexts.getTaskId(), jobName, executionSource, each);
            if (executorService.isShutdown()) {
                return;
            }
            // 多个分片并发执行
            executorService.submit(new Runnable() {
                
                @Override
                public void run() {
                    try {
                        // process-2
                        process(shardingContexts, each, jobExecutionEvent);
                    } finally {
                        latch.countDown();
                    }
                }
            });
        }
        try {
            latch.await();
        } catch (final InterruptedException ex) {
            Thread.currentThread().interrupt();
        }
    }

process-2

// 执行任务
    private void process(final ShardingContexts shardingContexts, final int item, final JobExecutionEvent startEvent) {
        if (shardingContexts.isAllowSendJobEvent()) {
            jobFacade.postJobExecutionEvent(startEvent);
        }
        log.trace("Job '{}' executing, item is: '{}'.", jobName, item);
        JobExecutionEvent completeEvent;
        try {
            // process-3
            process(new ShardingContext(shardingContexts, item));
            completeEvent = startEvent.executionSuccess();
            log.trace("Job '{}' executed, item is: '{}'.", jobName, item);
            if (shardingContexts.isAllowSendJobEvent()) {
                jobFacade.postJobExecutionEvent(completeEvent);
            }
            // CHECKSTYLE:OFF
        } catch (final Throwable cause) {
            // CHECKSTYLE:ON
            completeEvent = startEvent.executionFailure(cause);
            jobFacade.postJobExecutionEvent(completeEvent);
            itemErrorMessages.put(item, ExceptionUtil.transform(cause));
            jobExceptionHandler.handleException(jobName, cause);
        }
    }

process-3
SimpleJobExecutor.process

@Override
  protected void process(final ShardingContext shardingContext) {
      simpleJob.execute(shardingContext);
  }

/**
* 简单分布式作业接口.
* 
* @author zhangliang
*/
public interface SimpleJob extends ElasticJob {
  
  /**
   * 执行作业.
   *
   * @param shardingContext 分片上下文
   */
  void execute(ShardingContext shardingContext);
}

到这里就比较熟悉了，项目中自己定义的job就实现了这个接口的execute方法。

最后编辑于：2018.11.04 18:55:28

人面猴
序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
沈念sama阅读 222,590评论 6赞 517
死咒
序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
沈念sama阅读 95,157评论 3赞 399
救了他两次的神仙让他今天三更去死
文/潘晓璐我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
开封第一讲书人阅读 169,301评论 0赞 362
道士缉凶录：失踪的卖姜人
文/不坏的土叔我叫张陵，是天一观的道长。经常有香客问我，道长，这世上最难降的妖魔是什么？我笑而不...
开封第一讲书人阅读 60,078评论 1赞 300
港岛之恋（遗憾婚礼）
正文为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
茶点故事阅读 69,082评论 6赞 398
恶毒庶女顶嫁案：这布局不是一般人想出来的
文/花漫我一把揭开白布。她就那样静静地躺着，像睡着了一般。火红的嫁衣衬着肌肤如雪。梳的纹丝不乱的头发上，一...
开封第一讲书人阅读 52,682评论 1赞 312
城市分裂传说
那天，我揣着相机与录音，去河边找鬼。笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。我是一名探鬼主播，决...
沈念sama阅读 41,155评论 3赞 422
双鸳鸯连环套：你想象不到人心有多黑
文/苍兰香墨我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
开封第一讲书人阅读 40,098评论 0赞 277
万荣杀人案实录
序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
沈念sama阅读 46,638评论 1赞 319
护林员之死
正文独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋以下内容为张勋视角年9月15日...
茶点故事阅读 38,701评论 3赞 342
白月光启示录
正文我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
茶点故事阅读 40,852评论 1赞 353
活死人
序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
沈念sama阅读 36,520评论 5赞 351
日本核电站爆炸内幕
正文年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
茶点故事阅读 42,181评论 3赞 335
男人毒药：我在死后第九天来索命
文/蒙蒙一、第九天我趴在偏房一处隐蔽的房顶上张望。院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
开封第一讲书人阅读 32,674评论 0赞 25
一桩弑父案，背后竟有这般阴谋
文/苍兰香墨我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。一阵脚步声响...
开封第一讲书人阅读 33,788评论 1赞 274
情欲美人皮
我被黑心中介骗来泰国打工，没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。一个月前我还...
沈念sama阅读 49,279评论 3赞 379
代替公主和亲
正文我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
茶点故事阅读 45,851评论 2赞 361