在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.
网友评论