美文网首页0.面试技能
ElasticJob 源码解析之主节点选举分片实现

ElasticJob 源码解析之主节点选举分片实现

作者: 一滴水的坚持 | 来源:发表于2017-12-02 14:22 被阅读0次

    在elasticJob中,最重要的一个功能就是作业分片,作业分片是怎样实现的,由谁来负责分片?哈哈,肯定不是我来负责分片的,肯定是集群中的某台机器啦,一个集群由很多台机器,那到底是哪台机器来负责?万一这台机器挂掉了,那怎么办?

    ​ 原来在elasticJob中,每次有新机器上线,都会去触发分片,但并不是所有机器都去做分片,而是有一台主节点机器去负责分片,这个主节点是选举出来的。

    public void shardingIfNecessary() {
      List<JobInstance> availableJobInstances = instanceService.getAvailableJobInstances();
      //需要分片根据节点判断/{jobName}/sharding/necessary  启动过程中注册
      //是否有作业实例根据 /{jobName}/instances/ 
      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);
        // /{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);
    }
    

    所以主节点选举是在blockUntilShardingCompleted();中完成的,完成主节点选举,没选举完成就一直等待。

    private void blockUntilShardingCompleted() {
      while (!leaderService.isLeaderUntilBlock() && (jobNodeStorage.isJobNodeExisted(ShardingNode.NECESSARY) || jobNodeStorage.isJobNodeExisted(ShardingNode.PROCESSING))) {
        log.debug("Job '{}' sleep short time until sharding completed.", jobName);
        //thread.sleep 100ms 等待选举完成
        BlockUtils.waitingShortTime();
      }
    }
    
    public boolean isLeaderUntilBlock() {
      while (!hasLeader() && serverService.hasAvailableServers()) {
        log.info("Leader is electing, waiting for {} ms", 100);
        BlockUtils.waitingShortTime();
        if (!JobRegistry.getInstance().isShutdown(jobName) && serverService.isAvailableServer(JobRegistry.getInstance().getJobInstance(jobName).getIp())) {
          //选举主节点
          electLeader();
        }
      }
      return isLeader();
    }
    
    public void electLeader() {
      log.debug("Elect a new leader now.");
      //主节点选举
      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);
     }
    }
    

    这个LeaderLatch到底是什么?怎么实现选举?

    在curator中,提供了两种选举方案:Leader Latch和Leader Election;

    • Leader Latch
      • 随机从候选着中选出一台作为leader,选中之后除非调用close()释放leadship,否则其他的后选择无法成为leader。
    • Leader Election
      • 通过LeaderSelectorListener可以对领导权进行控制, 在适当的时候释放领导权,这样每个节点都有可能获得领导权。 而LeaderLatch则一直持有leadership, 除非调用close方法,否则它不会释放领导权。

    在elasticJob中,很明显用了第一种Leader Latch,调用 #await() 等待拿到这把。如果有多个线程执行了相同节点路径的 LeaderLatch 的 #await() 后,同一时刻有且仅有一个线程可以继续执行,其他线程需要等待。当该线程释放( LeaderLatch#close() )后,下一个线程可以拿到该继续执行。在这里,没有拿到锁的就一直等待,直到超时,中断线程。拿到锁的继续调用LeaderExecutionCallback(),将jobInstanceId注册到注册中心。在使用Leader Latch过程中,如果出现SUSPENDED或者LOST,leader会报告自己不再是leader(直到重新建立连接,否则不会有leader)。这个时候,blockUntilShardingCompleted()方法会会重新选举主节点(while循环),直到主节点选举出来。

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

    当要主节点作业停止执行,怎么将主节点删除,触发进而能够重新选举主节点?

    public final class JobShutdownHookPlugin extends ShutdownHookPlugin {
    
      private String jobName;
    
      @Override
      public void initialize(final String name, final Scheduler scheduler, final ClassLoadHelper classLoadHelper) throws SchedulerException {
        super.initialize(name, scheduler, classLoadHelper);
        jobName = scheduler.getSchedulerName();
      }
    
      @Override
      public void shutdown() {
        CoordinatorRegistryCenter regCenter = JobRegistry.getInstance().getRegCenter(jobName);
        if (null == regCenter) {
          return;
        }
        LeaderService leaderService = new LeaderService(regCenter, jobName);
        if (leaderService.isLeader()) {
          //删除主节点
          leaderService.removeLeader();
        }
        //删除作业实例
        new InstanceService(regCenter, jobName).removeInstance();
      }
    }
    
    

    当机器crashed时,超过最长时间,临时节点失效。

    具体的分片在作业配置存储有具体介绍job配置数据存储

    down.

    相关文章

      网友评论

        本文标题:ElasticJob 源码解析之主节点选举分片实现

        本文链接:https://www.haomeiwen.com/subject/okenbxtx.html