Week5 作业

一致性hash算法以及100万KV负载均衡性标准差测试代码

public class Main {



private static final long OFFSET_BASIS = 2166136261L;// 32位offset basis



private static final long PRIME = 0x01000193; // 32位prime



//服务器节点列表

private static final List<String> nodeList = Arrays.asList(

"192.168.1.1",

"192.168.1.2",

"192.168.1.3",

"192.168.1.4",

"192.168.1.5",

"192.168.1.6",

"192.168.1.7",

"192.168.1.8",

"192.168.1.9",

"192.168.1.10");

//一致性hash环

private static final TreeMap<Long, Integer> codeMapper = new TreeMap<>();



//统计节点服务器访问次数

private static final Map<String, Integer> nodeAccessCountMapper = new HashMap<>();



//初始化虚拟节点

public static void initVirtualNode(final int virtualNodeCount) {



nodeList.forEach(node -> {



for (long index = 0; index < virtualNodeCount; ++index) {



long code = localHashCode(String.format("%s#%d", node, index));

codeMapper.put(code, nodeList.indexOf(node));

}

});

}



//hashcode生成算法

public static int localHashCode(final String source) {



char[] bytes = source.toCharArray();

int hash = 0;

for (char b : bytes) {

hash ^= b;

hash += (hash << 1) + (hash << 4) + (hash << 7) + (hash << 8) + (hash << 24);

}

// System.out.println(hash);

return hash;

}



//查找节点

public static int findNode(final String key) {



long keyCode = localHashCode(key);

SortedMap<Long, Integer> resMapper = codeMapper.tailMap(keyCode);



if (null == resMapper || resMapper.isEmpty()) {



return codeMapper.firstEntry().getValue();

} else {



return codeMapper.get(resMapper.firstKey());

}

}



//计算标准差

public static double calSD(List<Integer> deltaList) {



int count = deltaList.size();

//求和

double sum = deltaList.stream().mapToInt(Integer::intValue).sum();



//求平均值

double dVar = 0;

double dAve = sum / count;

//求方差

for (int index = 0; index < count; index++) {



dVar += (deltaList.get(index) - dAve) * (deltaList.get(index) - dAve);

}

return Math.sqrt(dVar / count);

}



//测试100万KV数据，在不同节点数量时负载标准差

public static void main(String[] args) throws ExecutionException, InterruptedException {



calAccessSD(10);

calAccessSD(50);

calAccessSD(100);

calAccessSD(150);

calAccessSD(200);

calAccessSD(300);

calAccessSD(400);

calAccessSD(600);

calAccessSD(800);

calAccessSD(1000);

calAccessSD(1500);

calAccessSD(2000);

calAccessSD(5000);

}



//计算负载分布标准差

public static void calAccessSD(final int virtualNodeCount) throws InterruptedException, ExecutionException {



LocalDateTime begin = LocalDateTime.now();



//初始化虚拟节点

initVirtualNode(virtualNodeCount);

//开启20个线程模拟2000000(KV)

ExecutorService service = Executors.newFixedThreadPool(20);

CompletionService<Long> completionService = new ExecutorCompletionService<>(service);

for (int threadIndex = 0; threadIndex < 20; ++threadIndex) {



completionService.submit(new Task());

}



for (int threadIndex = 0; threadIndex < 20; ++threadIndex) {



completionService.take();

}



System.out.println(

String.format("%6d个虚拟节点,200万KV负载分布-标准差:%8.0f,-查找节点耗时:%s(ms)",

vitualNodeCount,

calSD(nodeAccessCountMapper.values().stream().collect(Collectors.toList())),

Duration.between(begin, LocalDateTime.now()).toMillis()

)

);

nodeAccessCountMapper.clear();

}



//100000个KV查询任务,并统计每个节点访问次数

static class Task implements Callable<Long> {



@Override

public Long call() throws Exception {



LocalDateTime begin = LocalDateTime.now();

List<Integer> deltaList = new ArrayList<>(Collections.nCopies(100000, 1));

deltaList.stream()

.forEachOrdered(item -> {



int findId = findNode(String.format("%s", UUID.randomUUID().toString()));

synchronized (nodeAccessCountMapper) {



String id = String.format("%d", findId);

if (nodeAccessCountMapper.containsKey(id)) {



int count = nodeAccessCountMapper.get(id);

nodeAccessCountMapper.put(id, count + 1);

} else {



nodeAccessCountMapper.put(id, 1);

}

}

});

return Duration.between(begin, LocalDateTime.now()).toMillis();

}

}

}



代码运行的结果是

10个虚拟节点,200万KV负载分布-标准差: 115626,-查找节点耗时:6417(ms)

50个虚拟节点,200万KV负载分布-标准差: 79339,-查找节点耗时:3545(ms)

100个虚拟节点,200万KV负载分布-标准差: 73707,-查找节点耗时:5023(ms)

150个虚拟节点,200万KV负载分布-标准差: 70132,-查找节点耗时:5053(ms)

200个虚拟节点,200万KV负载分布-标准差: 56498,-查找节点耗时:5675(ms)

300个虚拟节点,200万KV负载分布-标准差: 44061,-查找节点耗时:5195(ms)

400个虚拟节点,200万KV负载分布-标准差: 39863,-查找节点耗时:5231(ms)

600个虚拟节点,200万KV负载分布-标准差: 32664,-查找节点耗时:3372(ms)

800个虚拟节点,200万KV负载分布-标准差: 20455,-查找节点耗时:3198(ms)

1000个虚拟节点,200万KV负载分布-标准差: 15550,-查找节点耗时:3185(ms)

1500个虚拟节点,200万KV负载分布-标准差: 15606,-查找节点耗时:3198(ms)

2000个虚拟节点,200万KV负载分布-标准差: 15759,-查找节点耗时:3272(ms)

5000个虚拟节点,200万KV负载分布-标准差: 11406,-查找节点耗时:3311(ms)

总结下来我是不对的，请老师指正。