深入淺出負(fù)載均衡

public interface LoadBalance<N extends Node> {
    N select(List<N> nodes, String ip);
}

public abstract class BaseLoadBalance<N extends Node> implements LoadBalance<N> {
    @Override    public N select(List<N> nodes, String ip) {        if (CollectionUtil.isEmpty(nodes)) {            return null;        }
        // 如果 nodes 列表中僅有一個 node，直接返回即可，無需進(jìn)行負(fù)載均衡        if (nodes.size() == 1) {            return nodes.get(0);        }
        return doSelect(nodes, ip);    }
    protected abstract N doSelect(List<N> nodes, String ip);
}

public class Node implements Comparable<Node> {
    protected String url;
    protected Integer weight;
    protected Integer active;
    // ...}

public class RandomLoadBalance<N extends Node> extends BaseLoadBalance<N> implements LoadBalance<N> {
    private final Random random = new Random();
    @Override    protected N doSelect(List<N> nodes, String ip) {        // 在列表中隨機(jī)選取一個節(jié)點(diǎn)        int index = random.nextInt(nodes.size());        return nodes.get(index);    }
}

public class WeightRandomLoadBalance<N extends Node> extends BaseLoadBalance<N> implements LoadBalance<N> {
    private final Random random = ThreadLocalRandom.current();
    @Override    protected N doSelect(List<N> nodes, String ip) {
        int length = nodes.size();        AtomicInteger totalWeight = new AtomicInteger(0);        for (N node : nodes) {            Integer weight = node.getWeight();            totalWeight.getAndAdd(weight);        }
        if (totalWeight.get() > 0) {            int offset = random.nextInt(totalWeight.get());            for (N node : nodes) {                // 讓隨機(jī)值 offset 減去權(quán)重值                offset -= node.getWeight();                if (offset < 0) {                    // 返回相應(yīng)的 Node                    return node;                }            }        }
        // 直接隨機(jī)返回一個        return nodes.get(random.nextInt(length));    }
}

public class RoundRobinLoadBalance<N extends Node> extends BaseLoadBalance<N> implements LoadBalance<N> {
    private final AtomicInteger position = new AtomicInteger(0);
    @Override    protected N doSelect(List<N> nodes, String ip) {        int length = nodes.size();        // 如果位置值已經(jīng)等于節(jié)點(diǎn)數(shù)，重置為 0        position.compareAndSet(length, 0);        N node = nodes.get(position.get());        position.getAndIncrement();        return node;    }
}

public class WeightRoundRobinLoadBalance<N extends Node> extends BaseLoadBalance<N> implements LoadBalance<N> {
    /**     * 60秒     */    private static final int RECYCLE_PERIOD = 60000;
    /**     * Node hashcode 到 WeightedRoundRobin 的映射關(guān)系     */    private ConcurrentMap<Integer, WeightedRoundRobin> weightMap = new ConcurrentHashMap<>();
    /**     * 原子更新鎖     */    private AtomicBoolean updateLock = new AtomicBoolean();
    @Override    protected N doSelect(List<N> nodes, String ip) {
        int totalWeight = 0;        long maxCurrent = Long.MIN_VALUE;
        // 獲取當(dāng)前時間        long now = System.currentTimeMillis();        N selectedNode = null;        WeightedRoundRobin selectedWRR = null;
        // 下面這個循環(huán)主要做了這樣幾件事情：        //   1. 遍歷 Node 列表，檢測當(dāng)前 Node 是否有相應(yīng)的 WeightedRoundRobin，沒有則創(chuàng)建        //   2. 檢測 Node 權(quán)重是否發(fā)生了變化，若變化了，則更新 WeightedRoundRobin 的 weight 字段        //   3. 讓 current 字段加上自身權(quán)重，等價于 current += weight        //   4. 設(shè)置 lastUpdate 字段，即 lastUpdate = now        //   5. 尋找具有最大 current 的 Node，以及 Node 對應(yīng)的 WeightedRoundRobin，        //      暫存起來，留作后用        //   6. 計算權(quán)重總和        for (N node : nodes) {            int hashCode = node.hashCode();            WeightedRoundRobin weightedRoundRobin = weightMap.get(hashCode);            int weight = node.getWeight();            if (weight < 0) {                weight = 0;            }
            // 檢測當(dāng)前 Node 是否有對應(yīng)的 WeightedRoundRobin，沒有則創(chuàng)建            if (weightedRoundRobin == null) {                weightedRoundRobin = new WeightedRoundRobin();                // 設(shè)置 Node 權(quán)重                weightedRoundRobin.setWeight(weight);                // 存儲 url 唯一標(biāo)識 identifyString 到 weightedRoundRobin 的映射關(guān)系                weightMap.putIfAbsent(hashCode, weightedRoundRobin);                weightedRoundRobin = weightMap.get(hashCode);            }            // Node 權(quán)重不等于 WeightedRoundRobin 中保存的權(quán)重，說明權(quán)重變化了，此時進(jìn)行更新            if (weight != weightedRoundRobin.getWeight()) {                weightedRoundRobin.setWeight(weight);            }
            // 讓 current 加上自身權(quán)重，等價于 current += weight            long current = weightedRoundRobin.increaseCurrent();            // 設(shè)置 lastUpdate，表示近期更新過            weightedRoundRobin.setLastUpdate(now);            // 找出最大的 current            if (current > maxCurrent) {                maxCurrent = current;                // 將具有最大 current 權(quán)重的 Node 賦值給 selectedNode                selectedNode = node;                // 將 Node 對應(yīng)的 weightedRoundRobin 賦值給 selectedWRR，留作后用                selectedWRR = weightedRoundRobin;            }
            // 計算權(quán)重總和            totalWeight += weight;        }
        // 對 weightMap 進(jìn)行檢查，過濾掉長時間未被更新的節(jié)點(diǎn)。        // 該節(jié)點(diǎn)可能掛了，nodes 中不包含該節(jié)點(diǎn)，所以該節(jié)點(diǎn)的 lastUpdate 長時間無法被更新。        // 若未更新時長超過閾值后，就會被移除掉，默認(rèn)閾值為60秒。        if (!updateLock.get() && nodes.size() != weightMap.size()) {            if (updateLock.compareAndSet(false, true)) {                try {                    // 遍歷修改，即移除過期記錄                    weightMap.entrySet().removeIf(item -> now - item.getValue().getLastUpdate() > RECYCLE_PERIOD);                } finally {                    updateLock.set(false);                }            }        }
        if (selectedNode != null) {            // 讓 current 減去權(quán)重總和，等價于 current -= totalWeight            selectedWRR.decreaseCurrent(totalWeight);            // 返回具有最大 current 的 Node            return selectedNode;        }
        // should not happen here        return nodes.get(0);    }
    protected static class WeightedRoundRobin {
        // 服務(wù)提供者權(quán)重        private int weight;        // 當(dāng)前權(quán)重        private AtomicLong current = new AtomicLong(0);        // 最后一次更新時間        private long lastUpdate;
        public long increaseCurrent() {            // current = current + weight；            return current.addAndGet(weight);        }
        public long decreaseCurrent(int total) {            // current = current - total;            return current.addAndGet(-1 * total);        }
        public int getWeight() {            return weight;        }
        public void setWeight(int weight) {            this.weight = weight;            // 初始情況下，current = 0            current.set(0);        }
        public AtomicLong getCurrent() {            return current;        }
        public void setCurrent(AtomicLong current) {            this.current = current;        }
        public long getLastUpdate() {            return lastUpdate;        }
        public void setLastUpdate(long lastUpdate) {            this.lastUpdate = lastUpdate;        }
    }
}

public class LeastActiveLoadBalance<N extends Node> extends BaseLoadBalance<N> implements LoadBalance<N> {
    private final Random random = new Random();
    @Override    protected N doSelect(List<N> nodes, String ip) {        int length = nodes.size();        // 最小的活躍數(shù)        int leastActive = -1;        // 具有相同“最小活躍數(shù)”的服務(wù)者提供者（以下用 Node 代稱）數(shù)量        int leastCount = 0;        // leastIndexs 用于記錄具有相同“最小活躍數(shù)”的 Node 在 nodes 列表中的下標(biāo)信息        int[] leastIndexs = new int[length];        int totalWeight = 0;        // 第一個最小活躍數(shù)的 Node 權(quán)重值，用于與其他具有相同最小活躍數(shù)的 Node 的權(quán)重進(jìn)行對比，        // 以檢測是否“所有具有相同最小活躍數(shù)的 Node 的權(quán)重”均相等        int firstWeight = 0;        boolean sameWeight = true;
        // 遍歷 nodes 列表        for (int i = 0; i < length; i++) {            N node = nodes.get(i);            // 發(fā)現(xiàn)更小的活躍數(shù)，重新開始            if (leastActive == -1 || node.getActive() < leastActive) {                // 使用當(dāng)前活躍數(shù)更新最小活躍數(shù) leastActive                leastActive = node.getActive();                // 更新 leastCount 為 1                leastCount = 1;                // 記錄當(dāng)前下標(biāo)值到 leastIndexs 中                leastIndexs[0] = i;                totalWeight = node.getWeight();                firstWeight = node.getWeight();                sameWeight = true;
                // 當(dāng)前 Node 的活躍數(shù) node.getActive() 與最小活躍數(shù) leastActive 相同            } else if (node.getActive() == leastActive) {                // 在 leastIndexs 中記錄下當(dāng)前 Node 在 nodes 集合中的下標(biāo)                leastIndexs[leastCount++] = i;                // 累加權(quán)重                totalWeight += node.getWeight();                // 檢測當(dāng)前 Node 的權(quán)重與 firstWeight 是否相等，                // 不相等則將 sameWeight 置為 false                if (sameWeight && i > 0                    && node.getWeight() != firstWeight) {                    sameWeight = false;                }            }        }
        // 當(dāng)只有一個 Node 具有最小活躍數(shù)，此時直接返回該 Node 即可        if (leastCount == 1) {            return nodes.get(leastIndexs[0]);        }
        // 有多個 Node 具有相同的最小活躍數(shù)，但它們之間的權(quán)重不同        if (!sameWeight && totalWeight > 0) {            // 隨機(jī)生成一個 [0, totalWeight) 之間的數(shù)字            int offsetWeight = random.nextInt(totalWeight);            // 循環(huán)讓隨機(jī)數(shù)減去具有最小活躍數(shù)的 Node 的權(quán)重值，            // 當(dāng) offset 小于等于0時，返回相應(yīng)的 Node            for (int i = 0; i < leastCount; i++) {                int leastIndex = leastIndexs[i];                // 獲取權(quán)重值，并讓隨機(jī)數(shù)減去權(quán)重值                offsetWeight -= nodes.get(leastIndex).getWeight();                if (offsetWeight <= 0) {                    return nodes.get(leastIndex);                }            }        }        // 如果權(quán)重相同或權(quán)重為0時，隨機(jī)返回一個 Node        return nodes.get(leastIndexs[random.nextInt(leastCount)]);    }
}

public class IpHashLoadBalance<N extends Node> extends BaseLoadBalance<N> implements LoadBalance<N> {
    @Override    protected N doSelect(List<N> nodes, String ip) {        if (StrUtil.isBlank(ip)) {            ip = "127.0.0.1";        }
        int length = nodes.size();        int index = hash(ip) % length;        return nodes.get(index);    }
    public int hash(String text) {        return HashUtil.fnvHash(text);    }
}

public class ConsistentHashLoadBalance<N extends Node> extends BaseLoadBalance<N> implements LoadBalance<N> {
    private final ConcurrentMap<String, ConsistentHashSelector<?>> selectors = new ConcurrentHashMap<>();
    @SuppressWarnings("unchecked")    @Override    protected N doSelect(List<N> nodes, String ip) {        // 分片數(shù)，這里設(shè)為節(jié)點(diǎn)數(shù)的 4 倍        Integer replicaNum = nodes.size() * 4;        // 獲取 nodes 原始的 hashcode        int identityHashCode = System.identityHashCode(nodes);
        // 如果 nodes 是一個新的 List 對象，意味著節(jié)點(diǎn)數(shù)量發(fā)生了變化        // 此時 selector.identityHashCode != identityHashCode 條件成立        ConsistentHashSelector<N> selector = (ConsistentHashSelector<N>) selectors.get(ip);        if (selector == null || selector.identityHashCode != identityHashCode) {            // 創(chuàng)建新的 ConsistentHashSelector            selectors.put(ip, new ConsistentHashSelector<>(nodes, identityHashCode, replicaNum));            selector = (ConsistentHashSelector<N>) selectors.get(ip);        }        // 調(diào)用 ConsistentHashSelector 的 select 方法選擇 Node        return selector.select(ip);    }
    /**     * 一致性哈希選擇器     */    private static final class ConsistentHashSelector<N extends Node> {
        /**         * 存儲虛擬節(jié)點(diǎn)         */        private final TreeMap<Long, N> virtualNodes;
        private final int identityHashCode;
        /**         * 構(gòu)造器         *         * @param nodes            節(jié)點(diǎn)列表         * @param identityHashCode hashcode         * @param replicaNum       分片數(shù)         */        ConsistentHashSelector(List<N> nodes, int identityHashCode, Integer replicaNum) {            this.virtualNodes = new TreeMap<>();            this.identityHashCode = identityHashCode;            // 獲取虛擬節(jié)點(diǎn)數(shù)，默認(rèn)為 100            if (replicaNum == null) {                replicaNum = 100;            }            for (N node : nodes) {                for (int i = 0; i < replicaNum / 4; i++) {                    // 對 url 進(jìn)行 md5 運(yùn)算，得到一個長度為16的字節(jié)數(shù)組                    byte[] digest = md5(node.getUrl());                    // 對 digest 部分字節(jié)進(jìn)行 4 次 hash 運(yùn)算，得到四個不同的 long 型正整數(shù)                    for (int j = 0; j < 4; j++) {                        // h = 0 時，取 digest 中下標(biāo)為 0 ~ 3 的4個字節(jié)進(jìn)行位運(yùn)算                        // h = 1 時，取 digest 中下標(biāo)為 4 ~ 7 的4個字節(jié)進(jìn)行位運(yùn)算                        // h = 2, h = 3 時過程同上                        long m = hash(digest, j);                        // 將 hash 到 node 的映射關(guān)系存儲到 virtualNodes 中，                        // virtualNodes 需要提供高效的查詢操作，因此選用 TreeMap 作為存儲結(jié)構(gòu)                        virtualNodes.put(m, node);                    }                }            }        }
        public N select(String key) {            // 對參數(shù) key 進(jìn)行 md5 運(yùn)算            byte[] digest = md5(key);            // 取 digest 數(shù)組的前四個字節(jié)進(jìn)行 hash 運(yùn)算，再將 hash 值傳給 selectForKey 方法，            // 尋找合適的 Node            return selectForKey(hash(digest, 0));        }
        private N selectForKey(long hash) {            // 查找第一個大于或等于當(dāng)前 hash 的節(jié)點(diǎn)            Map.Entry<Long, N> entry = virtualNodes.ceilingEntry(hash);            // 如果 hash 大于 Node 在哈希環(huán)上最大的位置，此時 entry = null，            // 需要將 TreeMap 的頭節(jié)點(diǎn)賦值給 entry            if (entry == null) {                entry = virtualNodes.firstEntry();            }            // 返回 Node            return entry.getValue();        }
    }
    /**     * 計算 hash 值     */    public static long hash(byte[] digest, int number) {        return (((long) (digest[3 + number * 4] & 0xFF) << 24)            | ((long) (digest[2 + number * 4] & 0xFF) << 16)            | ((long) (digest[1 + number * 4] & 0xFF) << 8)            | (digest[number * 4] & 0xFF))            & 0xFFFFFFFFL;    }
    /**     * 計算 MD5 值     */    public static byte[] md5(String value) {        MessageDigest md5;        try {            md5 = MessageDigest.getInstance("MD5");        } catch (NoSuchAlgorithmException e) {            throw new IllegalStateException(e.getMessage(), e);        }        md5.reset();        byte[] bytes = value.getBytes(StandardCharsets.UTF_8);        md5.update(bytes);        return md5.digest();    }
}