负载均衡、轮询加权
LoadBalanceUtil
@Component
public class LoadBalanceUtil {private List<Node> nodes;/*** 初始化 约定的invoker和权重的键值对*/public void init(Map<Invoker, Integer> invokersWeight){if (invokersWeight != null && !invokersWeight.isEmpty()) {nodes = new ArrayList<>(invokersWeight.size());invokersWeight.forEach((invoker, weight) -> nodes.add(new Node(invoker, weight)));} else {nodes = null;}}/*** 算法逻辑：* 1. 对于每个请求，遍历集群中的所有可用后端，对于每个后端peer执行：* peer->current_weight += peer->effecitve_weight。* 同时累加所有peer的effective_weight，保存为total。* 2. 从集群中选出current_weight最大的peer，作为本次选定的后端。* 3. 对于本次选定的后端，执行：peer->current_weight -= total。** @Return ivoker*/public Invoker select() {if (!checkNodes()) {return null;} else if (nodes.size() == 1) {if (nodes.get(0).invoker.isAvalable()) {return nodes.get(0).invoker;} else {return null;}}Integer total = 0;Node nodeOfMaxWeight = null;for (Node node : nodes) {total += node.effectiveWeight;node.currentWeight += node.effectiveWeight;if (nodeOfMaxWeight == null) {nodeOfMaxWeight = node;} else {nodeOfMaxWeight = nodeOfMaxWeight.compareTo(node) > 0 ? nodeOfMaxWeight : node;}}nodeOfMaxWeight.currentWeight -= total;return nodeOfMaxWeight.invoker;}private boolean checkNodes() {return (nodes != null && nodes.size() > 0);}public interface Invoker {Boolean isAvalable();String nodeName();}private static class Node implements Comparable<Node> {final Invoker invoker;final Integer weight;Integer effectiveWeight;Integer currentWeight;Node(Invoker invoker, Integer weight) {this.invoker = invoker;this.weight = weight;this.effectiveWeight = weight;this.currentWeight = 0;}@Overridepublic int compareTo(Node o) {return currentWeight > o.currentWeight ? 1 : (currentWeight.equals(o.currentWeight) ? 0 : -1);}}}
invoker
if (totalCount > 0) {//根据计算结果修改权重for (LoadBalanceUtil.Invoker key : invokersWeight.keySet()) {if (weightMap.containsKey(key.nodeName())) {invokersWeight.put(key, weightMap.get(key.nodeName()));}}//初始化权重loadBalanceUtil.init(invokersWeight);//根据AI空闲数量，获取待分配任务List<TaskInfo> waitAssignList = scheduleService.listWaitAllocate(totalCount);if (ObjectUtil.isEmpty(waitAssignList)) {//没数据，停30秒。Thread.sleep(cronInterval * 1000);return ReturnT.SUCCESS;}for (TaskInfo item : waitAssignList) {LoadBalanceUtil.Invoker invoker = loadBalanceUtil.select();item.setAiServerName(invoker.nodeName());item.setStatus(ReportStatus.ALLOCATED.getStatus());item.setAllocationTime(new Date());scheduleService.allocationServer(item); String logMsg = StrUtil.format("{} 分配给 {}", item.getCustodyNo(), invoker.nodeName());logger.info(logMsg);}
}