WeightRoundRobinLoadBalancer extends AbstractLoadBalancer { @Override public ProviderInfo doSelect(SofaRequest request, List<ProviderInfo> providerInfos) { String key = getServiceKey(request); int length = providerInfos.size(); int maxWeight = 0; int minWeight = Integer.MAX_VALUE; final LinkedHashMap<ProviderInfo, IntegerWrapper> invokerToWeightMap = new LinkedHashMap<ProviderInfo, IntegerWrapper>(); int weightSum = 0; for (ProviderInfo providerInfo : providerInfos) { int weight = getWeight(providerInfo); maxWeight = Math.max(maxWeight, weight); minWeight = Math.min(minWeight, weight); if (weight > 0) { invokerToWeightMap.put(providerInfo, new IntegerWrapper(weight)); weightSum += weight; } } PositiveAtomicCounter sequence = sequences.get(key); if (sequence == null) { sequences.putIfAbsent(key, new PositiveAtomicCounter()); sequence = sequences.get(key); } int currentSequence = sequence.getAndIncrement(); if (maxWeight > 0 && minWeight < maxWeight) { int mod = currentSequence % weightSum; for (int i = 0; i < maxWeight; i++) { for (Map.Entry<ProviderInfo, IntegerWrapper> each : invokerToWeightMap.entrySet()) { final ProviderInfo k = each.getKey(); final IntegerWrapper v = each.getValue(); if (mod == 0 && v.getValue() > 0) { return k; } if (v.getValue() > 0) { v.decrement(); mod--; } } } } return providerInfos.get(currentSequence % length); } WeightRoundRobinLoadBalancer(ConsumerBootstrap consumerBootstrap); @Override ProviderInfo doSelect(SofaRequest request, List<ProviderInfo> providerInfos);  }
@Test public void doSelect() throws Exception { WeightRoundRobinLoadBalancer loadBalancer = new WeightRoundRobinLoadBalancer(null); Map<Integer, Integer> cnt = new HashMap<Integer, Integer>(); int size = 20; int total = 19000; SofaRequest request = new SofaRequest(); { for (int i = 0; i < size; i++) { cnt.put(9000 + i, 0); } List<ProviderInfo> providers = buildSameWeightProviderList(size); long start = System.currentTimeMillis(); for (int i = 0; i < total; i++) { ProviderInfo provider = loadBalancer.doSelect(request, providers); int port = provider.getPort(); cnt.put(port, cnt.get(port) + 1); } long end = System.currentTimeMillis(); LOGGER.info("elapsed" + (end - start) + "ms"); LOGGER.info("avg " + (end - start) * 1000 * 1000 / total + "ns"); int avg = total / size; for (int i = 0; i < size; i++) { Assert.assertTrue(avg == cnt.get(9000 + i)); } } { for (int i = 0; i < size; i++) { cnt.put(9000 + i, 0); } List<ProviderInfo> providers = buildDiffWeightProviderList(size); long start = System.currentTimeMillis(); for (int i = 0; i < total; i++) { ProviderInfo provider = loadBalancer.doSelect(request, providers); int port = provider.getPort(); cnt.put(port, cnt.get(port) + 1); } long end = System.currentTimeMillis(); LOGGER.info("elapsed" + (end - start) + "ms"); LOGGER.info("avg " + (end - start) * 1000 * 1000 / total + "ns"); Assert.assertTrue(cnt.get(9000) == 0); int count = 0; int sum = 0; for (int i = 0; i < size; i++) { count += i; sum += cnt.get(9000 + i); } Assert.assertTrue(sum == total); int per = total / count; for (int i = 1; i < size; i++) { Assert.assertTrue(per * i == cnt.get(9000 + i)); } } }