架构师训练营 1 期第 5 周：技术选型（一） - 作业

package com.test;

import java.util.*;
import java.util.stream.IntStream;

public class ConsistentHashTest {

    private static final String PRE_KEY = "Test";
    private static final int DATA_COUNT = 1000000;

    //使用FNV1_32_HASH算法计算服务器的Hash值,String默认hashCode导致散列值分布过于集中
    private static int hash(String str){
        final int p = 16777619;
        int hash = (int)2166136261L;
        for (int i = 0; i < str.length(); i++)
            hash = (hash ^ str.charAt(i)) * p;
        hash += hash << 13;
        hash ^= hash >> 7;
        hash += hash << 3;
        hash ^= hash >> 17;
        hash += hash << 5;
        // 如果算出来的值为负数则取其绝对值
        if (hash < 0)
            hash = Math.abs(hash);
        return hash;
    }

    //标准差计算
     private static double standardDiviation(Double[] x) {
         int m=x.length;
         double sum=0;
         for(int i=0;i<m;i++){//求和
             sum+=x[i];
         }
         double dAve=sum/m;//求平均值
         double dVar=0;
         for(int i=0;i<m;i++){//求方差
             dVar+=(x[i]-dAve)*(x[i]-dAve);
         }
         return Math.sqrt(dVar/m);
     }

    private static class Node {
        private String ip;
        private Map<String, Object> data;

        public Node (String ip) {
            this.ip = ip;
            this.data = new HashMap<>();
        }

        public String getIp() {
            return ip;
        }

        public Map<String, Object> getData() {
            return data;
        }

        public <T> void put(String key, T value) {
            data.put(key, value);
        }

        public void remove(String key){
            data.remove(key);
        }

        public <T> T get(String key) {
            return (T) data.get(key);
        }
    }

    private static abstract class Cluster {
        protected List<Node> nodes;

        public Cluster() {
            this.nodes = new ArrayList<>();
        }

        public abstract void addNode(Node node);
        public abstract void removeNode(Node node);
        public abstract Node get(String key);
    }

    private static class ConsistencyHashCluster extends Cluster {

        private SortedMap<Long, Node> virNodes = new TreeMap<Long, Node>();
        private static final String SPLIT = "#";
        private int vir_node_count = 150;
        public ConsistencyHashCluster(int vir_node_count) {
            super();
            this.vir_node_count = vir_node_count;
        }

        @Override
        public void addNode(Node node) {
            this.nodes.add(node);
            IntStream.range(0, this.vir_node_count)
                    .forEach(index -> {
                        long hash = hash(node.getIp() + SPLIT + index);
                        virNodes.put(hash, node);
                    });
        }

        @Override
        public void removeNode(Node node) {
            nodes.removeIf(o -> node.getIp().equals(o.getIp()));
            IntStream.range(0, this.vir_node_count)
                    .forEach(index -> {
                        long hash = hash(node.getIp() + SPLIT + index);
                        virNodes.remove(hash);
                    });
        }

        @Override
        public Node get(String key) {
            long hash = hash(key);
            SortedMap<Long, Node> subMap = hash >= virNodes.lastKey() ? virNodes.tailMap(0L) : virNodes.tailMap(hash);
            if (subMap.isEmpty()) {
                return null;
            }
            return subMap.get(subMap.firstKey());
        }
    }

    public static void main(String[] args) {
        // 10台服务器，每台服务器150个虚拟节点
        Cluster cluster = new ConsistencyHashCluster(150);
        cluster.addNode(new Node("192.168.0.1"));
        cluster.addNode(new Node("192.168.0.2"));
        cluster.addNode(new Node("192.168.0.3"));
        cluster.addNode(new Node("192.168.0.4"));
        cluster.addNode(new Node("192.168.0.5"));
        cluster.addNode(new Node("192.168.0.6"));
        cluster.addNode(new Node("192.168.0.7"));
        cluster.addNode(new Node("192.168.0.8"));
        cluster.addNode(new Node("192.168.0.9"));
        cluster.addNode(new Node("192.168.0.10"));

        // 100万个KV数据
        IntStream.range(0, DATA_COUNT)
                .forEach(index -> {
                    // 获取对应存储的服务器节点
                    Node node = cluster.get(PRE_KEY + index);
                    node.put(PRE_KEY + index, "Test Data");
                });

        //打印服务器数据分布，并计算分布数量的标准差，评估算法的存储负载不均衡性
        System.out.println("数据分布情况：");
        List<Double> countList = new ArrayList<>();
        cluster.nodes.forEach(node -> {
            System.out.println("IP:" + node.getIp() + ",数据量:" + node.getData().size());
            countList.add(Double.valueOf(node.getData().size()));
        });
        Double[] counts = new Double[countList.size()];
        countList.toArray(counts);
        double standardDiviation = standardDiviation(counts);
        System.out.println("分布数量的标准差：" + String.format("%.1f",standardDiviation) );
    }
}

虚拟节点数   标准差
   10     26427.6  
   50     13478.5   
  100     11394.0
  110      9832.1
  120      7671.1
  130      7209.0
  140      6235.3
  150      6853.5
  200      6979.5 
  300      6746.8
  400      4913.2
  500      5457.0
  600      4947.0
  700      3905.7
  800      3992.4
  900      3427.2
 1000      3378.2