第 5 周命题作业 - 实现一致性 HASH

 关注
发布于: 2020 年 07 月 08 日
1 用你熟悉的语言实现一致性HASH算法存储节点
此处并没有真正把值存储越来，只是统计一下命中次数
package cc.dawn.homework.nodehash;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
/**
 * 存储节点
 */
public class StoreNode {
    /**
     * 模拟存储量
     */
    private AtomicInteger count = new AtomicInteger(0);
    /**
     * 物理节点名称
     */
    private String hostName;
    /**
     * 键值存储数据集
     */
    private Map<Object, Object> kv = new HashMap<>();
    public StoreNode(String hostName) {
        this.hostName = hostName;
    }
    public void put(Object key, Object value) {
        count.incrementAndGet();
    }
    public String getHostName() {
        return hostName;
    }
    public void setHostName(String hostName) {
        this.hostName = hostName;
    }
    public Map<Object, Object> getKv() {
        return kv;
    }
    public void setKv(Map<Object, Object> kv) {
        this.kv = kv;
    }
    @Override
    public String toString() {
        return "StoreNode{" +
                "count=" + count +
                ", hostName='" + hostName + '\'' +
                '}';
    }
    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
        if (o == null || getClass() != o.getClass()) return false;
        StoreNode storeNode = (StoreNode) o;
        return hostName != null ? hostName.equals(storeNode.hostName) : storeNode.hostName == null;
    }
    @Override
    public int hashCode() {
        return hostName != null ? hostName.hashCode() : 0;
    }
    public AtomicInteger getCount() {
        return count;
    }
    public void setCount(AtomicInteger count) {
        this.count = count;
    }
}
﻿
虚拟节点
package cc.dawn.homework.nodehash;
public class VirtualNode {
    /**
     * 序号，通过每个物理节点对应多个虚拟节点
     */
    private int index;
    /**
     * 对应物理节点
     */
    private StoreNode storeNode;
    public VirtualNode(int index, StoreNode storeNode) {
        this.index = index;
        this.storeNode = storeNode;
    }
    public String getFullName() {
        return String.format("%s-%03d", storeNode.getHostName(), index);
    }
    public int getIndex() {
        return index;
    }
    public void setIndex(int index) {
        this.index = index;
    }
    @Override
    public String toString() {
        return "StoreNode{" +
                "index=" + index +
                ", node=" + storeNode +
                '}';
    }
    public StoreNode getStoreNode() {
        return storeNode;
    }
    public void setStoreNode(StoreNode storeNode) {
        this.storeNode = storeNode;
    }
}
﻿
一致性Hash实现类
package cc.dawn.homework.nodehash;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
/**
 * 一致性HASH实现类
 */
public class ConsistentHash {
    public static long MAX_UINT_32 = 0xffffffffL;
    public static int VNODE_COUNT = 100;
    public static int OFFSET = (int) (MAX_UINT_32 / VNODE_COUNT);
    private static Encrypt encrypt = new Encrypt();
    /**
     * 物理节点列表
     */
    private List<StoreNode> node = new ArrayList<>();
    /**
     * HashCode映射虚拟节点集合
     */
    private TreeMap<Integer, VirtualNode> nodeTreeMap = new TreeMap<>();
    /**
     * 根据key的hashCode找到对应的存储节点进行数据写入
     * @param key
     * @param value
     */
    public void put(Object key, Object value) {
        Map.Entry<Integer, VirtualNode> entry = getVirtualNodeEntry(key);
        entry.getValue().getStoreNode().put(key, value);
    }
    private Map.Entry<Integer, VirtualNode> getVirtualNodeEntry(Object key) {
        Map.Entry<Integer, VirtualNode> entry = nodeTreeMap.ceilingEntry(key.hashCode());
        if (entry == null) {
            entry = nodeTreeMap.firstEntry();
        }
        return entry;
    }
    public Object get(Object key) {
        Map.Entry<Integer, VirtualNode> entry = getVirtualNodeEntry(key);
        return entry.getValue().getStoreNode().getKv().get(key);
    }
    public synchronized void addNode(String hostName) {
        StoreNode storeNode = new StoreNode(hostName);
        if (node.contains(storeNode)) {
            return;
        }
//        int start = Utils.hashCode(hostName);
        int start = encrypt.MD5(hostName).hashCode();
        System.out.println("StoreNode:" + storeNode.getHostName() + ", startHS:" + start);
        for (int i = 0; i < VNODE_COUNT; i++) {
            nodeTreeMap.put((start + i * OFFSET), new VirtualNode(i, storeNode));
        }
        this.node.add(storeNode);
    }
    public synchronized void init(String[] ips) {
        int startOffset = (int) (MAX_UINT_32 / ips.length);
        for (int j = 0; j < ips.length; j++) {
            StoreNode storeNode = new StoreNode(ips[j]);
            if (node.contains(storeNode)) {
                return;
            }
            int start = j * startOffset;
            System.out.println("StoreNode:" + storeNode.getHostName() + ", startHS:" + start);
            for (int i = 0; i < VNODE_COUNT; i++) {
                nodeTreeMap.put((start + i * OFFSET), new VirtualNode(i, storeNode));
            }
            this.node.add(storeNode);
        }
//        int start = Utils.hashCode(hostName);
    }
    public List<StoreNode> getNode() {
        return node;
    }
    public void setNode(List<StoreNode> node) {
        this.node = node;
    }
    public TreeMap<Integer, VirtualNode> getNodeTreeMap() {
        return nodeTreeMap;
    }
    public void setNodeTreeMap(TreeMap<Integer, VirtualNode> nodeTreeMap) {
        this.nodeTreeMap = nodeTreeMap;
    }
    @Override
    public String toString() {
        return "ConsistentHash{" +
                "node=" + node +
                ", nodeTreeMap=" + nodeTreeMap +
                '}';
    }
}
对外接口
package cc.dawn.homework.nodehash;
public class CacheRoute implements ICacheRoute {
    private ConsistentHash consistentHash = new ConsistentHash();
    @Override
    public void put(Object key, Object value) {
        consistentHash.put(key, value);
    }
    @Override
    public Object get(Object key) {
        return consistentHash.get(key);
    }
    @Override
    public void clear(Object key) {
    }
    @Override
    public void init(String[] hostNames) {
        this.consistentHash.init(hostNames);
    }
    public ConsistentHash getConsistentHash() {
        return consistentHash;
    }
}
2 用例测试如果有100万KV数据写入10台服务器，测试计算数据分布数量的标准差
﻿
测试类
package cc.dawn.homework.nodehash;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.stream.Collectors;
public class TestConsistentHash {
    private static final int MAX_ORDER_PER_NODE = 10000;
    private static int OFFSET = (int) (ConsistentHash.MAX_UINT_32 / ConsistentHash.VNODE_COUNT);
    public static void main(String[] args) {
        CacheRoute cacheRoute = new CacheRoute();
        cacheRoute.init(new String[]{"192.168.0.1", "192.168.0.2", "192.168.0.3", "192.168.0.4", "192.168.0.5", "192.168.0.6",
                "192.168.0.7", "192.168.0.8", "192.168.0.9", "192.168.0.10"});
        CountDownLatch countDownLatch = new CountDownLatch(100);
        for (int i = 0; i < 100; i++) {
            int finalI = i;
            new Thread(() -> {
                int count = 0;
                while (count++ < MAX_ORDER_PER_NODE) {
                    cacheRoute.put(UUID.randomUUID(), null);
                    if (count % 100 == 0) {
                        System.out.println(String.format("thread-%d----count:%d", finalI, count));
                    }
                }
                countDownLatch.countDown();
            }).start();
        }
        try {
            countDownLatch.await();
            ConsistentHash consistentHash = cacheRoute.getConsistentHash();
            System.out.println(consistentHash.getNode().toString());
            List<Integer> hitCountList = consistentHash.getNode().stream().map(node -> node.getCount().intValue()).collect(Collectors.toList());
            Integer[] array = hitCountList.toArray(new Integer[]{});
            double standardDeviation = Utils.standardDeviation(array);
            System.out.println("标准差:" + standardDeviation);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
﻿
打印截图
﻿
发布于: 2020 年 07 月 08 日阅读数: 38
Dawn

 关注
还未添加个人签名 2018.07.04 加入
还未添加个人简介
发布
暂无评论