7.实现任务的rebalance

1.设计

1.1 背景

系统启动后，所有任务都在被执行，如果这时某个节点宕机，那它负责的任务就不能执行了，这对有稳定性要求的任务是不能接受的，所以系统要实现rebalance的功能。

1.2 设计

下面是Job分配与执行的业务点，重分配就是在 follower下线、controller下线、节点新上线进行重分配。理清楚接下来实现就是水到渠成了

2. 实现

2.1 RebalanceJobType

定义了重平衡job的类型

public enum RebalanceJobType {FOLLOWER_OFFLINE(0), CONTROLLER_OFFLINE(1), NODE_ONLINE(2);private int code;RebalanceJobType(int code) {this.code = code;}public boolean isFollowerOffline() {return this.code == FOLLOWER_OFFLINE.code;}public boolean isControllerOffline() {return this.code == CONTROLLER_OFFLINE.code;}public boolean isNodeOnline() {return this.code == NODE_ONLINE.code;}}

2.2 AverageJobAllotStrategy

添加了 rebalanceJob的方法，只有Controller才能调用，对不同的重平衡情况进行分别处理

private Map<Long, List<DttaskJob>> getDttaskJobMap() {List<DttaskJob> allDttaskJob = getAllDttaskJob();return average(allDttaskJob);
}@Override
public void rebalanceJob(RebalanceJobContext rebalanceJobContext) {if (rebalanceJobContext.getType().isFollowerOffline()|| rebalanceJobContext.getType().isControllerOffline()) {long offlineServerId = rebalanceJobContext.getServerId();log.info("{}节点={}下线->重平衡job={}",rebalanceJobContext.getType().isFollowerOffline() ? "follower" : "controller",offlineServerId,rebalanceJobContext);List<DttaskJob> dttaskJobs = getByDttaskId(offlineServerId);List<NodeInfo> nodeInfoList = ServerInfo.getNodeInfoList();Map<Long, List<DttaskJob>> allotMap = new HashMap<>();int i = 0;int nodeCount = nodeInfoList.size();while (i < dttaskJobs.size()) {DttaskJob dttaskJob = dttaskJobs.get(i);NodeInfo nodeInfo = nodeInfoList.get(i % nodeCount);i++;List<DttaskJob> dttaskJobList = allotMap.getOrDefault(nodeInfo.getServerId(), new ArrayList<>());dttaskJobList.add(dttaskJob);allotMap.put(nodeInfo.getServerId(), dttaskJobList);}executeDttaskJob(new ExecuteDttaskJobContext(allotMap, true));} else if (rebalanceJobContext.getType().isNodeOnline()) {log.info("节点上线->重平衡job={}", rebalanceJobContext);long onlineServerId = rebalanceJobContext.getServerId();Map<Long, List<DttaskJob>> dttaskJobMap = BeanUseHelper.entityHelpService().queryDttaskJob();Map<Long, List<DttaskJob>> allotDttaskJobMap = getDttaskJobMap();Map<Long, List<DttaskJob>> stopDttaskJobMapOfOldNodes = new HashMap<>();Map<Long, List<DttaskJob>> startDttaskJobMapOfNewNodes = new HashMap<>();List<DttaskJob> startDttaskJobs = new ArrayList<>();dttaskJobMap.forEach((serverId, dttaskJobList) -> {int size = dttaskJobList.size();int newSize = allotDttaskJobMap.get(serverId).size();if (size > newSize) {List<DttaskJob> dttaskJobs = dttaskJobList.subList(0, size - newSize);stopDttaskJobMapOfOldNodes.put(serverId, dttaskJobs);startDttaskJobs.addAll(dttaskJobs);}});startDttaskJobMapOfNewNodes.put(onlineServerId, startDttaskJobs);executeDttaskJob(new ExecuteDttaskJobContext(stopDttaskJobMapOfOldNodes, false));executeDttaskJob(new ExecuteDttaskJobContext(startDttaskJobMapOfNewNodes, true));}
}

2.3 ServerClientChannelHandler

对节点下线进行重平衡处理