失效转移： 运行中的作业服务器崩溃不会导致重新分片，只会在下次作业启动时分片。启用失效转移功能可以在本次作业执行过程中，监测其他作业服务器空闲，抓取未完成的孤儿分片项执行。
实现失效转移功能，在某台服务器执行完毕后主动抓取未分配的分片，并且在某台服务器下线后主动寻找可用的服务器执行任务。
当系统crashed的时候，和zk连接的临时节点会被删除，从而触发对临时节点的监控.(实测下来，只要机器关闭，都会执行一次failover，不知道是不是bug)具体是这样实现的，首先会判断是否是自己机器接收到该事件，如果是，则不处理。然后获取失效节点（/{jobName}/sharding/{index}/failover），若失效节点不为null，则将该分片状态置为failover状态（创建{jobName}/leader/failover/items/{index}节点），然后判断是否需要failOver，需要则执行failOver，不需要则结束。如果失效节点为null，则获取所有该jobInstance的分片项，创建需要failover的分片项{jobName}/leader/failover/items/{index}，判断是否需要failover，需要则执行failover，不需要则结束。
class JobCrashedJobListener extends AbstractJobListener {
@Override
protected void dataChanged(final String path, final Type eventType, final String data) {
if (isFailoverEnabled() && Type.NODE_REMOVED == eventType && instanceNode.isInstancePath(path)) {
String jobInstanceId = path.substring(instanceNode.getInstanceFullPath().length() + 1);
if (jobInstanceId.equals(JobRegistry.getInstance().getJobInstance(jobName).getJobInstanceId())) {
return;
}
List<Integer> failoverItems = failoverService.getFailoverItems(jobInstanceId);
if (!failoverItems.isEmpty()) {
for (int each : failoverItems) {
// /{jobName}/sharding/{index}/failOver
failoverService.setCrashedFailoverFlag(each);
//如果需要失效转移则失效转移
failoverService.failoverIfNecessary();
}
} else {
for (int each : shardingService.getShardingItems(jobInstanceId)) {
//将所有该作业的实例都置为failOver。
failoverService.setCrashedFailoverFlag(each);
//如果需要失效转移则失效转移
failoverService.failoverIfNecessary();
}
}
}
}
}
public void setCrashedFailoverFlag(final int item) {
if (!isFailoverAssigned(item)) {
//创建节点 /{jobName}/leader/failover/items/{index}
jobNodeStorage.createJobNodeIfNeeded(FailoverNode.getItemsNode(item));
}
}
而在失效转移过程中，需要判断是否需要失效转移，判断是否/{jobName}/leader/failover/items/{index}该节点存在，且该job不是处于run状态，然后选举一个主节点/{jobName}/leader/failover/latch,然后由主节点回调。
public void failoverIfNecessary() {
if (needFailover()) {
//选举主节点
jobNodeStorage.executeInLeader(FailoverNode.LATCH, new FailoverLeaderExecutionCallback());
}
}
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);
}
}
而在回调过程中，则将分片参数绑定/{jobName}/sharding/{index}/failover绑定jobInstanceId。然后将{jobName}/leader/failover/items/{index}删除。每次触发一个分片，若有多个分片，则for循环处理了，这里是根据每个分片做处理，所以这里只需要拿到一个分片处理即可，再去执行triggerJob，执行作业。
class FailoverLeaderExecutionCallback implements LeaderExecutionCallback {
@Override
public void execute() {
if (JobRegistry.getInstance().isShutdown(jobName) || !needFailover()) {
return;
}
int crashedItem = Integer.parseInt(jobNodeStorage.getJobNodeChildrenKeys(FailoverNode.ITEMS_ROOT).get(0));
log.debug("Failover job '{}' begin, crashed item '{}'", jobName, crashedItem);
jobNodeStorage.fillEphemeralJobNode(FailoverNode.getExecutionFailoverNode(crashedItem), JobRegistry.getInstance().getJobInstance(jobName).getJobInstanceId());
jobNodeStorage.removeJobNodeIfExisted(FailoverNode.getItemsNode(crashedItem));
// TODO 不应使用triggerJob, 而是使用executor统一调度
JobScheduleController jobScheduleController = JobRegistry.getInstance().getJobScheduleController(jobName);
if (null != jobScheduleController) {
jobScheduleController.triggerJob();
}
}
}
如果所有节点都处于run状态，则在作业的执行完成，会去主动调用jobFacade.failoverIfNecessary();这样就能保证上面所说的失效转移功能，在某台服务器执行完毕后主动抓取未分配的分片，并且在某台服务器下线后主动寻找可用的服务器执行任务。
当非作业运行时，系统crached或者说修改了分片参数， 会设置重新分片。
class ShardingTotalCountChangedJobListener extends AbstractJobListener {
@Override
protected void dataChanged(final String path, final Type eventType, final String data) {
if (configNode.isConfigPath(path) && 0 != JobRegistry.getInstance().getCurrentShardingTotalCount(jobName)) {
int newShardingTotalCount = LiteJobConfigurationGsonFactory.fromJson(data).getTypeConfig().getCoreConfig().getShardingTotalCount();
//当分片参数修改时，会重新分片
if (newShardingTotalCount != JobRegistry.getInstance().getCurrentShardingTotalCount(jobName)) {
//重新分片
shardingService.setReshardingFlag();
JobRegistry.getInstance().setCurrentShardingTotalCount(jobName, newShardingTotalCount);
}
}
}
}
class ListenServersChangedJobListener extends AbstractJobListener {
@Override
protected void dataChanged(final String path, final Type eventType, final String data) {、
//当作业未关闭，且作业实例变化 或者是服务器状态发生变化
if (!JobRegistry.getInstance().isShutdown(jobName) && (isInstanceChange(eventType, path) || isServerChange(path))) {
shardingService.setReshardingFlag();
}
}
//如果系统crashed时，作业实例节点为NODE.REMOVED事件， 也就是会重新分片。
private boolean isInstanceChange(final Type eventType, final String path) {
return instanceNode.isInstancePath(path) && Type.NODE_UPDATED != eventType;
}
private boolean isServerChange(final String path) {
return serverNode.isServerPath(path);
}
}
