[享学Netflix] 四十七、Ribbon多区域选择

LoadBalancerStats：

	volatile Map<String, List<? extends Server>> upServerListZoneMap = new ConcurrentHashMap<>();
    public Set<String> getAvailableZones() {
        return upServerListZoneMap.keySet();
    }

ZoneAvoidanceRule：

    static String randomChooseZone(
    		Map<String, ZoneSnapshot> snapshot,
            Set<String> chooseFrom) {
        if (chooseFrom == null || chooseFrom.size() == 0) {
            return null;
        }
		// 注意：默认选择的是第一个zone区域
		// 若总共就1个区域，那就是它了。若有多个，那就需要随机去选
		String selectedZone = chooseFrom.iterator().next();
        if (chooseFrom.size() == 1) {
        	return chooseFrom.iterator().next();
        }

		// 所有的区域中总的Server实例数
        int totalServerCount = 0;
        for (String zone : chooseFrom) {
            totalServerCount += snapshot.get(zone).getInstanceCount();
        }
        // 从所有的实例总数中随机选个数字。比如总数是10台机器
        // 那就是从[1-10]之间随机选个数字，比如选中为6
        int index = random.nextInt(totalServerCount) + 1;
		
		// sum代表当前实例统计的总数
		// 它的逻辑是：当sum超过这个index时，就以这个区域为准
        int sum = 0;
        for (String zone : chooseFrom) {
            sum += snapshot.get(zone).getInstanceCount();
            if (index <= sum) {
                selectedZone = zone;
                break;
            }
        }
	
	}

ZoneAvoidanceRule：

	// snapshot：zone对应的ZoneSnapshot的一个map
	// triggeringLoad：
	// triggeringBlackoutPercentage：
	public static Set<String> getAvailableZones(
			Map<String, ZoneSnapshot> snapshot,
			double triggeringLoad,
			double triggeringBlackoutPercentage) {
		// 为毛一个都木有不返回空集合？？？有点乱啊。。。。不过没关系
        if (snapshot.isEmpty()) {
            return null;
        }
		
		//最终需要return的可用区，中途会进行排除的逻辑
		Set<String> availableZones = new HashSet<>(snapshot.keySet());
		// 如果有且仅有一个zone可用，再糟糕也得用，不用进行其他逻辑了
        if (availableZones.size() == 1) {
            return availableZones;
        }
		
		// 记录很糟糕
		Set<String> worstZones = new HashSet<>();
		// 所有zone中，平均负载最高值
		double maxLoadPerServer = 0;
		// true：zone有限可用
		// false：zone全部可用
		boolean limitedZoneAvailability = false;

		// 对每个zone的情况逐一分析
		for (Map.Entry<String, ZoneSnapshot> zoneEntry : snapshot.entrySet()) {
            String zone = zoneEntry.getKey();
            ZoneSnapshot zoneSnapshot = zoneEntry.getValue();
            int instanceCount = zoneSnapshot.getInstanceCount();
            
            // 若该zone内一个实例都木有了，那就是完全不可用，那就移除该zone
            // 然后标记zone是有限可用的（并非全部可用喽）
            if (instanceCount == 0) {
                availableZones.remove(zone);
                limitedZoneAvailability = true;
            } else {
            	// 该zone的平均负载
				double loadPerServer = zoneSnapshot.getLoadPerServer();
				
				// 机器的熔断总数 / 总实例数已经超过了阈值（默认为1，也就是全部熔断才会认为该zone完全不可用）
				// 或者 loadPerServer < 0 （啥时候小于0？？？下面说）
                if (((double) zoneSnapshot.getCircuitTrippedCount()) / instanceCount >= triggeringBlackoutPercentage
                        || loadPerServer < 0) {
					// 证明这个zone完全不可用，就移除掉
                    availableZones.remove(zone);
                    limitedZoneAvailability = true;
				} else { // 并不是完全不可用，就看看状态是不是很糟糕

					// 若当前负载和最大负载相当，那认为已经很糟糕了
					if (Math.abs(loadPerServer - maxLoadPerServer) < 0.000001d) {
						worstZones.add(zone);
					
					// 或者若当前负载大于最大负载了
					} else if (loadPerServer > maxLoadPerServer) {
						maxLoadPerServer = loadPerServer;
						worstZones.clear();
						worstZones.add(zone);
					}
				}
            }
		}

		// 若最大负载小于设定的负载阈值 并且limitedZoneAvailability=false
		// 就是说全部zone都可用，并且最大负载都还没有达到阈值，那就把全部zone返回
        if (maxLoadPerServer < triggeringLoad && !limitedZoneAvailability) {
            // zone override is not needed here
            return availableZones;
        }
        String zoneToAvoid = randomChooseZone(snapshot, worstZones);
        if (zoneToAvoid != null) {
            availableZones.remove(zoneToAvoid);
        }
        return availableZones;
	}

	if (Math.abs(loadPerServer - maxLoadPerServer) < 0.000001d) {
	    // they are the same considering double calculation
	    // round error
	    worstZones.add(zone);
	} else if (loadPerServer > maxLoadPerServer) {
	    maxLoadPerServer = loadPerServer;
	    worstZones.clear();
	    worstZones.add(zone);
	}

ZoneAvoidanceRule：

	// 实际调用仍旧为getAvailableZones方法~
	// 它友好的只需要传参LoadBalancerStats即可，内部帮你构建snapshot这个Map
    public static Set<String> getAvailableZones(LoadBalancerStats lbStats,
            double triggeringLoad, double triggeringBlackoutPercentage) {
        if (lbStats == null) {
            return null;
        }
        Map<String, ZoneSnapshot> snapshot = createSnapshot(lbStats);
        return getAvailableZones(snapshot, triggeringLoad,
                triggeringBlackoutPercentage);
    }

// 单独线程模拟刷页面，获取监控到的数据
private void monitor(LoadBalancerStats lbs) {
    List<String> zones = Arrays.asList("华南", "华东", "华北");
    new Thread(() -> {
        ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1);
        executorService.scheduleWithFixedDelay(() -> {
            // 打印当前可用区
            // 获取可用区
            Set<String> availableZones = ZoneAvoidanceRule.getAvailableZones(lbs, 0.2d, 0.99999d);
            System.out.println("=====当前可用区为：" + availableZones);

            zones.forEach(zone -> {
                System.out.printf("区域[" + zone + "]概要：");
                int instanceCount = lbs.getInstanceCount(zone);
                int activeRequestsCount = lbs.getActiveRequestsCount(zone);
                double activeRequestsPerServer = lbs.getActiveRequestsPerServer(zone);
                // ZoneSnapshot zoneSnapshot = lbs.getZoneSnapshot(zone);

                System.out.printf("实例总数：%s，活跃请求总数：%s，平均负载：%s\n", instanceCount, activeRequestsCount, activeRequestsPerServer);
                // System.out.println(zoneSnapshot);
            });
            System.out.println("======================================================");
        }, 5, 5, TimeUnit.SECONDS);
    }).start();
}


// 请注意：请必须保证Server的id不一样，否则放不进去List的（因为Server的equals hashCode方法仅和id有关）
// 所以此处使用index作为port，以示区分
private Server createServer(String zone, int index) {
    Server server = new Server("www.baidu" + zone + ".com", index);
    server.setZone(zone);
    return server;
}


// 多线程，模拟请求
private void request(ServerStats serverStats) {
    new Thread(() -> {
        // 每10ms发送一个请求（每个请求处理10-200ms的时间），持续不断
        ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1);
        executorService.scheduleWithFixedDelay(() -> {
            new Thread(() -> {
                // 请求之前 记录活跃请求数
                serverStats.incrementActiveRequestsCount();
                serverStats.incrementNumRequests();
                long rt = doSomething();
                // 请求结束， 记录响应耗时
                serverStats.noteResponseTime(rt);
                serverStats.decrementActiveRequestsCount();
            }).start();
        }, 10, 10, TimeUnit.MILLISECONDS);
    }).start();
}

// 模拟请求耗时，返回耗时时间
private long doSomething() {
    try {
        int rt = randomValue(10, 200);
        TimeUnit.MILLISECONDS.sleep(rt);
        return rt;
    } catch (InterruptedException e) {
        e.printStackTrace();
        return 0L;
    }
}

// 本地使用随机数模拟数据收集
private int randomValue(int min, int max) {
    return min + (int) (Math.random() * ((max - min) + 1));
}

// ============单元测试
@Test
public void fun6() throws InterruptedException {
    LoadBalancerStats lbs = new LoadBalancerStats("YoutBatman");

    // 添加Server
    List<Server> serverList = new ArrayList<>();
    serverList.add(createServer("华南", 1));
    serverList.add(createServer("华东", 1));
    serverList.add(createServer("华东", 2));

    serverList.add(createServer("华北", 1));
    serverList.add(createServer("华北", 2));
    serverList.add(createServer("华北", 3));
    serverList.add(createServer("华北", 4));
    lbs.updateServerList(serverList);

    Map<String, List<Server>> zoneServerMap = new HashMap<>();
    // 模拟向每个Server发送请求  记录ServerStatus数据
    serverList.forEach(server -> {
        ServerStats serverStat = lbs.getSingleServerStat(server);
        request(serverStat);

        // 顺便按照zone分组
        String zone = server.getZone();
        if (zoneServerMap.containsKey(zone)) {
            zoneServerMap.get(zone).add(server);
        } else {
            List<Server> servers = new ArrayList<>();
            servers.add(server);
            zoneServerMap.put(zone, servers);
        }
    });
    lbs.updateZoneServerMapping(zoneServerMap);

    // 从lbs里拿到一些监控数据
    monitor(lbs);

    TimeUnit.SECONDS.sleep(500);
}

=====当前可用区为：[华南, 华东]
区域[华南]概要：实例总数：1，活跃请求总数：10，平均负载：10.0
区域[华东]概要：实例总数：2，活跃请求总数：18，平均负载：9.0
区域[华北]概要：实例总数：4，活跃请求总数：41，平均负载：10.25
======================================================
=====当前可用区为：[华南, 华北]
区域[华南]概要：实例总数：1，活跃请求总数：9，平均负载：9.0
区域[华东]概要：实例总数：2，活跃请求总数：22，平均负载：11.0
区域[华北]概要：实例总数：4，活跃请求总数：34，平均负载：8.5
======================================================
=====当前可用区为：[华南, 华东]
区域[华南]概要：实例总数：1，活跃请求总数：9，平均负载：9.0
区域[华东]概要：实例总数：2，活跃请求总数：18，平均负载：9.0
区域[华北]概要：实例总数：4，活跃请求总数：37，平均负载：9.25
======================================================
=====当前可用区为：[华北, 华东]
区域[华南]概要：实例总数：1，活跃请求总数：10，平均负载：10.0
区域[华东]概要：实例总数：2，活跃请求总数：17，平均负载：8.5
区域[华北]概要：实例总数：4，活跃请求总数：39，平均负载：9.75
======================================================
...