elastic-job 源码解读之job配置数据存储

​ elastic-job作业相关的数据都是配置在zk上的，包括分片参数，作业失效转移，运行实例等等都是保存在ZK上的，那具体的zk节点的树形结构会是什么样子？每一个节点又是什么时候注册到zk上的？


Job.png

​ 在job的启动过程中(JobScheduler.init())，会将启动信息注册到注册中心，再看一下具体的节点信息：

 public void init() {
        ///{jobName}/config路径在这里
        LiteJobConfiguration liteJobConfigFromRegCenter =
          schedulerFacade.updateJobConfiguration(liteJobConfig); JobRegistry.getInstance().setCurrentShardingTotalCount(liteJobConfigFromRegCenter.getJobName(), liteJobConfigFromRegCenter.getTypeConfig().getCoreConfig().getShardingTotalCount());
        JobScheduleController jobScheduleController = new JobScheduleController(
                createScheduler(), createJobDetail(liteJobConfigFromRegCenter.getTypeConfig().getJobClass()), liteJobConfigFromRegCenter.getJobName());
        JobRegistry.getInstance().registerJob(liteJobConfigFromRegCenter.getJobName(), jobScheduleController, regCenter);
        /** 
        /{jobName}/leader/election/latch
        /leader/election/instance
        /{jobName}/services/{ServerIp}
        /{jobName}/instances/{instanceIndex}    
        /{jobName}/sharding/necessary   
        **/
        schedulerFacade.registerStartUpInfo(!liteJobConfigFromRegCenter.isDisabled());
        jobScheduleController.scheduleJob(liteJobConfigFromRegCenter.getTypeConfig().getCoreConfig().getCron());
}

持久化job的配置信息，首先将job的配置信息持久到zk节点上,看代码：

LiteJobConfiguration liteJobConfigFromRegCenter=schedulerFacade.updateJobConfiguration(liteJobConfig);

   public LiteJobConfiguration updateJobConfiguration(final LiteJobConfiguration liteJobConfig) {
          configService.persist(liteJobConfig);//
          return configService.load(false);
   }

   public void persist(final LiteJobConfiguration liteJobConfig) {
        checkConflictJob(liteJobConfig);
        //configurationNode.ROOT=/{jobName}/config
        if (!jobNodeStorage.isJobNodeExisted(ConfigurationNode.ROOT) || liteJobConfig.isOverwrite()) {
            jobNodeStorage.replaceJobNode(ConfigurationNode.ROOT, LiteJobConfigurationGsonFactory.toJson(liteJobConfig));
        }
    }

  public void replaceJobNode(final String node, final Object value) {
  /** 
    节点：
    /{jobName}/config
        在这里注册
  **/
        regCenter.persist(jobNodePath.getFullPath(node), value.toString());
    }

在job启动注册启动信息的时候，会注册很多信息，具体如下：

//JobScheduler.init();
schedulerFacade.registerStartUpInfo(!liteJobConfigFromRegCenter.isDisabled());

public void registerStartUpInfo(final boolean enabled) {
  listenerManager.startAllListeners();
  /** 
    节点：
    /{jobName}/leader/election/latch
    /{jobName}/leader/election/instance
    在这里实现
  **/
  leaderService.electLeader();
  /**
  节点：
    /{jobName}/servers/{ServerIp}   
     在这里创建
  **/
  serverService.persistOnline(enabled);
  /**
  节点：
    /{jobName}/instances/{instanceId}   
  在这里创建
  **/
  instanceService.persistOnline();
  /**
  节点：
    /{jobName}/sharding/necessary   
  在这里创建
  **/
  shardingService.setReshardingFlag();
  monitorService.listen();
  if (!reconcileService.isRunning()) {
    reconcileService.startAsync();
  }
}


public void electLeader() {
  log.debug("Elect a new leader now.");
  //
  //选举主节点 在主节点下面创建节点LeaderNode.LATCH=/{jobName}/leader/election/latch
  jobNodeStorage.executeInLeader(LeaderNode.LATCH, new LeaderElectionExecutionCallback());
  log.debug("Leader election completed.");
}

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

//在主节点选举完成之后，执行callBack
@RequiredArgsConstructor
class LeaderElectionExecutionCallback implements LeaderExecutionCallback {
  @Override
  public void execute() {
    if (!hasLeader()) {
      ///{jobName}/leader/election/instance 在这里
      jobNodeStorage.fillEphemeralJobNode(LeaderNode.INSTANCE, JobRegistry.getInstance().getJobInstance(jobName).getJobInstanceId());
    }
  }
}

再看一下执行过程，最重要的一段获取分片上下文，在获取分片上下文的时候，首先会判断是不是需要重新分片，需要分片的话，重新设置分片信息，在这里会做所有相关分片的逻辑。

//AbstractElasticJobExecutor 获取上下文
ShardingContexts shardingContexts = jobFacade.getShardingContexts();

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() {
  List<JobInstance> availableJobInstances = instanceService.getAvailableJobInstances();
  if (!isNeedSharding() || availableJobInstances.isEmpty()) {
    return;
  }
  if (!leaderService.isLeaderUntilBlock()) {
    blockUntilShardingCompleted();
    return;
  }
  waitingOtherJobCompleted();
  LiteJobConfiguration liteJobConfig = configService.load(false);
  int shardingTotalCount = liteJobConfig.getTypeConfig().getCoreConfig().getShardingTotalCount();
  log.debug("Job '{}' sharding begin.", jobName);
  //分片之前，将zk节点状态改为processing，分片中的状态，等待分片结束
  /**
  /{jobName}/sharding/processing
  **/
  jobNodeStorage.fillEphemeralJobNode(ShardingNode.PROCESSING, "");
  //重新设置分片项参数
  resetShardingInfo(shardingTotalCount);
  //获取分片策略类
  JobShardingStrategy jobShardingStrategy = JobShardingStrategyFactory.getStrategy(liteJobConfig.getJobShardingStrategyClass());
  ///分片
  jobNodeStorage.executeInTransaction(new PersistShardingInfoTransactionExecutionCallback(jobShardingStrategy.sharding(availableJobInstances, jobName, shardingTotalCount)));
  log.debug("Job '{}' sharding complete.", jobName);
}

/**
    重新设子分片信息 
**/
private void resetShardingInfo(final int shardingTotalCount) {
  for (int i = 0; i < shardingTotalCount; i++) {
    /** 删除jobInstance节点
        /{jobName}/sharing/{instanceIndex}分片项节点删除
    **/
    jobNodeStorage.removeJobNodeIfExisted(ShardingNode.getInstanceNode(i));
    /** 删除jobInstance节点
        /{jobName}/sharing/{instanceIndex}重新设置分片项
    **/
    jobNodeStorage.createJobNodeIfNeeded(ShardingNode.ROOT + "/" + i);
  }
  int actualShardingTotalCount = jobNodeStorage.getJobNodeChildrenKeys(ShardingNode.ROOT).size();
  if (actualShardingTotalCount > shardingTotalCount) {
    for (int i = shardingTotalCount; i < actualShardingTotalCount; i++) {
      //有多余分片删除
      jobNodeStorage.removeJobNodeIfExisted(ShardingNode.ROOT + "/" + i);
    }
  }
}

/** 分片 **/
@RequiredArgsConstructor
class PersistShardingInfoTransactionExecutionCallback implements TransactionExecutionCallback {

  private final Map<JobInstance, List<Integer>> shardingResults;

  @Override
  public void execute(final CuratorTransactionFinal curatorTransactionFinal) throws Exception {
    for (Map.Entry<JobInstance, List<Integer>> entry : shardingResults.entrySet()) {
      for (int shardingItem : entry.getValue()) {
       /**
        每个分片项创建一个实例
        {jobName}/sharing/{instanceIndex}/
       **/
 curatorTransactionFinal.create().forPath(jobNodePath.getFullPath(ShardingNode.getInstanceNode(shardingItem)), entry.getKey().getJobInstanceId().getBytes()).and();
      }
    }
    /**
        删除节点
        /{jobName}/sharding/necessary
        /{jobName}/sharding/processing
    **/    curatorTransactionFinal.delete().forPath(jobNodePath.getFullPath(ShardingNode.NECESSARY)).and(); curatorTransactionFinal.delete().forPath(jobNodePath.getFullPath(ShardingNode.PROCESSING)).and();
  }
}

在获取分片上下文后，根据每个分片项判断有无作业是运行中的状态，如果有，则标记为misfire

jobFacade.misfireIfRunning(shardingContexts.getShardingItemParameters().keySet())
  
public boolean misfireIfRunning(final Collection<Integer> shardingItems) {
  return executionService.misfireIfHasRunningItems(shardingItems);
}
    
/**
     * 如果当前分片项仍在运行则设置任务被错过执行的标记.
     * 
     * @param items 需要设置错过执行的任务分片项
     * @return 是否错过本次执行
     */
public boolean misfireIfHasRunningItems(final Collection<Integer> items) {
  if (!hasRunningItems(items)) {
    return false;
  }
  setMisfire(items);
  return true;
}

  /**
     * 设置任务被错过执行的标记.
     *
     * @param items 需要设置错过执行的任务分片项
     */
public void setMisfire(final Collection<Integer> items) {
  for (int each : items) {
    /**
    /{jobName}/{itemNum}/misfire
    **/
    jobNodeStorage.createJobNodeIfNeeded(ShardingNode.getMisfireNode(each));
  }
}

misfire判断结束之后，回去执行job，执行开始时，会将作业状态改为running状态，作业执行完成，将running节点删除。


private void execute(final ShardingContexts shardingContexts, final JobExecutionEvent.ExecutionSource executionSource) {
  if (shardingContexts.getShardingItemParameters().isEmpty()) {
    if (shardingContexts.isAllowSendJobEvent()) {
      jobFacade.postJobStatusTraceEvent(shardingContexts.getTaskId(), State.TASK_FINISHED, String.format("Sharding item for job '%s' is empty.", jobName));
    }
    return;
  }
  /**这里修改作业状态
        {jobName}/{itemNum}/running
        **/
  jobFacade.registerJobBegin(shardingContexts);
  String taskId = shardingContexts.getTaskId();
  if (shardingContexts.isAllowSendJobEvent()) {
    jobFacade.postJobStatusTraceEvent(taskId, State.TASK_RUNNING, "");
  }
  try {
    // failOver逻辑在这里
    process(shardingContexts, executionSource);
  } finally {
    // TODO 考虑增加作业失败的状态，并且考虑如何处理作业失败的整体回路
    // 删除running节点
    //{jobName}/{itemNum}/running 
    jobFacade.registerJobCompleted(shardingContexts);
    if (itemErrorMessages.isEmpty()) {
      if (shardingContexts.isAllowSendJobEvent()) {
        jobFacade.postJobStatusTraceEvent(taskId, State.TASK_FINISHED, "");
      }
    } else {
      if (shardingContexts.isAllowSendJobEvent()) {
        jobFacade.postJobStatusTraceEvent(taskId, State.TASK_ERROR, itemErrorMessages.toString());
      }
    }
  }
}
/**
     * 注册作业启动信息.
     * 
     * @param shardingContexts 分片上下文
     */
public void registerJobBegin(final ShardingContexts shardingContexts) {
  JobRegistry.getInstance().setJobRunning(jobName, true);
  if (!configService.load(true).isMonitorExecution()) {
    return;
  }
  for (int each : shardingContexts.getShardingItemParameters().keySet()) {
    /**这里修改作业状态
        {jobName}/{itemNum}/running
        **/
    jobNodeStorage.fillEphemeralJobNode(ShardingNode.getRunningNode(each), "");
  }
}

/**
     * 注册作业完成信息.
     * 
     * @param shardingContexts 分片上下文
     */
public void registerJobCompleted(final ShardingContexts shardingContexts) {
  JobRegistry.getInstance().setJobRunning(jobName, false);
  if (!configService.load(true).isMonitorExecution()) {
    return;
  }
  for (int each : shardingContexts.getShardingItemParameters().keySet()) {
     /**在这里删除节点
        {jobName}/{itemNum}/running
        **/
 jobNodeStorage.removeJobNodeIfExisted(ShardingNode.getRunningNode(each));
  }
}
最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 206,602评论 6赞 481
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 88,442评论 2赞 382
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 152,878评论 0赞 344
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 55,306评论 1赞 279
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 64,330评论 5赞 373
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 49,071评论 1赞 285
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 38,382评论 3赞 400
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 37,006评论 0赞 259
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 43,512评论 1赞 300
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,965评论 2赞 325
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 38,094评论 1赞 333
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 33,732评论 4赞 323
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 39,283评论 3赞 307
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 30,286评论 0赞 19
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,512评论 1赞 262
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 45,536评论 2赞 354
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 42,828评论 2赞 345