public class NacosClient {
    //nacos監(jiān)聽器,處理初始化hash環(huán)等邏輯
    private Nodelistener nodelistener;
    //初始化nacosClient,并且監(jiān)聽服務(wù)
    public void init() {
        NamingService naming = null;
        try {
            System.out.println(System.getProperty("serveAddr"));
            naming = NamingFactory.createNamingService(System.getProperty("serveAddr"));
            //注冊監(jiān)聽器,當(dāng)集群節(jié)點(diǎn)變化的時(shí)候調(diào)用nodelistener處理節(jié)點(diǎn)信息
            naming.subscribe("test", event -> {
                if (event instanceof NamingEvent) {
                    nodelistener.handlerChange((NamingEvent)event);
                }
            });
        } catch (NacosException e) {
            e.printStackTrace();
        }
    }
}

public class Nodelistener {
    private List<Instance> servers;
    //利用treeMap構(gòu)建hash環(huán)
    private volatile SortedMap<Long, Instance> sortedMap = new TreeMap<Long, Instance>();
    //虛擬節(jié)點(diǎn)
    private int virtualNodeCount = 100;
    public synchronized void handlerChange(NamingEvent event) {
            List<Instance> servers = new ArrayList<Instance>();
            event.getInstances().stream().filter(instance -> {
                return instance.isEnabled() && instance.isHealthy();
            }).forEach(instance -> servers.add(instance));
            this.onChange(servers);
    }
    //每次集群節(jié)點(diǎn)變化時(shí),重新構(gòu)建hash環(huán)
    public void onChange(List<Instance> servers) {
        //只有一個(gè)節(jié)點(diǎn)的時(shí)候這里暫不考慮,讀者可以自行處理
        if(servers.size() != 1) {
            SortedMap<Long, Instance> newSortedMap = new TreeMap<Long, Instance>();
            for (int i = 0; i < servers.size(); i ++) {
                for (int j = 0; j < this.virtualNodeCount; j++) {
                    //計(jì)算虛擬節(jié)點(diǎn)的hash,這里用到的是MurMurHash,網(wǎng)上還有很多其他hash實(shí)現(xiàn),
                    //有興趣可以自行查閱，具體實(shí)現(xiàn)細(xì)節(jié)就不列出了
                    Long hash = HashUtil.getHash(servers.get(i).getIp() + ":" + 
                                                                     servers.get(i).getPort() + j);
                    //把虛擬節(jié)點(diǎn)加入hash環(huán)
                    sortedMap.put(hash, servers.get(i));
                }
            }
            sortedMap = newSortedMap;
        }
        this.servers = servers;
    }
    //根據(jù)傳入的key獲取hash環(huán)上順時(shí)針到hash環(huán)尾部部分所有節(jié)點(diǎn)
    public Instance getInstance(String str) {
        Long hash = HashUtil.getHash(str);
        SortedMap<Long, Instance> map = sortedMap.tailMap(hash);
        //這里證明剛好獲取的是尾部，所以返回所有的節(jié)點(diǎn)，其實(shí)是獲取第一個(gè)節(jié)點(diǎn)
        if (map.isEmpty()) {
            map = sortedMap;
        }
        return map.get(map.firstKey());
    }
}

public class LoadBalance {
    private Nodelistener nodelistener;
    //只需要簡單的從hash環(huán)中獲取第一個(gè)節(jié)點(diǎn)
    public Instance doSelect(String key) {
        return nodelistener.getInstance(key);
    }
}

    public void test2() {
        Map<String, Integer> map = new HashMap<String, Integer>();
        Random random = new Random();
        for (int i = 0; i < 10000; i ++) {
            String key = String.valueOf(random.nextLong());
            Instance instance = loadBalance.doSelect(key);
            if(!map.containsKey(instance.getIp())) {
                map.put(instance.getIp(), 0);
            }else {
                map.put(instance.getIp(), map.get(instance.getIp()) + 1);
            }
            System.out.println("test2 count :" + i);
            System.out.printf("select IP is :" + instance.getIp());
        }
        System.out.println(map.toString());
    }

    public void test(String key) throws InterruptedException {
        //這里可以設(shè)置重試次數(shù)
        for (;;) {
            Instance instance = loadBalance.doSelect(key);
            String addr = instance.getIp() + ":" + instance.getPort();
            //測試請求
            if(post(addr)) {
                //成功邏輯
                .....
                break;
            }else {
                //等待兩秒,即可以使文件服務(wù)器不同節(jié)點(diǎn)之間同步文件，還可以等待更新本地hash環(huán)
                Thread.sleep(2000);
                //失敗則選取下一個(gè)節(jié)點(diǎn)
                instance = loadBalance.doSelect(key);
                //此處可以增加重試次數(shù)邏輯和如果重試到hash環(huán)上最后一個(gè)節(jié)點(diǎn)則重新獲取hash環(huán)第一個(gè)節(jié)點(diǎn)邏輯，
                // 在此就不做論述，讀者可以自由發(fā)揮
                continue;
            }
        }
    }

    public void onChange(List<Instance> newServers) {
        //單節(jié)點(diǎn)時(shí)這里暫不考慮
        if(servers.size() != 1 ) {
            //TODO ..
        }
        Map<String, Instance> oldAddrs = 
               this.servers.stream()
                           .collect(Collectors.toMap(Instance::toInetAddr, instance -> instance));
        Map<String, Instance> newAddrs = 
               newServers.stream()
                           .collect(Collectors.toMap(Instance::toInetAddr, instance -> instance));
        //remove
        oldAddrs.forEach((key, value) -> {
            if (!newAddrs.containsKey(key)) {
                for(int j = 0; j < virtualNodeCount; j++) {
                    Long hash = HashUtil.getHash( value.toInetAddr() + "&&VM"+ j);
                    sortedMap.remove(hash);
                }
            }
        });
        //add
        newAddrs.forEach((key, value) -> {
            if (!oldAddrs.containsKey(key)) {
                for(int j = 0; j < virtualNodeCount; j++) {
                    Long hash = HashUtil.getHash(value.toInetAddr() + "&&VM" + j);
                    sortedMap.put(hash, value);
                }
            }
        });
        this.servers = newServers;
    }