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

在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循环），直到主节点选举出来。

@RequiredArgsConstructorclass 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时，超过最长时间，临时节点失效。

作者：一滴水的坚持
链接：https://www.jianshu.com/p/49eefc7df554