架构师训练营作业 -Week5

1、实现一致性 hash 算法

2、测试 100 万 KV 数据，10 个服务器节点的情况下，KV 数据在服务器上分布数量的标准差

代码如下：

public interface Node {

String getIp();

String getName();

boolean isVirtualNode();

}

public class PhysicalNode implements Node {

private String ip;

public PhysicalNode(String ip) {

this.ip = ip;

}

@Override

public String getIp() {

return ip;

}

@Override

public String getName() {

return ip;

}

@Override

public boolean isVirtualNode() {

return false;

}

}

public class VirtualNode implements Node {

private static final String VIRTUALNODENAME_SUFFIX = "_VN";

private Node physicalNode;

private int virtualNumber;

public VirtualNode(Node physicalNode, int virtualNumber) {

this.physicalNode = physicalNode;

this.virtualNumber = virtualNumber;

}

@Override

public String getIp() {

return physicalNode.getIp();

}

@Override

public String getName() {

return getIp() + VIRTUALNODENAME_SUFFIX + virtualNumber;

}

@Override

public boolean isVirtualNode() {

return true;

}

}

public class ConsistentHashRing {

private List<String> physicalNodeIps;

private int virtualNodeCount;

private HashMap<Node, List<Node>> physicalNodeToVirtualNodesMap;

private SortedMap<Long, Node> ringNodeMap;

public ConsistentHashRing(List<String> physicalNodeIps, int virtualNodeCount) {

this.physicalNodeIps = new ArrayList<>();

this.virtualNodeCount = virtualNodeCount;

this.physicalNodeToVirtualNodesMap = new HashMap<>(physicalNodeIps.size());

this.ringNodeMap = new TreeMap<>();

init(physicalNodeIps);

}

private void init(List<String> physicalNodeIps) {

physicalNodeIps.forEach(physicalNodeIp -> {

addNode(physicalNodeIp);

});

}

public void addNode(String ip) {

if (physicalNodeIps.contains(ip)) {

return;

}

PhysicalNode physicalNode = new PhysicalNode(ip);

List<Node> virtualNodes = new ArrayList<>();

for (int i = 0; i < virtualNodeCount; i++) {

virtualNodes.add(new VirtualNode(physicalNode, i));

}

putVirtualNodesToRing(virtualNodes);

physicalNodeToVirtualNodesMap.put(physicalNode, virtualNodes);

physicalNodeIps.add(ip);

}

public void removeNode(String ip) {

physicalNodeToVirtualNodesMap.keySet().stream()

.filter(physicalNode -> physicalNode.getIp().equals(ip))

.findFirst()

.map(physicalNodeToVirtualNodesMap::get)

.ifPresent(this::removeVirtualNodesFromRing);

}

public String queryServerNodeIp(String key) {

long hashCode = getHashCode(key);

SortedMap<Long, Node> nodeMapWithGreaterThenQueriedHashCode = ringNodeMap.tailMap(hashCode);

Long resultHashCode = nodeMapWithGreaterThenQueriedHashCode.isEmpty() ?

ringNodeMap.firstKey() : nodeMapWithGreaterThenQueriedHashCode.firstKey();

return ringNodeMap.get(resultHashCode).getIp();

}

public List<String> getAllServerIps() {

return physicalNodeToVirtualNodesMap.keySet().stream().map(Node::getIp).collect(Collectors.toList());

}

private void putVirtualNodesToRing(List<Node> virtualNodes) {

virtualNodes.forEach(virtualNode -> {

long hasCode = getHashCode(virtualNode.getName());

ringNodeMap.put(hasCode, virtualNode);

});

}

private void removeVirtualNodesFromRing(List<Node> virtualNodes) {

virtualNodes.forEach(virtualNode -> {

long hasCode = getHashCode(virtualNode.getName());

ringNodeMap.remove(hasCode, virtualNode);

});

}

private long getHashCode(String str) {

return str.hashCode() << 16;

}

}

public class ConsistentHashTest {

public static void main(String[] args) {

List<String> serverIps = Arrays.asList("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");

//计算 100 万 KV 数据，在 10 个服务器节点的情况下，数据在服务器上分布数量的标准差

double standardDistributionStatistics = standardDeviationCalculation(1000000, serverIps.size(), new ConsistentHashRing(serverIps, 200));

System.out.println(standardDistributionStatistics);

}

private static double standardDeviationCalculation(int totalDataNumbers, int serverNumbers, ConsistentHashRing consistentHashRing) {

double expectedDistributionValue = totalDataNumbers / serverNumbers;

Map<String, Integer> distributionStatisticsMap = new HashMap<>();

for (int i = 0; i < totalDataNumbers; i++) {

String serverIp = consistentHashRing.queryServerNodeIp("value" + i);

Integer distributionStatistics = distributionStatisticsMap.getOrDefault(serverIp, 0);

distributionStatisticsMap.put(serverIp, distributionStatistics + 1);

}

double sum = 0.0;

for (String serverIp : consistentHashRing.getAllServerIps()) {

int serverDistributionStatistics = distributionStatisticsMap.getOrDefault(serverIp, 0);

sum += Math.pow(serverDistributionStatistics - expectedDistributionValue, 2);

}

return Math.sqrt(sum / serverNumbers);

}

}