dubbo源碼解析之負載均衡

在分佈式系統中,負載均衡是必不可少的一個模塊,dubbo 中提供了五種負載均衡的實現,在閱讀這塊源碼之前,建議先學習負載均衡的基礎知識。把看源碼當做一個印證自己心中所想的過程,這樣會得到事半功倍的效果

以下源碼分析基於 dubbo 2.77 版本

類結構

先來看一下這一塊的類結構圖

大部分算法都是在權重比的基礎上進行負載均衡,RandomLoadBalance 是默認的算法

類型 描述 是否默認 是否加權
RandomLoadBalance 隨機 是,默認權重相同
RoundRobinLoadBalance 輪訓 是,默認權重相同
LeastActiveLoadBalance 最少活躍數調用 不完全是,默認權重相同,僅在活躍數相同時按照權重比隨機
ConsistentHashLoadBalance 一致性hash
ShortestResponseLoadBalance 最短時間調用 不完全是,默認權重相同,僅在預估調用相同時按照權重比隨機

AbstractLoadBalance

AbstractLoadBalance 對一些通用的操作做了處理,是一個典型的模板方法模式的實現

select 方法只做一些簡單的範圍校驗,具體的實現有子類通過 doSelect 方法去實現

    @Override
    public <T> Invoker<T> select(List<Invoker<T>> invokers, URL url, Invocation invocation) {
        if (CollectionUtils.isEmpty(invokers)) {
            return null;
        }
        if (invokers.size() == 1) {
            return invokers.get(0);
        }
        return doSelect(invokers, url, invocation);
    }

getWeight方法封裝了獲取一個調用者的權重值的方法,並加入了預熱處理

    int getWeight(Invoker<?> invoker, Invocation invocation) {
        int weight;
        URL url = invoker.getUrl();
        // Multiple registry scenario, load balance among multiple registries.
        // 註冊中心不需要預熱
        if (REGISTRY_SERVICE_REFERENCE_PATH.equals(url.getServiceInterface())) {
            weight = url.getParameter(REGISTRY_KEY + "." + WEIGHT_KEY, DEFAULT_WEIGHT);
        } else {
            // 獲取配置的權重值
            weight = url.getMethodParameter(invocation.getMethodName(), WEIGHT_KEY, DEFAULT_WEIGHT);
            if (weight > 0) {
                // 獲取服務提供者啟動時的時間戳
                long timestamp = invoker.getUrl().getParameter(TIMESTAMP_KEY, 0L);
                if (timestamp > 0L) {
                    //  獲取啟動時長
                    long uptime = System.currentTimeMillis() - timestamp;
                    // 當前時間小於服務提供者啟動時間,直接給一個最小權重1
                    if (uptime < 0) {
                        return 1;
                    }
                    // 獲取預熱時間
                    int warmup = invoker.getUrl().getParameter(WARMUP_KEY, DEFAULT_WARMUP);
                    // 如果小於預熱時間,計算權重
                    if (uptime > 0 && uptime < warmup) {
                        weight = calculateWarmupWeight((int)uptime, warmup, weight);
                    }
                }
            }
        }
        // 取與零比較的最大值,保證不會出現負值權重
        return Math.max(weight, 0);
    }

calculateWarmupWeight 方法用來計算權重,保證隨着預熱時間的增加,權重逐漸達到設置的權重

    static int calculateWarmupWeight(int uptime, int warmup, int weight) {
        // 運行時間/(預熱時間/權重)
        int ww = (int) ( uptime / ((float) warmup / weight));
        // 保證計算的權重最小值是1,並且不能超過設置的權重
        return ww < 1 ? 1 : (Math.min(ww, weight));
    }

RandomLoadBalance

隨機調用是負載均衡算法中最常用的算法之一,也是 dubbo 的默認負載均衡算法,實現起來也較為簡單
隨機調用的缺點是在調用量比較少的情況下,有可能出現不均勻的情況

	@Override
    protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
        // Number of invokers
        int length = invokers.size();
        // Every invoker has the same weight?
        boolean sameWeight = true;
        // the weight of every invokers
        int[] weights = new int[length];
        // the first invoker's weight
        int firstWeight = getWeight(invokers.get(0), invocation);
        weights[0] = firstWeight;
        // The sum of weights
        int totalWeight = firstWeight;
        for (int i = 1; i < length; i++) {
            int weight = getWeight(invokers.get(i), invocation);
            // save for later use
            // 依次把權重放到數組對應的位置
            weights[i] = weight;
            // Sum
            // 累加權重
            totalWeight += weight;
            // 如果出現權重不一樣的,sameWeight 設為false
            if (sameWeight && weight != firstWeight) {
                sameWeight = false;
            }
        }
        if (totalWeight > 0 && !sameWeight) {
            // If (not every invoker has the same weight & at least one invoker's weight>0), select randomly based on totalWeight.
            // 在總權重裏面隨機選擇一個偏移量
            int offset = ThreadLocalRandom.current().nextInt(totalWeight);
            // Return a invoker based on the random value.
            for (int i = 0; i < length; i++) {
                offset -= weights[i];
                // 依次用偏移量減去當前權重,小於0說明選中
                if (offset < 0) {
                    return invokers.get(i);
                }
            }
        }
        // If all invokers have the same weight value or totalWeight=0, return evenly.
        // 如果所有的調用者有同樣的權重或者總權重為0,則隨機選擇一個
        return invokers.get(ThreadLocalRandom.current().nextInt(length));
    }

RoundRobinLoadBalance

輪訓算法避免了隨機算法在小數據量產生的不均勻問題,我個人認為,輪訓算法可以理解為隨機算法的一種特例,在大量請求的情況下,從調用次數看,和隨機並無區別,主要區別在於短時間內的調用分配上

加權輪訓算法給人的直觀感受,實現起來並不複雜,算出一權重總量,依次調用即可
例如A,B,C 三個節點的權重比依次 1,200,1000,如果依次輪訓調用,就會出現先調用A 10 次,再調用B 200次,最後調用 C 1000次,不斷重複前面的過程
但這樣有一個問題,我們可以發現C 被練習調用1000次,會對C瞬間造成很大的壓力

dubbo的新版本採用的是平滑加權輪詢算法,輪訓的過程中節點之間穿插調用,可以避免了上面說的問題,因此這塊源碼看起來會稍有難度

輪訓算法 在dubbo 在升級的過程中,做過多次優化,有興趣的可以去了解下該算法的優化過程,也是件很有意思的事情

public class RoundRobinLoadBalance extends AbstractLoadBalance {
    public static final String NAME = "roundrobin";

    private static final int RECYCLE_PERIOD = 60000;

    protected static class WeightedRoundRobin {
        // 權重值
        private int weight;
        // 當前權重值
        private AtomicLong current = new AtomicLong(0);
        // 最後一次使用該對象時間
        private long lastUpdate;

        public int getWeight() {
            return weight;
        }

        public void setWeight(int weight) {
            this.weight = weight;
            current.set(0);
        }

        // 獲取自增權重基數的當前權重值
        public long increaseCurrent() {
            return current.addAndGet(weight);
        }

        public void sel(int total) {
            current.addAndGet(-1 * total);
        }

        public long getLastUpdate() {
            return lastUpdate;
        }

        // 設置最後一次更新時間戳
        public void setLastUpdate(long lastUpdate) {
            this.lastUpdate = lastUpdate;
        }
    }

    private ConcurrentMap<String, ConcurrentMap<String, WeightedRoundRobin>> methodWeightMap = new ConcurrentHashMap<String, ConcurrentMap<String, WeightedRoundRobin>>();

    /**
     * get invoker addr list cached for specified invocation
     * <p>
     * <b>for unit test only</b>
     *
     * @param invokers
     * @param invocation
     * @return
     */
    protected <T> Collection<String> getInvokerAddrList(List<Invoker<T>> invokers, Invocation invocation) {
        String key = invokers.get(0).getUrl().getServiceKey() + "." + invocation.getMethodName();
        Map<String, WeightedRoundRobin> map = methodWeightMap.get(key);
        if (map != null) {
            return map.keySet();
        }
        return null;
    }

    @Override
    protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
        // {group}/{interfaceName}:{version} + methoName 獲取當前消費者的唯一標示
        String key = invokers.get(0).getUrl().getServiceKey() + "." + invocation.getMethodName();
        // 獲取對應的 WeightedRoundRobin map,如果不存在,new 一個map放進去
        ConcurrentMap<String, WeightedRoundRobin> map = methodWeightMap.computeIfAbsent(key, k -> new ConcurrentHashMap<>());
        int totalWeight = 0;
        long maxCurrent = Long.MIN_VALUE;
        long now = System.currentTimeMillis();
        Invoker<T> selectedInvoker = null;
        WeightedRoundRobin selectedWRR = null;
        for (Invoker<T> invoker : invokers) {
            // 服務提供者在的唯一標識
            String identifyString = invoker.getUrl().toIdentityString();
            int weight = getWeight(invoker, invocation);
            WeightedRoundRobin weightedRoundRobin = map.computeIfAbsent(identifyString, k -> {
                WeightedRoundRobin wrr = new WeightedRoundRobin();
                wrr.setWeight(weight);
                return wrr;
            });
            // 如果權重改變了,更新 weightedRoundRobin 裏面權重的值
            if (weight != weightedRoundRobin.getWeight()) {
                //weight changed
                weightedRoundRobin.setWeight(weight);
            }
            // 當前權重自增自身權重
            long cur = weightedRoundRobin.increaseCurrent();
            // 設置最後一次更新時間戳
            weightedRoundRobin.setLastUpdate(now);
            // 如果當前權重大於最大當前權重
            if (cur > maxCurrent) {
                // 重置最大當前權重的值
                maxCurrent = cur;
                // 把當前提供者設為選中的提供者
                selectedInvoker = invoker;
                // 把當前輪訓權重實例設為選中
                selectedWRR = weightedRoundRobin;
            }
            // 累計總權重
            totalWeight += weight;
        }
        // 提供者有變化
        if (invokers.size() != map.size()) {
            // 超過60s沒有使用,刪除掉
            map.entrySet().removeIf(item -> now - item.getValue().getLastUpdate() > RECYCLE_PERIOD);
        }
        if (selectedInvoker != null) {
            // 減去總權重
            // 關於這個地方為什麼要減去總權重,是一個很容易造成迷惑的地方
            // 我的理解:每一次調用循環 每個提供者的 當前權重 都會自增自己的權重
            // 因此在選中后(只有一個被選中),再減去總權重,正好保證了所有 WeightedRoundRobin 中當前權重之和永遠等於0
            selectedWRR.sel(totalWeight);
            return selectedInvoker;
        }
        // 理論上不會走到這個地方
        // should not happen here
        return invokers.get(0);
    }

}

LeastActiveLoadBalance

最少活躍數調用算法是指在調用時判斷此時每個服務提供者此時正在處理的請求個數,選取最小的調用

dubbo 在實現該算法時的具體邏輯如下

  1. 選取所有活躍數最少的提供者
  2. 如果只有一個,直接返回
  3. 如果權重不同,加權隨機選擇一個
  4. 如果權重相同,隨機選擇一個
    @Override
    protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
        // Number of invokers
        int length = invokers.size();
        // The least active value of all invokers
        // 最少活躍數量
        int leastActive = -1;
        // The number of invokers having the same least active value (leastActive)
        // 有同樣活躍值的提供者數量
        int leastCount = 0;
        // The index of invokers having the same least active value (leastActive)
        int[] leastIndexes = new int[length];
        // the weight of every invokers
        // 每一個提供者的權重
        int[] weights = new int[length];
        // The sum of the warmup weights of all the least active invokers
        // 最少活躍提供者的總權重
        int totalWeight = 0;
        // The weight of the first least active invoker
        int firstWeight = 0;
        // Every least active invoker has the same weight value?
        // 所有的最少活躍提供者是否擁有同樣的權重值
        boolean sameWeight = true;


        // Filter out all the least active invokers
        for (int i = 0; i < length; i++) {
            Invoker<T> invoker = invokers.get(i);
            // Get the active number of the invoker
            // 活躍數量
            int active = RpcStatus.getStatus(invoker.getUrl(), invocation.getMethodName()).getActive();
            // Get the weight of the invoker's configuration. The default value is 100.
            // 獲取權重值
            int afterWarmup = getWeight(invoker, invocation);
            // save for later use
            // 保存權重留着後面用
            weights[i] = afterWarmup;
            // If it is the first invoker or the active number of the invoker is less than the current least active number
            // 如果是第一個提供者,或者當前活躍數量比最少的少
            if (leastActive == -1 || active < leastActive) {
                // Reset the active number of the current invoker to the least active number
                // 重置最少活躍數量
                leastActive = active;
                // Reset the number of least active invokers
                // 重置最少活躍提供者的數量
                leastCount = 1;
                // Put the first least active invoker first in leastIndexes
                // 把最少活躍提供者的索引保存起來
                leastIndexes[0] = i;
                // Reset totalWeight
                // 重置總權重
                totalWeight = afterWarmup;
                // Record the weight the first least active invoker
                // 記錄第一個最少活躍提供者的權重
                firstWeight = afterWarmup;
                // Each invoke has the same weight (only one invoker here)
                // 每個最少活躍提供者是否有同樣的權重???
                sameWeight = true;
                // If current invoker's active value equals with leaseActive, then accumulating.
                // 如果當前活躍數量等於最少活躍數量
            } else if (active == leastActive) {
                // Record the index of the least active invoker in leastIndexes order
                // 最少活躍提供者的索引依次放入 leastIndexes
                leastIndexes[leastCount++] = i;
                // Accumulate the total weight of the least active invoker
                // 累計最少活躍提供者的總權重
                totalWeight += afterWarmup;
                // If every invoker has the same weight?
                // 如果當前權重和第一個最少活躍的權重不同,sameWeight 設為false
                if (sameWeight && afterWarmup != firstWeight) {
                    sameWeight = false;
                }
            }
        }
        // Choose an invoker from all the least active invokers
        // 最少活躍提供者只有一個,直接返回
        if (leastCount == 1) {
            // If we got exactly one invoker having the least active value, return this invoker directly.
            return invokers.get(leastIndexes[0]);
        }
        // 如擁有不同的權重,在權重的基礎上隨機選取一個,可以參考 RandomLoadBalance,有同樣的寫法
        if (!sameWeight && totalWeight > 0) {
            // If (not every invoker has the same weight & at least one invoker's weight>0), select randomly based on 
            // totalWeight.
            int offsetWeight = ThreadLocalRandom.current().nextInt(totalWeight);
            // Return a invoker based on the random value.
            for (int i = 0; i < leastCount; i++) {
                int leastIndex = leastIndexes[i];
                offsetWeight -= weights[leastIndex];
                if (offsetWeight < 0) {
                    return invokers.get(leastIndex);
                }
            }
        }
        // 權重相同,隨機選取一個
        // If all invokers have the same weight value or totalWeight=0, return evenly.
        return invokers.get(leastIndexes[ThreadLocalRandom.current().nextInt(leastCount)]);
    }

ShortestResponseLoadBalance

最短時間調用調用算法是指預估出來每個處理完請求的提供者所需時間,然後又選擇最少最短時間的提供者進行調用,整體處理邏輯和最少活躍數算法基本相似

dubbo 在實現該算法時的具體邏輯如下

  1. 選取所有預估處理時間最短的提供者
  2. 如果只有一個,直接返回
  3. 如果權重不同,加權隨機選擇一個
  4. 如果權重相同,隨機選擇一個
    @Override
    protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
        // Number of invokers
        int length = invokers.size();
        // Estimated shortest response time of all invokers
        // 最少響應時間
        long shortestResponse = Long.MAX_VALUE;
        // The number of invokers having the same estimated shortest response time
        // 最少響應時間的提供者數量
        int shortestCount = 0;
        // The index of invokers having the same estimated shortest response time
        int[] shortestIndexes = new int[length];
        // the weight of every invokers
        int[] weights = new int[length];
        // The sum of the warmup weights of all the shortest response  invokers
        // 最少響應時間的提供者的總權重
        int totalWeight = 0;
        // The weight of the first shortest response invokers
        // 第一個最少響應時間的權重
        int firstWeight = 0;
        // Every shortest response invoker has the same weight value?
        // 所有的最少響應時間提供者是否擁有同樣的權重值
        boolean sameWeight = true;

        // Filter out all the shortest response invokers
        for (int i = 0; i < length; i++) {
            Invoker<T> invoker = invokers.get(i);
            RpcStatus rpcStatus = RpcStatus.getStatus(invoker.getUrl(), invocation.getMethodName());
            // Calculate the estimated response time from the product of active connections and succeeded average elapsed time.
            //  平均響應成功時間
            long succeededAverageElapsed = rpcStatus.getSucceededAverageElapsed();
            // 活躍的連接連接數量
            int active = rpcStatus.getActive();
            // 預估響應時間
            long estimateResponse = succeededAverageElapsed * active;
            // 獲取權重值
            int afterWarmup = getWeight(invoker, invocation);
            // 保存權重留着後面用
            weights[i] = afterWarmup;
            // Same as LeastActiveLoadBalance
            // 如果預估時間小於最少的響應時間
            if (estimateResponse < shortestResponse) {
                // 重置最少響應時間
                shortestResponse = estimateResponse;
                // 最少響應時間的提供者數量設為1
                shortestCount = 1;
                // 保存提供者下標
                shortestIndexes[0] = i;
                // 重置最少響應時間的提供者的總權重
                totalWeight = afterWarmup;
                // 重置第一個最少響應時間的權重
                firstWeight = afterWarmup;
                sameWeight = true;
                // 如果當前最少響應時間等於最少響應時間
            } else if (estimateResponse == shortestResponse) {
                // 最少最少響應時間的下標依次放入 shortestIndexes
                shortestIndexes[shortestCount++] = i;
                // 累計最少響應時間的總權重
                totalWeight += afterWarmup;
                // 如果當前權重和第一個最少響應時間的權重不同,sameWeight 設為false
                if (sameWeight && i > 0
                        && afterWarmup != firstWeight) {
                    sameWeight = false;
                }
            }
        }
        // 最少最少響應時間只有一個,直接返回
        if (shortestCount == 1) {
            return invokers.get(shortestIndexes[0]);
        }
        // 如擁有不同的權重,在權重的基礎上隨機選取一個,可以參考 RandomLoadBalance,有同樣的寫法
        if (!sameWeight && totalWeight > 0) {
            int offsetWeight = ThreadLocalRandom.current().nextInt(totalWeight);
            for (int i = 0; i < shortestCount; i++) {
                int shortestIndex = shortestIndexes[i];
                offsetWeight -= weights[shortestIndex];
                if (offsetWeight < 0) {
                    return invokers.get(shortestIndex);
                }
            }
        }
        // 權重相同,隨機選取一個
        return invokers.get(shortestIndexes[ThreadLocalRandom.current().nextInt(shortestCount)]);
    }

ConsistentHashLoadBalance

一致性hash算法是一種廣泛應用與分佈式緩存中的算法,該算法的優勢在於新增和刪除節點后,只有少量請求發生變動,大部分請求仍舊映射到原來的節點
為了防止節點過少,造成節點分佈不均勻,一般採用虛擬節點的方式,dubbo默認的是160個虛擬節點

網上關於一致性hash算法的文章有很多,這裏就不再多贅述,以下是dubbo中的實現,需要說明的是, 一致性hash算法中權重配置不起作用

    @Override
    protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
        String methodName = RpcUtils.getMethodName(invocation);
        // {group}/{interfaceName}:{version} + methoName 獲取當前消費者的唯一標示
        String key = invokers.get(0).getUrl().getServiceKey() + "." + methodName;
        // using the hashcode of list to compute the hash only pay attention to the elements in the list
        int invokersHashCode = invokers.hashCode();
        // 獲取當前消費者的一致性hash選擇器
        ConsistentHashSelector<T> selector = (ConsistentHashSelector<T>) selectors.get(key);
        // 如果 selector 還沒初始化,或者 invokers 已經變化,重新初始化 selector
        if (selector == null || selector.identityHashCode != invokersHashCode) {
            selectors.put(key, new ConsistentHashSelector<T>(invokers, methodName, invokersHashCode));
            selector = (ConsistentHashSelector<T>) selectors.get(key);
        }
        return selector.select(invocation);
    }
    // 一致性hash選擇器
    private static final class ConsistentHashSelector<T> {

        // 存儲hash環的數據結構 節點 -> 提供者
        private final TreeMap<Long, Invoker<T>> virtualInvokers;

        // 虛擬節點數量
        private final int replicaNumber;

        // 用來標示所有提供者是唯一標示
        private final int identityHashCode;
        // 用來存儲計算hash值參數下標的數組,例如計算第一個和第三個參數 該數組為[0,2]
        private final int[] argumentIndex;

        ConsistentHashSelector(List<Invoker<T>> invokers, String methodName, int identityHashCode) {
            this.virtualInvokers = new TreeMap<Long, Invoker<T>>();
            this.identityHashCode = identityHashCode;
            URL url = invokers.get(0).getUrl();
            // 虛擬節點數量,默認 160
            this.replicaNumber = url.getMethodParameter(methodName, HASH_NODES, 160);
            // 默認只對第一個參數進行hash
            String[] index = COMMA_SPLIT_PATTERN.split(url.getMethodParameter(methodName, HASH_ARGUMENTS, "0"));
            argumentIndex = new int[index.length];
            for (int i = 0; i < index.length; i++) {
                argumentIndex[i] = Integer.parseInt(index[i]);
            }
            for (Invoker<T> invoker : invokers) {
                String address = invoker.getUrl().getAddress();
                // 關於這個地方為什麼要除以4,我理解的是md5後為16字節的數組,計算hash值只需要用到四個字節,所以可以用四次
                // 因此除以4,算是一個性能優化點
                for (int i = 0; i < replicaNumber / 4; i++) {
                    // md5, 獲得一個長度為16的字節數組
                    byte[] digest = md5(address + i);
                    for (int h = 0; h < 4; h++) {
                        // 如果h=0,則用第0,1,2,3四個字節進行位運算,得出一個0-2^32-1的值
                        // 如果h=1,則用第4,5,6,7四個字節進行位運算,得出一個0-2^32-1的值
                        // 如果h=2,則用第8,9,10,11四個字節進行位運算,得出一個0-2^32-1的值
                        // 如果h=3,則用第12,13,14,15四個字節進行位運算,得出一個0-2^32-1的值
                        long m = hash(digest, h);
                        virtualInvokers.put(m, invoker);
                    }
                }
            }
        }

        public Invoker<T> select(Invocation invocation) {
            String key = toKey(invocation.getArguments());
            byte[] digest = md5(key);
            return selectForKey(hash(digest, 0));
        }
        // 根據配置生成計算hash值的key
        private String toKey(Object[] args) {
            StringBuilder buf = new StringBuilder();
            for (int i : argumentIndex) {
                if (i >= 0 && i < args.length) {
                    buf.append(args[i]);
                }
            }
            return buf.toString();
        }

        private Invoker<T> selectForKey(long hash) {
            // 找到hash值在hash環上的位置
            // ceilingEntry 方法返回大於或者等於當前key的鍵值對
            Map.Entry<Long, Invoker<T>> entry = virtualInvokers.ceilingEntry(hash);
            // 如果返回為空,說明落在了hash環中2的32次方-1的最後,直接返回第一個
            if (entry == null) {
                entry = virtualInvokers.firstEntry();
            }
            return entry.getValue();
        }
        // 得出一個0-2^32-1的值, 四個字節組成一個長度為32位的二進制数字並轉化為long值
        private 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;
        }

        private 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();
        }

    }

總結

以上就是dubbo負載均衡源碼的全部解析,如果還是不明白,可以看下官方文檔的解析  
http://dubbo.apache.org/zh-cn/docs/source_code_guide/loadbalance.html

dubbo的負載均衡算法總體來說並不複雜,代碼寫的也很優雅,簡潔,看起來很舒服,而且有很多細節的處理值得稱讚,例如預熱處理,輪訓算法的平滑處理等。

我們平時使用時,可以根據自己的業務場景,選擇適合自己的算法,當然,一般情況下,默認的的隨機算法就能滿足我們的日常需求,而且隨機算法的性能足夠好。

如果覺得dubbo提供的五種算法都不能滿足自己的需求,還可以通過dubbo的SPI機制很方便的擴展自己的負載均衡算法。

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

台北網頁設計公司這麼多,該如何挑選?? 網頁設計報價省錢懶人包"嚨底家"

網頁設計公司推薦更多不同的設計風格,搶佔消費者視覺第一線

※想知道購買電動車哪裡補助最多?台中電動車補助資訊懶人包彙整

南投搬家費用,距離,噸數怎麼算?達人教你簡易估價知識!

※教你寫出一流的銷售文案?

Shader專題:卡通着色(一)控制顏色的藝術

什麼是 Shader?

關於什麼是 Shader ,各種百科各種教程都有說過,但是今天我們就從一個另一個角度去試着理解什麼是 Shader?

我們先看下 Shade 的英文意思,如下:
v.給…遮擋(光線);把…塗暗

其中 把…塗暗 更貼近我們想要的意思。
所以:Shader 這個單詞從字面上理解,就是把什麼東西塗暗。

再強調一次:Shader 從單詞字面上理解,就是把什麼東西塗暗。
再強調一次:把什麼東西塗暗的就是 Shader,就是着色器。

Shader 把什麼塗暗了?

當然是遊戲世界的各個物體,總所周知:有光明就有黑暗,有光照物體就有明暗對比,同時也會有陰影,而 Shader 之所以叫 Shader 是因為起初的時候,Shader 就是用來給物體增加明暗對比的,有了明暗對比,物體在遊戲世界中就會更加立體,從而畫面會更加真實。

所以 Shader 的作用就是給物體添加明暗對比。

Shader 為什麼叫 Shader

當然以上純屬個人推測。現在 Shader 不止可以給物體添加明暗對比,而且還可以做很多濾鏡效果,也可以做很多性能優化(比如減少包大小、減少圖片內存等)的事情。

也許,一開始給 Shader 起名叫 Shader 的時候,Shader 功能非常有限,僅僅只是給物體添加明暗對比(也就是光照計算),後來由於硬件和軟件的發展, 很多離線渲染(電影 CG)的算法都逐步應用在實時渲染(主要是 遊戲 和3D 仿真等),Shader 能做的事情就越來越多,發展到今天,Shader 主要的功能並不只有光照計算。這樣導致,在概念理解上給很多初學者增加了很多阻礙。

教練有一次聽過一位搞圖形學的朋友說:“我們搞實時渲染的都是那些搞視頻(離線渲染)玩剩的”。

Shader 是着色器

什麼是 Shader,中文叫做着色器,也就是給物體上色的意思,也就是說寫 Shader 就是給物體上色的藝術。而這個上色不只是簡單的色彩填充,而是涵蓋了非常多的技巧(幾何計算、顏色計算、貼圖等)

所以中文的着色器,是一個非常精準的翻譯。

群內的笑笑說了一個比較不錯的說法:Shader 主要是光線數據作用在不同數據的物體上產生不同效果。

Shader 學習的順序

不管是 Shader 還是其它某個科目,都有一些最常用、最簡單的知識點。

而這些知識點很容易學以致用,也就是說,這種知識點,我們學習完了就能馬上落地。

所以,教練要做的就是,把 Shader 中的知識點按照是否常用和是否簡單這兩個維度進行排列篩選,然後把它們一個個整理成案例,這樣童鞋們的學習體驗就會大幅上升。

主題式研究第三個階段

  • 第一個階段:確定主題(關鍵字)
  • 第二個階段:搜索資料、搜索信息(搜集情報)
  • 第三個階段:構建知識體系(畫腦圖、寫大綱)

到此,Shader 這個主題,我們目前已經到了第三個階段,也就是構建知識體系的階段。

當然,這一整篇,都再講,我們要怎麼怎麼做,接下來幹嗎,並沒有學習 Shader 的任何一個知識點。

那麼今天就學習一點 Shader 知識意思一下。

顏色的控制

現有一張貼圖,如下:

用來控制顏色的 shader 代碼如下:

float4 frag (v2f i) : SV_Target
{
    // 圖片上每個像素的顏色值
    float4 color = tex2D(_MainTex, i.uv);
                
    // 返回顏色,表示將改像素的顏色值輸出到屏幕上
    return color;
}

我們只看方法中的代碼,先不要在意一些細節。

雖然,我們沒有 Shader 的語法學習經驗,但是憑我們的 C# 經驗,可以將上述代碼推測個大概來。

首先 float4 是一個類型,可以存儲 4 個 float 數值。而顏色一般都是由 r(red 紅色)、g(green,綠色)、b(blue,藍色)、a(alpha,透明度) 四個值控制。所以 float4 可以存儲一個顏色。

現在,我們把圖片中每個像素顏色重的紅色值設置為 0,圖片結果則如下所示:

代碼如下所示:

float4 frag (v2f i) : SV_Target
{
    // 圖片上每個像素的顏色值
    float4 color = tex2D(_MainTex, i.uv);
                
    color.r = 0;

    // 返回顏色,表示將改像素的顏色值輸出到屏幕上
    return color;
}

我們看到,圖片變成了藍綠色。

小結

Shader 是一門控制顏色的藝術,Shader 的核心也是如此。
在此篇,我們學習了 Shader 的兩個重要知識點:

  1. float4 結構體
  2. 顏色的 rgb 控制

這兩個知識點非常簡單,也非常基礎,但是是非常常用的兩個知識點。

這片文章的內容就這些。

知識地圖

相關下載:

轉載請註明地址:liangxiegame.com

更多內容
QFramework 地址:https://github.com/liangxiegame/QFramework
QQ 交流群:623597263
涼鞋的主頁:https://liangxiegame.com/zhuanlan
關注公眾號:liangxiegame 獲取第一時間更新通知及更多的免費內容。

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

網頁設計公司推薦更多不同的設計風格,搶佔消費者視覺第一線

※廣告預算用在刀口上,網站設計公司幫您達到更多曝光效益

※自行創業 缺乏曝光? 下一步"網站設計"幫您第一時間規劃公司的門面形象

南投搬家前需注意的眉眉角角,別等搬了再說!

新北清潔公司,居家、辦公、裝潢細清專業服務

歐盟2020年環保目標難達陣 生物多樣性挑戰尤多

摘錄自2019年12月4日中央社報導

聯合國氣候變化綱要公約第25次締約方會議(COP25)2日在西班牙馬德里開議,將持續至13日。歐洲環保署在配合會議出版的報告中指出,儘管大部分原定2020年達成的環保目標勢必已無法達成,尤其是在生物多樣性領域,歐盟仍有機會實現為2030年和2050年設定的較長遠目標。

報告強調,有鑑於生物多樣性降低的程度令人憂心、氣候變遷衍生的多方面衝擊日益嚴重,以及天然資源遭過度消耗,歐洲必須在未來10年儘速行動。

報告指出,儘管1990至2017年期間,歐洲的溫室氣體排放量已減少22%,且再生能源的使用比例也提升,歐洲在環保領域仍有進步空間。

根據歐洲環保署,在為2020年設定的13個生物多樣性政策目標中,只有兩個達標:劃設海洋保護區和陸地保護區。然而,物種、天然棲地、水生態系統、溼地和土壤狀況的保護,以及化學物排放與空氣和噪音的污染,仍令人擔憂。

本站聲明:網站內容來源環境資訊中心https://e-info.org.tw/,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

台北網頁設計公司這麼多,該如何挑選?? 網頁設計報價省錢懶人包"嚨底家"

網頁設計公司推薦更多不同的設計風格,搶佔消費者視覺第一線

※想知道購買電動車哪裡補助最多?台中電動車補助資訊懶人包彙整

南投搬家費用,距離,噸數怎麼算?達人教你簡易估價知識!

※教你寫出一流的銷售文案?

澳洲野火持續燃燒 摧毀約四成台灣面積

整理:劉妙慈(環境資訊中心實習編輯)

本站聲明:網站內容來源環境資訊中心https://e-info.org.tw/,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

網頁設計公司推薦更多不同的設計風格,搶佔消費者視覺第一線

※廣告預算用在刀口上,網站設計公司幫您達到更多曝光效益

※自行創業 缺乏曝光? 下一步"網站設計"幫您第一時間規劃公司的門面形象

南投搬家前需注意的眉眉角角,別等搬了再說!

新北清潔公司,居家、辦公、裝潢細清專業服務

德國總理許諾為汽車製造商掃除法律障礙 推動無人車、電動車研發

德國總理安格拉•默克爾稱,自動駕駛汽車很快便能在德國進行上路測試,許諾為汽車製造商們掃除法律障礙。

德國擁有全球最大的幾家汽車製造商。安格拉•默克爾表示,德國汽車行業應該起草一份請願書提交柏林,以加快無人駕駛汽車的研發與推出,請願書中最好附上時間規劃。

目前,全球的汽車製造商都在致力於研發自動駕駛汽車,然而其原型至少將在2020年才能獲得推出。安格拉•默克爾12日在柏林的一次工業活動中透露,德國政府機關將於5月底舉行會議,討論下一步工作計畫,如若待辦事項均已完成,內閣便可開始推進車輛測試相關法律依據的制定工作。她告知戴姆勒集團CEO蔡澈(Dieter Zetsche)等稱,“這一話題在政府內部並不存在爭議。”

德國政府還考慮支持電動車的研發工作,拉動消費者需求。然而德國內政部長朔伊布勒(Wolfgang Schaeuble)上月表示,政府機關會設法支援電動車研發,但可能無法滿足汽車製造商們的所有願望。

德國工業領袖已經向政府施壓,要求推出激勵措施拉動電動車需求增長,並稱如果德國想要保持在汽車製造行業領先,那麼推出激勵措施是必需的。

社會民主黨資深議員Hubertus Heil則對此表示,執政聯盟將在本週三的會議上就相關問題達成一致,“我相信我們能夠說服朔伊布勒。”

本站聲明:網站內容來源於EnergyTrend https://www.energytrend.com.tw/ev/,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

台北網頁設計公司這麼多,該如何挑選?? 網頁設計報價省錢懶人包"嚨底家"

網頁設計公司推薦更多不同的設計風格,搶佔消費者視覺第一線

※想知道購買電動車哪裡補助最多?台中電動車補助資訊懶人包彙整

南投搬家費用,距離,噸數怎麼算?達人教你簡易估價知識!

※教你寫出一流的銷售文案?

Nissan、BMW聯手 在美安裝更多電動車充電樁

Nissan與BMW宣布,為共同強化各自旗下的電動車產品Nissan LEAF與BMW i3的客戶服務網,將合作在全美19州、共120處安裝雙頭快速充電樁。此舉未來可望刺激美國市場的電動車銷售量,並幫助電動車車主享有更便捷、廣泛的服務。

Nissan LEAF是目前全球總銷量最好的電動車款之一,充電樁設備分布在社區、企業公司以及Nissan的經銷門市。與BMW的合作,將可加速提升充電樁設備的數量,強化用戶使用經驗同時開拓市場。

BMW的北美電子移動經理Cliff Fietzek強調,與Nissan的合作還包含遠程行車的服務。一但電動車能提供更遠程的行駛距離,使用者就更能體驗電子移動科技的便利之處。

這項合作將以Nissan的充電樁技術與BMW的DC快充技術為基礎,每座快充充電樁都有CHAdeMO與SAE Combo雙街頭,輸出功率50kW,可適用於Nissan LEAF與BMW i3兩種車款以及其他擁有快充埠的電動車。50kW的快充設備可在20-30分鐘之內將Nissan LEAF或BMW i3的電池充到80%左右,比目前最常見的Level 2 240V 充電服務的效率更好。

Nissan與BMW的電動車充電樁服務範圍包括:加州、康乃迪克州、佛羅里達、喬治亞州、伊利諾州、印第安納州、馬里蘭州、明尼蘇達州、密蘇里州、新墨西哥州、內華達州、紐約、南北卡羅萊納州、俄亥俄州、賓州、田納西州、維吉尼亞州以及威斯康辛州。只要透過BWM i3的ConnectedDrive軟體,或者Nissan EZ-Charge智慧型手機app,就能快速找到充電站位置,電費可用Nissan EZ-Charge卡片結帳。

本站聲明:網站內容來源於EnergyTrend https://www.energytrend.com.tw/ev/,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

網頁設計公司推薦更多不同的設計風格,搶佔消費者視覺第一線

※廣告預算用在刀口上,網站設計公司幫您達到更多曝光效益

※自行創業 缺乏曝光? 下一步"網站設計"幫您第一時間規劃公司的門面形象

南投搬家前需注意的眉眉角角,別等搬了再說!

新北清潔公司,居家、辦公、裝潢細清專業服務

PCA降維的原理、方法、以及python實現。

參考:菜菜的sklearn教學之降維算法.pdf!!

PCA(主成分分析法)

1. PCA(最大化方差定義或者最小化投影誤差定義)是一種無監督算法,也就是我們不需要標籤也能對數據做降維,這就使得其應用範圍更加廣泛了。那麼PCA的核心思想是什麼呢?

  • 例如D維變量構成的數據集,PCA的目標是將數據投影到維度為K的子空間中,要求K<D且最大化投影數據的方差。這裏的K值既可以指定,也可以利用主成分的信息來確定。
  • PCA其實就是方差與協方差的運用。
  • 降維的優化目標:將一組 N 維向量降為 K 維,其目標是選擇 K 個單位正交基,使得原始數據變換到這組基上后,各變量兩兩間協方差為 0,而變量方差則盡可能大(在正交的約束下,取最大的 K 個方差)。

2. PCA存在的問題:

  • 原來的數據中比如包括了年齡,性別,身高等指標降維后的數據既然維度變小了,那麼每一維都是什麼含義呢?這個就很難解釋了,所以PCA本質來說是無法解釋降維后的數據的物理含義,換句話說就是降維完啦計算機能更好的認識這些數據,但是咱們就很難理解了。
  • PCA對數據有兩個假設:數據必須是連續數值型;數據中沒有缺失值。
  • 過擬合:PCA 保留了主要信息,但這個主要信息只是針對訓練集的,而且這個主要信息未必是重要信息。有可能捨棄了一些看似無用的信息,但是這些看似無用的信息恰好是重要信息,只是在訓練集上沒有很大的表現,所以 PCA 也可能加劇了過擬合;

3. PCA的作用:

  • 緩解維度災難:PCA 算法通過捨去一部分信息之後能使得樣本的採樣密度增大(因為維數降低了),這是緩解維度災難的重要手段;
  • 降噪:當數據受到噪聲影響時,最小特徵值對應的特徵向量往往與噪聲有關,將它們捨棄能在一定程度上起到降噪的效果;
  • 特徵獨立:PCA 不僅將數據壓縮到低維,它也使得降維之後的數據各特徵相互獨立

4. 方差的作用:咱們可以想象一下,如果一群人都堆疊在一起,我們想區分他們是不是比較困難,但是如果這群人站在馬路兩側,我們就可以很清晰的判斷出來應該這是兩伙人。所以基於方差我們可以做的就是讓方差來去判斷咱們數據的擁擠程度,在這裏我們認為方差大的應該辨識度更高一些,因為分的比較開(一條馬路給隔開啦)。方差可以度量數值型數據的,數據若是想要區分開來,他那他們的離散程度就需要比較大,也就是方差比較大。

5. 協方差的作用:

6. 計算過程:(下圖為採用特徵值分解的計算過程,若採用SVM算法,則無需計算協方差矩陣!)

為什麼我們需要協方差矩陣?我們最主要的目的是希望能把方差和協方差統一到一個矩陣里,方便後面的計算。

  假設我們只有 a 和 b 兩個變量,那麼我們將它們按行組成矩陣 X:(與matlab不同的是,在numpy中每一列表示每個樣本的數據,每一行表示一個變量。比如矩陣X,該矩陣表示的意義為:有m個樣本點,每個樣本點由兩個變量組成!)

  然後:

          

  Cov(a,a) = E[(a-E(a))(a-E(a))], Cov(b,a) = E[(b-E(b))(a-E(a))],因為E(b)=E(a)=0,所以大大簡化了計算!!!(這就體現了去中心化的作用!)

  我們可以看到這個矩陣對角線上的分別是兩個變量的方差,而其它元素是 a 和 b 的協方差。兩者被統一到了一個矩陣里。

7. 特徵值與特徵向量的計算方法—--特徵值分解奇異值分解法(SVD)(有關特徵值與奇異值可見我的博文!)

(1) 特徵值分解的求解過程較為簡單,以下圖為例子

(2) 特徵值分解存在的缺點:

  • 特徵值分解中要求協方差矩陣A必須是方陣,即規模必須為n*n。
  • 後期計算最小投影維度K時,計算量過大。
  • 當樣本維度很高時,協方差矩陣計算太慢;

(3) SVD算法(奇異值分解)的提出克服這些缺點,目前幾乎所有封裝好的PCA算法內部採用的都是SVD算法進行特徵值、特徵向量以及K值的求解。

  • 奇異值(每個矩陣都有):設A是一個mXn矩陣,稱正半定矩陣A‘A的特徵值的非負平方根為矩陣A的奇異值,其中A‘表示矩陣A的共扼轉置矩陣(實數矩陣的共軛就是轉置矩陣,複數矩陣的共軛轉置矩陣就是上面所說的行列互換后每個元素取共軛)
  • 只有方陣才有特徵值。

(4) SVD算法的計算過程:(numpy中已經將SVD進行了封裝,所以只需要調用即可)

可以發現,採用SVD算法無需計算協方差矩陣,這樣在數據量非常大的時候可以降低消耗。

  • A為數據矩陣,大小為M*N(2*5)
  • U是一個由與數據點之間具有最小投影誤差的方向向量所構成的矩陣,大小為M*M(2*2),假如想要將數據由M維降至K維,只需要從矩陣U中選擇前K個列向量,得到一個M*K的矩陣,記為Ureduce。按照下面的公式即可計算降維后的新數據:降維后的數據矩陣G = A.T * Ureduce. 
  • sigma為一個列向量,其包含的值為矩陣A的奇異值。
  • VT是一個大小為N*N的矩陣,具體意義我們無需了解。

利用python實現PCA降維(採用SVD的方法):

 1 from numpy import linalg as la
 2 import numpy as np
 3 #1.矩陣A每個變量的均值都為0,所以不用進行“去平均值”處理。倘若矩陣A的每個變量的均值不為0,則首先需要對數據進行預處理
 4 #  才可以進行協方差矩陣的求解。
 5 #2.與matlab不同的是,在numpy中每一列表示每個樣本的數據,每一行表示一個變量。
 6 #  比如矩陣A,該矩陣表示的意義為:有5個樣本點,每個樣本點由兩個變量組成!
 7 #3.np.mat()函數中矩陣的乘積可以使用 * 或 .dot()函數
 8 #  array()函數中矩陣的乘積只能使用 .dot()函數。而星號乘(*)則表示矩陣對應位置元素相乘,與numpy.multiply()函數結果相同。
 9 A = np.mat([[-1, -1, 0, 2, 0], [-2, 0, 0, 1, 1]])
10 # A = np.mat([[-1, -2], [-1, 0], [0, 0], [2, 1], [0, 1]]).T
11 U, sigma, VT = la.svd(A)
12 print("U:")
13 print(U)
14 print("sigma:")
15 print(sigma)
16 print("VT:")
17 print(VT)
18 print("-"*30)
19 print("降維前的數據:")
20 print(A.T)
21 print("降維后的數據:")
22 print(A.T * U[:,0])

運行結果圖:與上文採用特徵值分解所得到的降維結果一致!

8.PCA的重建

 眾所周知,PCA可以將高維數據壓縮為較少維度的數據,由於維度有所減少,所以PCA屬於有損壓縮,也就是,壓縮后的數據沒有保持原來數據的全部信息,根據壓縮數據無法重建原本的高維數據,但是可以看作原本高維數據的一種近似。

 還原的近似數據矩陣Q = 降維后的矩陣G * Ureduce.T

9.採用sklearn封裝好的PCA實現數據降維(採用的是SVD算法):

 1 import numpy as np
 2 from sklearn.decomposition import PCA
 3 # 利用sklearn進行PCA降維處理的時候,數據矩陣A的行數表示數據的個數,數據矩陣A的列數表示每條數據的維度。這與numpy中是相反的!
 4 # A = np.mat([[-1, -1, 0, 2, 0], [-2, 0, 0, 1, 1]]).T
 5 A = np.mat([[-1, -2], [-1, 0], [0, 0], [2, 1], [0, 1]])
 6 pca = PCA(n_components = 1)
 7 pca.fit(A)
 8 # 投影后的特徵維度的方差比例
 9 print(pca.explained_variance_ratio_)
10 # 投影后的特徵維度的方差
11 print(pca.explained_variance_)
12 print(pca.transform(A))

 可以發現,採用sklearn封裝的方法實現PCA與上文的方法達到的結果一致!

10.如何確定主成分數量(針對於Sklearn封裝的PCA方法而言)

PCA算法將D維數據降至K維,顯然K是需要選擇的參數,表示要保持信息的主成分數量。我們希望能夠找到一個K值,既能大幅降低維度,又能最大限度地保持原有數據內部的結構信息。實現的過程是通過SVD方法得到的S矩陣進行操作求解,

 

11.sklearn中封裝的PCA方法的使用介紹。

PCA的函數原型

 (1)主要參數介紹

n_components

  • 這個參數類型有int型,float型,string型,默認為None。 它的作用是指定PCA降維后的特徵數(也就是降維后的維度)。 
  • 若取默認(None),則n_components==min(n_samples, n_features),即降維后特徵數取樣本數和原有特徵數之間較小的那個;
  • 若n_components}設置為‘mle’並且svd_solver設置為‘full’則使用MLE算法根據特徵的方差分佈自動去選擇一定數量的主成分特徵來降維; 
  • 若0<n_components<1,則n_components的值為主成分方差的閾值; 通過設置該變量,即可調整主成分數量K。
  • 若n_components≥1,則降維后的特徵數為n_components; 

copy

  •  bool (default True) 
  • 在運行算法時,將原始訓練數據複製一份。參數為bool型,默認是True,傳給fit的原始訓練數據X不會被覆蓋;若為False,則傳給fit后,原始訓練數據X會被覆蓋。 

whiten

  • bool, optional (default False)
  • 是否對降維后的數據的每個特徵進行歸一化。參數為bool型,默認是False。

(2)主要方法介紹:

fit(X,y=None) :用訓練數據X訓練模型,由於PCA是無監督降維,因此y=None。 

transform(X,y=None) :對X進行降維。 

fit_transform(X) :用訓練數據X訓練模型,並對X進行降維。相當於先用fit(X),再用transform(X)。 

inverse_transform(X) :將降維后的數據轉換成原始數據。(PCA的重建)

 (3)主要屬性介紹:

components:array, shape (n_components, n_features) ,降維后各主成分方向,並按照各主成分的方差值大小排序。 

explained_variance:array, shape (n_components,) ,降維后各主成分的方差值,方差值越大,越主要。 

explained_variance_ratio:array, shape (n_components,) ,降維后的各主成分的方差值佔總方差值的比例,比例越大,則越主要。 

singular_values:array, shape (n_components,) ,奇異值分解得到的前n_components個最大的奇異值。

 

 二、LDA

1. 類間距離最大,類內距離最小(核心思想)

2. LDA的原理,公式推導見西瓜書,這裏主要講一下PCA與LDA的異同點!

  • PCA為非監督降維,LDA為有監督降維PCA希望投影后的數據方差盡可能的大(最大可分性),因為其假設方差越多,則所包含的信息越多;而LDA則希望投影后相同類別的組內方差小,而組間方差大。LDA能合理運用標籤信息,使得投影后的維度具有判別性,不同類別的數據盡可能的分開。舉個簡單的例子,在語音識別領域,如果單純用PCA降維,則可能功能僅僅是過濾掉了噪聲,還是無法很好的區別人聲,但如果有標籤識別,用LDA進行降維,則降維后的數據會使得每個人的聲音都具有可分性,同樣的原理也適用於臉部特徵識別。
  • 所以,可以歸納總結為有標籤就盡可能的利用標籤的數據(LDA),而對於純粹的非監督任務,則還是得用PCA進行數據降維。
  • LDA降維最低可以降維到(類別數-1),而PCA沒有限制

 

參考資料:https://zhuanlan.zhihu.com/p/77151308?utm_source=qq&utm_medium=social&utm_oi=1095998405318430720

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

台北網頁設計公司這麼多,該如何挑選?? 網頁設計報價省錢懶人包"嚨底家"

網頁設計公司推薦更多不同的設計風格,搶佔消費者視覺第一線

※想知道購買電動車哪裡補助最多?台中電動車補助資訊懶人包彙整

南投搬家費用,距離,噸數怎麼算?達人教你簡易估價知識!

※教你寫出一流的銷售文案?

C表達式中的彙編指令

C 表達式中的彙編指令

asm 為 gcc 中的關鍵字,asm 表達式為在 C代碼中嵌套彙編指令,該表達式只是單純的替換出彙編代碼,並不對彙編代碼的含義進行解析。

asm 表達式有兩種形式,第二種 asm-qualifiers 包含了 goto 語句。
第一種形式為常見的用法,AssemblerTemplate 和 OutputOperands 必須存在, 其中 Clobbers 存在需要 InputOperands 也出現。

asm asm-qualifiers ( AssemblerTemplate 
                 : OutputOperands 
                 [ : InputOperands
                 [ : Clobbers ] ])

asm asm-qualifiers ( AssemblerTemplate 
                      : 
                      : InputOperands
                      : Clobbers
                      : GotoLabels)

Qualifiers 的類型

  • volatile, 避免編譯器的優化
  • inline, 內斂限定符,最小的體積
  • goto, 包含跳轉指令

參數

  • AssemblerTemplate
    – 彙編指令模板是包含彙編器指令的文字字符串,編輯器替換引用輸入,編譯器不會解析該指令的含義。
  • OutputOperands
    – 由 AssemblerTemplate 中的指令修改的C變量的逗號分隔列表,允許使用空列表。
  • InputOperands
    – 由 AssemblerTemplate 中的指令讀取的C變量的逗號分隔列表,允許使用空列表。
  • Clobbers
    – 用逗號分隔的寄存器列表或由 AssemblerTemplate 修改的值,不能出現在 OutputOperands 和 InputOperands 中被提及,允許使用空列表。
  • GotoLabels
    – 當使用asm的goto形式時,此部分包含 AssemblerTemplate 中的代碼可能跳轉到的所有C標籤的列表。

AssemblerTemplate

彙編指令由一個字符串給出,多條彙編指令結合在一起使用的時候,中間以 \r\t 隔開,如

asm("inc %0\n\tinc %0" : "=r"(res) : "0"(res));

/APP
# 11 "asm.c" 1
        inc %rax
        inc %rax
# 0 "" 2
/NO_APPs

需要轉義的字符:%, =, {, }, |

故在ATT彙編中,對寄存器進行操作的需要雙 %%, 如 inc %%rax.

OutputOperands

操作數之間用逗號分隔。 每個操作數具有以下格式:

[ [asmSymbolicName] ] constraint (cvariablename)
  • asmSymbolicName
    – 為操作數指定名稱,格式為 %[name]
    c // res = num asm("movq %[num], %[res]" : [res] "=r"(res) : [num] "m"(num));
    – 如果未指定名稱使用数字, 從 output 域開始,第一個參數為 %0, 一次類推, 這裏的 res 為 %0, num 為 %1
    c // res = num asm("movq %1, %0" : "=r"(res) : "m"(num));
  • constraint
    – 一個字符串常量,用於指定對操作數的存儲的 , 需要以 “=” 或 “+” 開頭
  • cvariablename
    – 指定一個C左值表達式來保存輸出,通常是一個變量名。 括號是語法的必需部分

第一個參數為增加可讀性使用的,現在我們有代碼如下

int64_t res;
int64_t num = 1;

asm("movq %[num], %[res]" : [res] "=r"(res) : [num] "m"(num));
asm("movq %1, %0" : "=r"(res) : "m"(num));
asm("movq %1, %0" : "=m"(res) : "m"(num));
asm("movq %1, %0" : "=r"(res) : "r"(num));

// 對應的彙編代碼, 只保留asm表達式中的代碼
# 13 "asm.c" 1
        movq -16(%rbp), %rax  // asm-1
 # 0 "" 2
/NO_APP

/APP
 # 15 "asm.c" 1
        movq -16(%rbp), %rax  // asm-2
 # 0 "" 2
/NO_APP

/APP
 # 17 "asm.c" 1
        movq -16(%rbp), -8(%rbp)  // asm-3
 # 0 "" 2
/NO_APP

/APP
 # 19 "asm.c" 1
        movq %rax, %rax  // asm-4
 # 0 "" 2
/NO_APP
  1. 使用名稱替換和数字替換效果一樣,見 asm-1 和 asm-2
  2. 約束的用法,這裏使用比較簡單通用的的兩種情況,r 為通過寄存器尋址操作,m 通過內存尋址操作,所以看到當約束了 r 就對應寄存器的操作。
  3. 結果保存在 res 也就是 cvariablename 中

InputOperands

輸入操作數使C變量和表達式中的值可用於彙編代碼。

[ [asmSymbolicName] ] constraint (cexpression)
  • asmSymbolicName 和輸出列表的用法完全一致
  • constraint 約束不能使用 =+. 可以使用 “0”, 這表明在輸出約束列表中(從零開始)的條目,指定的輸入必須與輸出約束位於同一位置。
int64_t res = 3;
int64_t num = 1;
asm("addq %1, %0" : "=g"(res) : "0"(num));

// 輸入輸出位置相同
        movq    $3, -8(%rbp)
        movq    $1, -16(%rbp)
        movq    -16(%rbp), %rax
/APP
# 32 "asm.c" 1
        addq %rax, %rax
# 0 "" 2
/NO_APP
  • cexpression 可以不為左值,作為彙編表達式的輸入值即可

Clobbers

破壞列表,主要用於指示編譯器生成的彙編指令。

從asm表達式中看到輸出操作數中列出條目的更改編譯器是可以確定的,但內聯彙編代碼可能不僅對輸出進行了修改。 例如,計算可能需要其他寄存器,或者處理器可能會由於特定彙編程序指令而破壞寄存器的值。 為了將這些更改通知編譯器,在Clobber列表中列出這些會產生副作用的條目。 破壞列表條目可以是寄存器名稱,也可以是特殊的破壞列表項(在下面列出)。 每個內容列表條目都是一個字符串常量,用雙引號引起來並用逗號分隔。

  • 寄存器

      ```c
      asm volatile("movc3 %0, %1, %2"
              : /* No outputs. */
              : "r"(from), "r"(to), "g"(count)
              : "%rbx", "%rcx", "%rdx", "memory");
    
      /APP
      # 25 "asm.c" 1
              movc3 %rax, %r8, -72(%rbp)
      # 0 "" 2
      /NO_APP
      ```
    
      可以看到使用到了 rax 寄存器,然後修改程序在 Clobbers 增加 %rax, 結果如下
    
      ```c
      asm volatile("movc3 %0, %1, %2"
              : /* No outputs. */
              : "r"(from), "r"(to), "g"(count)
              : "%rax", "%rbx", "%rcx", "%rdx", "memory");
    
      /APP
      # 25 "asm.c" 1
              movc3 %r8, %r9, -72(%rbp)
      # 0 "" 2
      /NO_APP
      ```
      編譯器在產生的彙編代碼中就未使用 %rax 寄存器了。
  • 特殊破壞列表項
    – “cc”, 表示彙編代碼修改了標誌寄存器
    – “memory”, 為了確保內存中包含正確的值,編譯器可能需要在執行asm之前將特定的寄存器值刷新到內存中

編譯器為了破壞列表項的值受到破壞,當這些條目是寄存器時,不對其進行使用;為特殊參數時,重新刷新得到最新的值。

約束

  • 一些基礎的約束
約束名 說明
whitespace 空白字符被忽略
m 允許使用內存操作數,以及機器通常支持的任何類型的地址
o 允許使用內存操作數,但前提是地址是可偏移的
V 允許使用內存操作數,不可偏移的內存地址,與 “o’互斥
r 允許在通用寄存器中使用的寄存器操作數,其中可以指定寄存器,如 a(%rax), b(%rbx)
i 允許使用立即整數操作數
n 允許使用具有已知數值的立即整數操作數, ‘I’, ‘J’, ‘K’, … ‘P’ 更應該使用 n
F 允許使用浮點立即數
g 允許使用任何寄存器,內存或立即數整數操作數,但非通用寄存器除外
X 允許任何操作數, ‘0’, ‘1’, ‘2’, … ‘9’
p 允許使用有效內存地址的操作數
  • 標識符約束
標識符 說明
= 表示此操作數是由該指令寫入的:先前的值將被丟棄並由新數據替換
+ 表示該操作數由指令讀取和寫入
& 表示(在特定替代方法中)此操作數是早期指令操作數,它是在使用輸入操作數完成指令之前寫入的,故輸入操作數部分不能分配與輸出操作數相同的寄存器
% 表示該操作數與後續操作數的可交換指令

內核示例

  1. x86 的內存屏障指令。
// 避免編譯器的優化,聲明此處內存可能發生破壞
#define barrier() asm volatile("" ::: "memory")
// 在32位的CPU下,lock 指令為鎖總線,加上一條內存操作指令就達到了內存屏障的作用,64位的cpu已經有新增的 *fence 指令可以使用
// mb() 執行一個內存屏障作用的指令,為指定CPU操作;破壞列表聲明 cc memory 指示避免編譯器進行優化
#ifdef CONFIG_X86_32
#define mb() asm volatile(ALTERNATIVE("lock; addl $0,-4(%%esp)", "mfence", \
                                X86_FEATURE_XMM2) ::: "memory", "cc")
#define rmb() asm volatile(ALTERNATIVE("lock; addl $0,-4(%%esp)", "lfence", \
                                X86_FEATURE_XMM2) ::: "memory", "cc")
#define wmb() asm volatile(ALTERNATIVE("lock; addl $0,-4(%%esp)", "sfence", \
                                X86_FEATURE_XMM2) ::: "memory", "cc")
#else
#define mb()    asm volatile("mfence":::"memory")
#define rmb()   asm volatile("lfence":::"memory")
#define wmb()   asm volatile("sfence" ::: "memory")
#endif
  1. x86 下獲取 current 的值
DECLARE_PER_CPU(struct task_struct *, current_task);

#define this_cpu_read_stable(var)   percpu_stable_op("mov", var)

static __always_inline struct task_struct *get_current(void)
{
        return this_cpu_read_stable(current_task);
}

#define percpu_stable_op(op, var)           \
({                          \
        typeof(var) pfo_ret__;              \
        switch (sizeof(var)) {              \
        case 8:                     \
                asm(op "q "__percpu_arg(P1)",%0"    \
                : "=r" (pfo_ret__)          \
                : "p" (&(var)));            \
                break;                  \
        }                       \
        pfo_ret__;                  \
})

current_task 為一個 struct task_struct 類型的指針,追蹤宏調用,在x86-64 下命中了 case 8: 的彙編代碼, 展開的代碼為

asm("mov" "q ""%%""gs" ":" "%" "P1"",%0" : "=r" (pfo_ret__) : "p" (&(current_task)));
// 變換一下為
asm("movq %%gs:%P1, %0" : "=r"(pfo_ret__) : "p"(&(current_task)));

這行代碼的含義為將 約束輸入部分必須為有效的地址(p約束), 將CPU id(通過段寄存器gs和偏移通過GDT得到,這裏後文分析了)通過寄存器(r約束)賦值給 pfo_ret__.

參考

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

網頁設計公司推薦更多不同的設計風格,搶佔消費者視覺第一線

※廣告預算用在刀口上,網站設計公司幫您達到更多曝光效益

※自行創業 缺乏曝光? 下一步"網站設計"幫您第一時間規劃公司的門面形象

南投搬家前需注意的眉眉角角,別等搬了再說!

新北清潔公司,居家、辦公、裝潢細清專業服務

HtmlSpanner 使用小結 — 安卓解析html

如何利用 HtmlSpanner解析 HTML格式 的字符串:

1. GitHub 下載HtmlSpanner項目 https://github.com/NightWhistler/HtmlSpanner

2. 最好是直接放在java目錄下在,這樣不需要改引用的包路徑

3.  引入需要的依賴包

    implementation 'net.sourceforge.htmlcleaner:htmlcleaner:2.21'
    implementation 'com.osbcp:cssparser:1.7'

  4. 使用方法:

 // 頁面上用於展示 html格式的文本布局
 TextView textView = findViewById(R.id.htmlSpanner);
 // 直接 new一個 HtmlSpanner對象
 HtmlSpanner htmlSpanner = new HtmlSpanner(); // 格式化
 // 解析 html得到 spannable對象
 Spannable spannable1 = htmlSpanner.fromHtml("<span style='color:red'>html格式的文字1</span>");
 // 显示到 TextView上
 textView.setText(spannable1);

 5. 在使用中遇到的問題——當富文本中顏色格式是rgb格式,解析失敗

 

 

 

 

 解決思路:

  1. 首先我們解析的是style=’color:rgb(0,255,255)’ 這種格式,於是看源碼覺得 CSSCompiler 這個類很有問題

  2. 找與顏色相關的於是就找到了 parseCSSColor( String colorString ) 這個方法,看起來就是轉換顏色用的

  3. 源碼的寫法如下:(是沒有對於rgb格式的算法,所以不能解析就很合理啦)

  

 

   4. 想法修改:( 遇到 0rgb格式就先處理成我們的 hex格式,這樣不就完美了嘛 )

  5. 工具類代碼如下:

package com.xxx.xxx.xxx;

public class ColorUtil {

     /**
     * rgb 格式的顏色轉 hex格式顏色
     * @param rgb
     * @return
     */
    public static String rgb2hex(String rgb) {
        int r = 0;
        int g = 0;
        int b = 0;
        int left = rgb.indexOf("(");
        int right = rgb.indexOf(")");
        if (left > -1 && right > -1 && right > left) {
            String substring = rgb.substring(left + 1, right);
            String[] split = substring.split(",");
            if (split.length == 3){
                r = Integer.valueOf(split[0].trim());
                g = Integer.valueOf(split[1].trim());
                b = Integer.valueOf(split[2].trim());
            }
        }
        String rFString, rSString, gFString, gSString,
                bFString, bSString, result;
        int red, green, blue;
        int rred, rgreen, rblue;
        red = r / 16;
        rred = r % 16;
        if (red == 10) rFString = "A";
        else if (red == 11) rFString = "B";
        else if (red == 12) rFString = "C";
        else if (red == 13) rFString = "D";
        else if (red == 14) rFString = "E";
        else if (red == 15) rFString = "F";
        else rFString = String.valueOf(red);

        if (rred == 10) rSString = "A";
        else if (rred == 11) rSString = "B";
        else if (rred == 12) rSString = "C";
        else if (rred == 13) rSString = "D";
        else if (rred == 14) rSString = "E";
        else if (rred == 15) rSString = "F";
        else rSString = String.valueOf(rred);

        rFString = rFString + rSString;

        green = g / 16;
        rgreen = g % 16;

        if (green == 10) gFString = "A";
        else if (green == 11) gFString = "B";
        else if (green == 12) gFString = "C";
        else if (green == 13) gFString = "D";
        else if (green == 14) gFString = "E";
        else if (green == 15) gFString = "F";
        else gFString = String.valueOf(green);

        if (rgreen == 10) gSString = "A";
        else if (rgreen == 11) gSString = "B";
        else if (rgreen == 12) gSString = "C";
        else if (rgreen == 13) gSString = "D";
        else if (rgreen == 14) gSString = "E";
        else if (rgreen == 15) gSString = "F";
        else gSString = String.valueOf(rgreen);

        gFString = gFString + gSString;

        blue = b / 16;
        rblue = b % 16;

        if (blue == 10) bFString = "A";
        else if (blue == 11) bFString = "B";
        else if (blue == 12) bFString = "C";
        else if (blue == 13) bFString = "D";
        else if (blue == 14) bFString = "E";
        else if (blue == 15) bFString = "F";
        else bFString = String.valueOf(blue);

        if (rblue == 10) bSString = "A";
        else if (rblue == 11) bSString = "B";
        else if (rblue == 12) bSString = "C";
        else if (rblue == 13) bSString = "D";
        else if (rblue == 14) bSString = "E";
        else if (rblue == 15) bSString = "F";
        else bSString = String.valueOf(rblue);
        bFString = bFString + bSString;
        result = "#" + rFString + gFString + bFString;
        return result;
    }
}

 

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

台北網頁設計公司這麼多,該如何挑選?? 網頁設計報價省錢懶人包"嚨底家"

網頁設計公司推薦更多不同的設計風格,搶佔消費者視覺第一線

※想知道購買電動車哪裡補助最多?台中電動車補助資訊懶人包彙整

南投搬家費用,距離,噸數怎麼算?達人教你簡易估價知識!

※教你寫出一流的銷售文案?

Redis 的底層數據結構(壓縮列表)

上一篇我們介紹了 redis 中的整數集合這種數據結構的實現,也談到了,引入這種數據結構的一個很大的原因就是,在某些僅有少量整數元素的集合場景,通過整數集合既可以達到字典的效率,也能使用遠少於字典的內存達到同樣的效果。

我們本篇介紹的壓縮列表,相信你從他的名字里應該也能看出來,又是一個為了節約內存而設計的數據結構,它的數據結構相對於整數集合來說會複雜了很多,但是整數集合只能允許存儲少量的整型數據,而我們的壓縮列表可以允許存儲少量的整型數據或字符串。

這是他們之間的一個區別,下面我們來看看這種數據結構。

一、基本的結構定義

  • ZIPLIST_BYTES:四個字節,記錄了整個壓縮列表總共佔用了多少字節數
  • ZIPLIST_TAIL_OFFSET:四個字節,記錄了整個壓縮列表第一個節點到最後一個節點跨越了多少個字節,通故這個字段可以迅速定位到列表最後一個節點位置
  • ZIPLIST_LENGTH:兩個字節,記錄了整個壓縮列表中總共包含幾個 zlentry 節點
  • zlentry:非固定字節,記錄的是單個節點,這是一個複合結構,我們等下再說
  • 0xFF:一個字節,十進制的值為 255,標誌壓縮列表的結尾

其中,zlentry 在 redis 中確實有着這樣的結構體定義,但實際上這個結構定義了一堆類似於 length 這樣的字段,記錄前一個節點和自身節點佔用的字節數等等信息,用處不多,而我們更傾向於使用這樣的邏輯結構來描述 zlentry 節點。

這種結構在 redis 中是沒有具體結構體定義的,請知悉,網上的很多博客文章都直接描述 zlentry 節點是這樣的一種結構,其實是不準確的。

簡單解釋一下這三個字段的含義:

  • previous_entry_length:每個節點會使用一個或者五個字節來描述前一個節點佔用的總字節數,如果前一個節點佔用的總字節數小於 254,那麼就用一個字節存儲,反之如果前一個節點佔用的總字節數超過了 254,那麼一個字節就不夠存儲了,這裡會用五個字節存儲並將第一個字節的值存儲為固定值 254 用於區分。
  • encoding:壓縮列表可以存儲 16位、32位、64位的整數以及字符串,encoding 就是用來區分後面的 content 字段中存儲於的到底是哪種內容,分別佔多少字節,這個我們等下細說。
  • content:沒什麼特別的,存儲的就是具體的二進制內容,整數或者字符串。

下面我們細說一個 encoding 具體是怎麼存儲的。

主要分為兩種,一種是字符串的存儲格式:

編碼 編碼長度 content類型
00xxxxxx 一個字節 長度小於 63 的字符串
01xxxxxx xxxxxxxx 兩個字節 長度小於 16383 的字符串
10xxxxxx xxxxxxxx xxxxxxxx xxxxxxxx xxxxxxxx 五個字節 長度小於 4294967295 的字符串

content 的具體長度,由編碼除去高兩位剩餘的二進制位表示。

編碼 編碼長度 content類型
11000000 一個字節 int16_t 類型的整數
11010000 一個字節 int32_t 類型的整數
11100000 一個字節 int64_t 類型的整數
11110000 一個字節 24 位有符號整數
11111110 一個字節 8 位有符號整數

注意,整型數據的編碼是固定 11 開頭的八位二進制,而字符串類型的編碼都是非固定的,因為它還需要通過後面的二進制位得到字符串的長度,稍有區別。

這就是壓縮列表的基本的結構定義情況,下面我們通過節點的增刪改查方法源碼實現來看看 redis 中具體的實現情況。

二、redis 的具體源碼實現

1、ziplistNew

我們先來看看壓縮列表初始化的方法實現:

unsigned char *ziplistNew(void) {
    //bytes=2*4+2
    //分配壓縮列表結構所需要的字節數
    //ZIPLIST_BYTES + ZIPLIST_TAIL_OFFSET + ZIPLIST_LENGTH
    unsigned int bytes = ZIPLIST_HEADER_SIZE+1;
    unsigned char *zl = zmalloc(bytes);
    //初始化 ZIPLIST_BYTES 字段
    ZIPLIST_BYTES(zl) = intrev32ifbe(bytes);
    //初始化 ZIPLIST_TAIL_OFFSET
    ZIPLIST_TAIL_OFFSET(zl) = intrev32ifbe(ZIPLIST_HEADER_SIZE);
    //初始化 ZIPLIST_LENGTH 字段
    ZIPLIST_LENGTH(zl) = 0;
    //為壓縮列表最後一個字節賦值 255
    zl[bytes-1] = ZIP_END;
    return zl;
}

2、ziplistPush

接着我們看新增節點的源碼實現:

unsigned char *ziplistPush(unsigned char *zl, unsigned char *s
        ,unsigned int slen, int where) {
    unsigned char *p;
    //找到待插入的位置,頭部或者尾部
    p = (where == ZIPLIST_HEAD) ? ZIPLIST_ENTRY_HEAD(zl) : ZIPLIST_ENTRY_END(zl);
    return __ziplistInsert(zl,p,s,slen);
}

解釋一下 ziplistPush 的幾個入參的含義。

zl 指向一個壓縮列表的首地址,s 指向一個字符串首地址),slen 指向字符串的長度(如果節點存儲的值是整型,存儲的就是整型值),where 指明新節點的插入方式,頭插亦或尾插。

ziplistPush 方法的核心是 __ziplistInsert:

unsigned char *__ziplistInsert(unsigned char *zl, unsigned char *p, unsigned char *s, unsigned int slen) {
    size_t curlen = intrev32ifbe(ZIPLIST_BYTES(zl)), reqlen;
    unsigned int prevlensize, prevlen = 0;
    size_t offset;
    int nextdiff = 0;
    unsigned char encoding = 0;
    long long value = 123456789; 
    zlentry tail;
    //prevlensize 存儲前一個節點長度,本節點使用了幾個字節 1 or 5
    //prelen  存儲前一個節點實際佔用了幾個字節
    if (p[0] != ZIP_END) {
        ZIP_DECODE_PREVLEN(p, prevlensize, prevlen);
    } else {
        unsigned char *ptail = ZIPLIST_ENTRY_TAIL(zl);
        if (ptail[0] != ZIP_END) {
            prevlen = zipRawEntryLength(ptail);
        }
    }
    if (zipTryEncoding(s,slen,&value,&encoding)) {
        //s 指針指向一個整數,嘗試進行一個轉換並得到存儲這個整數佔用了幾個字節
        reqlen = zipIntSize(encoding);
    } else {
        //s 指針指向一個字符串(字符數組),slen 就是他佔用的字節數
        reqlen = slen;
    }
    //當前節點存儲數據佔用 reqlen 個字節,加上存儲前一個節點長度佔用的字節數
    reqlen += zipStorePrevEntryLength(NULL,prevlen);
    //encoding 字段存儲實際佔用字節數
    reqlen += zipStoreEntryEncoding(NULL,encoding,slen);
    //至此,reqlen 保存了存儲當前節點數據佔用字節數和 encoding 編碼佔用的字節數總和
    int forcelarge = 0;
    //當前節點佔用的總字節減去存儲前一個節點字段佔用的字節
    //記錄的是這一個節點的插入會引起下一個節點佔用字節的變化量
    nextdiff = (p[0] != ZIP_END) ? zipPrevLenByteDiff(p,reqlen) : 0;
    if (nextdiff == -4 && reqlen < 4) {
        nextdiff = 0;
        forcelarge = 1;
    }
    //擴容有可能導致 zl 的起始位置偏移,故記錄 p 與 zl 首地址的相對偏差數,事後還原 p 指針指向
    offset = p-zl;
    zl = ziplistResize(zl,curlen+reqlen+nextdiff);
    p = zl+offset;
    if (p[0] != ZIP_END) {
        memmove(p+reqlen,p-nextdiff,curlen-offset-1+nextdiff);
        //把當前節點佔用的字節數存儲到下一個節點的頭部字段
        if (forcelarge)
            zipStorePrevEntryLengthLarge(p+reqlen,reqlen);
        else
            zipStorePrevEntryLength(p+reqlen,reqlen);

        //更新 tail_offset 字段,讓他保存從頭節點到尾節點之間的距離
        ZIPLIST_TAIL_OFFSET(zl) =
            intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))+reqlen);
        zipEntry(p+reqlen, &tail);
        if (p[reqlen+tail.headersize+tail.len] != ZIP_END) {
            ZIPLIST_TAIL_OFFSET(zl) =
                intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))+nextdiff);
        }
    } else {
        ZIPLIST_TAIL_OFFSET(zl) = intrev32ifbe(p-zl);
    }
    //是否觸發連鎖更新
    if (nextdiff != 0) {
        offset = p-zl;
        zl = __ziplistCascadeUpdate(zl,p+reqlen);
        p = zl+offset;
    }
    //將節點寫入指定位置
    p += zipStorePrevEntryLength(p,prevlen);
    p += zipStoreEntryEncoding(p,encoding,slen);
    if (ZIP_IS_STR(encoding)) {
        memcpy(p,s,slen);
    } else {
        zipSaveInteger(p,value,encoding);
    }
    ZIPLIST_INCR_LENGTH(zl,1);
    return zl;
}

具體細節我不再贅述,總結一下整個插入節點的步驟。

  1. 計算並得到前一個節點的總長度,並判斷得到當前待插入節點保存前一個節點長度的 previous_entry_length 佔用字節數
  2. 根據傳入的 s 和 slen,計算並保存 encoding 字段內容
  3. 構建節點並將數據寫入節點添加到壓縮列表中

ps:重點要去理解壓縮列表節點的數據結構定義,previous_entry_length、encoding、content 字段,這樣才能比較容易理解節點新增操作的實現。

三、連鎖更新

談到 redis 的壓縮列表,就必然會談到他的連鎖更新,我們先引一張圖:

假設原本 entry1 節點佔用字節數為 211(小於 254),那麼 entry2 的 previous_entry_length 會使用一個字節存儲 211,現在我們新插入一個節點 NEWEntry,這個節點比較大,佔用了 512 個字節。

那麼,我們知道,NEWEntry 節點插入后,entry2 的 previous_entry_length 存儲不了 512,那麼 redis 就會重分配內存,增加 entry2 的內存分配,並分配給 previous_entry_length 五個字節存儲 NEWEntry 節點長度。

看似沒什麼問題,但是如果極端情況下,entry2 擴容四個字節后,導致自身佔用字節數超過 254,就會又觸發后一個節點的內存佔用空間擴大,非常極端情況下,會導致所有的節點都擴容,這就是連鎖更新,一次更新導致大量甚至全部節點都更新內存的分配。

如果連鎖更新發生的概率很高的話,壓縮列表無疑就會是一個低效的數據結構,但實際上連鎖更新發生的條件是非常苛刻的,其一是需要大量節點長度小於 254 連續串聯連接,其二是我們更新的節點位置恰好也導致后一個節點內存擴充更新。

基於這兩點,且少量的連鎖更新對性能是影響不大的,所以這裏的連鎖更新對壓縮列表的性能是沒有多大的影響的,可以忽略,但需要知曉。

同樣的,如果覺得我寫的對你有點幫助的話,順手點一波關注吧,也歡迎加作者微信深入探討,我們逐漸開始走近 redis 比較實用性的相關內容了,盡請關注。

關注公眾不迷路,一個愛分享的程序員。


公眾號回復「1024」加作者微信一起探討學習!


每篇文章用到的所有案例代碼素材都會上傳我個人 github





歡迎來踩!

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

台北網頁設計公司這麼多,該如何挑選?? 網頁設計報價省錢懶人包"嚨底家"

網頁設計公司推薦更多不同的設計風格,搶佔消費者視覺第一線

※想知道購買電動車哪裡補助最多?台中電動車補助資訊懶人包彙整

南投搬家費用,距離,噸數怎麼算?達人教你簡易估價知識!

※教你寫出一流的銷售文案?