问题3 当某一实例 挂掉时候 ，集群的leader 会报异常 ，当任务完成的时候 ， 只会重新分片 ， 如果 任务没有完成 类似于 问题2 ， leader 报异常后 会报超时异常 。
完成超时异常 ，同问题2 。这里我们讲一讲 leader 什么时候 会重新选举
leader 相关的服务主要在 LeaderService 中 ， 我们可以看到 ，有3个地方调用了 electLeader 方法进行选举。
具体分析
SchedulerFacade 的registerStartUpInfo(boolean) 方法中调用了选举方法 leaderService.electLeader();
/**
* 注册作业启动信息.
*
* @param enabled 作业是否启用
*/
public void registerStartUpInfo(final boolean enabled) {
listenerManager.startAllListeners();
leaderService.electLeader();
serverService.persistOnline(enabled);
instanceService.persistOnline();
shardingService.setReshardingFlag();
monitorService.listen();
if (!reconcileService.isRunning()) {
reconcileService.startAsync();
}
}
实例启动的时候 会进行一次选举 , 会创建创建一临时节点 位置 leader/election/instance 。
LeaderService isLeaderUntilBlock()调用了 选举方法leaderService.electLeader();
/**
* 判断当前节点是否是主节点.
*
* <p>
* 如果主节点正在选举中而导致取不到主节点, 则阻塞至主节点选举完成再返回.
* </p>
*
* @return 当前节点是否是主节点
*/
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();
}
当有leader 的时候 ， 集群没有合适实例 isLeaderUntilBlock 中的选举方法 是不会执行的 。
在满足以下条件下, 会执行electLeader
1 没有 leader !hasLeader()
2 有合适的服务器 serverService.hasAvailableServers()
3 当前job 没有 shutdown !JobRegistry.getInstance().isShutdown(jobName)
4 当前服务器是合适的 serverService.isAvailableServer
isLeaderUntilBlock 会在 job 执行 ，获取分片上下文的时候 执行 ,如果当前集群 ，分片未结束，blockUntilShardingCompleted方法 会阻塞 ，等待分片结束
/**
* 如果需要分片且当前节点为主节点, 则作业分片.
*
* <p>
* 如果当前无可用节点则不分片.
* </p>
*/
public void shardingIfNecessary() {
List<JobInstance> availableJobInstances = instanceService.getAvailableJobInstances();
if (!isNeedSharding() || availableJobInstances.isEmpty()) {
return;
}
if (!leaderService.isLeaderUntilBlock()) {
blockUntilShardingCompleted();
return;
}
ElectionListenerManager.LeaderElectionJobListener 中的 dataChanged方法中调用了 选举方法
class LeaderElectionJobListener extends AbstractJobListener {
@Override
protected void dataChanged(final String path, final Type eventType, final String data) {
if (!JobRegistry.getInstance().isShutdown(jobName) && (isActiveElection(path, data) || isPassiveElection(path, eventType))) {
leaderService.electLeader();
}
}
该listener 会监听 zk ， jobname 级别tree node 变化 ， 如果当前job 没有shutdown ， 满足主动选举 或者被动选举 任意条件 会 进行选举 。
主动选举代码如下 ， 当前集群没有leader ， 当前server没有disable 情况下
会返回true
比如 当前job 没有shutdown ，没有leader ， 本地路径 /javaSimpleJob/servers/192.168.200.151 与path 一致 ， 节点 data 不等于disable 可以参加选举 ，和事件无关哈
private boolean isActiveElection(final String path, final String data) {
return !leaderService.hasLeader() && isLocalServerEnabled(path, data);
}
被动选举代码如下 如果leader crash ， 如果当前实例可用 ，isPassiveElection 会返回 true 可以参加选举
private boolean isPassiveElection(final String path, final Type eventType) {
return isLeaderCrashed(path, eventType) && serverService.isAvailableServer(JobRegistry.getInstance().getJobInstance(jobName).getIp());
}
isLeaderCrashed 判断当前event path 是否是 /leader/election/instance下的节点 ， eventType 是否NODE_REMOVED 如果都满足 ，return true
...
private boolean isLeaderCrashed(final String path, final Type eventType) {
return leaderNode.isLeaderInstancePath(path) && Type.NODE_REMOVED == eventType;
}
...
