IRule是定义了各种负载均衡，我们分析其中几种
image.png

  1. public abstract class AbstractLoadBalancerRule implements IRule, IClientConfigAware {
  2.     // 负载均衡器
  3.     private ILoadBalancer lb;
  4.     // 省略get/set
  5. }

随机选择RandomRule

  1. public class RandomRule extends AbstractLoadBalancerRule {
  2.     Random rand;
  4.     public RandomRule() {
  5.         rand = new Random();
  6.     }
  7.     @Override
  8.     public Server choose(Object key) {
  9.         return choose(getLoadBalancer(), key);
  10.     }
  11.     public Server choose(ILoadBalancer lb, Object key) {
  12.         if (lb == null) {
  13.             return null;
  14.         }
  15.         Server server = null;
  16.         while (server == null) {
  17.             // 检查中断
  18.             if (Thread.interrupted()) {
  19.                 return null;
  20.             }
  21.             // 可访问的服务
  22.             List<Server> upList = lb.getReachableServers();
  23.             // 所有服务
  24.             List<Server> allList = lb.getAllServers();
  26.             int serverCount = allList.size();
  27.             if (serverCount == 0) {
  28.                 // 没有服务可用
  29.                 return null;
  30.             }
  31.             // 生成随机数
  32.             int index = rand.nextInt(serverCount);
  33.             server = upList.get(index);
  35.             if (server == null) {
  36.                 // 该服务不可访问，重新选择
  37.                 Thread.yield();
  38.                 continue;
  39.             }
  40.             // 可用
  41.             if (server.isAlive()) {
  42.                 return (server);
  43.             }
  44.             // 重新选择
  45.             server = null;
  46.             Thread.yield();
  47.         }
  48.         return server;
  49.     }
  51.     @Override
  52.     public void initWithNiwsConfig(IClientConfig clientConfig) {
  53.         // TODO Auto-generated method stub
  54.     }
  55. }

轮训RoundRobinRule

  1. public class RoundRobinRule extends AbstractLoadBalancerRule {
  3.     private AtomicInteger nextServerCyclicCounter;
  4.     private static final boolean AVAILABLE_ONLY_SERVERS = true;
  5.     private static final boolean ALL_SERVERS = false;
  7.     public RoundRobinRule() {
  8.         // 计数器
  9.         nextServerCyclicCounter = new AtomicInteger(0);
  10.     }
  12.     public Server choose(ILoadBalancer lb, Object key) {
  13.         if (lb == null) {
  14.             return null;
  15.         }
  17.         Server server = null;
  18.         int count = 0;
  19.         while (server == null && count++ < 10) {
  20.             // 
  21.             List<Server> reachableServers = lb.getReachableServers();
  22.             List<Server> allServers = lb.getAllServers();
  23.             int upCount = reachableServers.size();
  24.             int serverCount = allServers.size();
  26.             // 没有可用的服务
  27.             if ((upCount == 0) || (serverCount == 0)) {
  28.                 return null;
  29.             }
  30.             // 
  31.             int nextServerIndex = incrementAndGetModulo(serverCount);
  32.             server = allServers.get(nextServerIndex);
  33.             // 重新选择
  34.             if (server == null) {
  35.                 Thread.yield();
  36.                 continue;
  37.             }
  38.             // 可用服务
  39.             if (server.isAlive() && (server.isReadyToServe())) {
  40.                 return (server);
  41.             }
  43.             // Next.
  44.             server = null;
  45.         }
  46.         // 重试超过10次
  47.         if (count >= 10) {
  48.             log.warn("No available alive servers after 10 tries from load balancer: " + lb);
  49.         }
  50.         return server;
  51.     }
  53.     // 轮训选择
  54.     private int incrementAndGetModulo(int modulo) {
  55.         for (;;) {
  56.             int current = nextServerCyclicCounter.get();
  57.             // 保证原子性
  58.             int next = (current + 1) % modulo;
  59.             if (nextServerCyclicCounter.compareAndSet(current, next))
  60.                 return next;
  61.         }
  62.     }
  64.     @Override
  65.     public Server choose(Object key) {
  66.         return choose(getLoadBalancer(), key);
  67.     }
  69.     @Override
  70.     public void initWithNiwsConfig(IClientConfig clientConfig) {
  71.     }
  72. }

响应实现权重分配WeightedResponseTimeRule

  1. public class WeightedResponseTimeRule extends RoundRobinRule {
  3.     public static final IClientConfigKey<Integer> WEIGHT_TASK_TIMER_INTERVAL_CONFIG_KEY = new IClientConfigKey<Integer>() {
  4.         @Override
  5.         public String key() {
  6.             return "ServerWeightTaskTimerInterval";
  7.         }
  8.     };
  9.     // 默认权重更新实现间隔
  10.     public static final int DEFAULT_TIMER_INTERVAL = 30 * 1000;
  12.     private int serverWeightTaskTimerInterval = DEFAULT_TIMER_INTERVAL;
  13.     // 权重累计
  14.     private volatile List<Double> accumulatedWeights = new ArrayList<Double>();
  15.     // 随机工具
  16.     private final Random random = new Random();
  17.     // 定时更新权重
  18.     protected Timer serverWeightTimer = null;
  19.     // 多线程更新权重时，保证只有一个执行就行
  20.     protected AtomicBoolean serverWeightAssignmentInProgress = new AtomicBoolean(false);
  21.     String name = "unknown";
  23.     // 构造方法
  24.     @Override
  25.     public void setLoadBalancer(ILoadBalancer lb) {
  26.         super.setLoadBalancer(lb);
  27.         if (lb instanceof BaseLoadBalancer) {
  28.             name = ((BaseLoadBalancer) lb).getName();
  29.         }
  30.         // 初始化权重定时计算器
  31.         initialize(lb);
  32.     }
  34.     // 初始化权重计算定时器
  35.     void initialize(ILoadBalancer lb) {        
  36.         if (serverWeightTimer != null) {
  37.             serverWeightTimer.cancel();
  38.         }
  39.         serverWeightTimer = new Timer("NFLoadBalancer-serverWeightTimer-" + name, true);
  40.         serverWeightTimer.schedule(new DynamicServerWeightTask(), 0, serverWeightTaskTimerInterval);
  41.         // do a initial run
  42.         // 一个定时任务的执行逻辑
  43.         ServerWeight sw = new ServerWeight();
  44.         sw.maintainWeights();
  45.     }
  46. }

内部类的定时任务

class DynamicServerWeightTask extends TimerTask {

    public void run() {
        // Runable任务
        ServerWeight serverWeight = new ServerWeight();
        try {
            serverWeight.maintainWeights();
        } catch (Exception e) {
            logger.error("Error running DynamicServerWeightTask for {}", name, e);
        }
    }
}

class ServerWeight {

    // 定时任务，重算权重列表
    public void maintainWeights() {
        ILoadBalancer lb = getLoadBalancer();
        if (lb == null) {
            return;
        }
        if (!serverWeightAssignmentInProgress.compareAndSet(false,  true))  {
            return; 
        }
        // 
        try {
            logger.info("Weight adjusting job started");
            AbstractLoadBalancer nlb = (AbstractLoadBalancer) lb;
            LoadBalancerStats stats = nlb.getLoadBalancerStats();
            if (stats == null) {
                // no statistics, nothing to do
                return;
            }
            // 开始重算
            double totalResponseTime = 0;
            // find maximal 95% response time
            for (Server server : nlb.getAllServers()) {
                // this will automatically load the stats if not in cache
                ServerStats ss = stats.getSingleServerStat(server);
                totalResponseTime += ss.getResponseTimeAvg();
            }
            // weight for each server is (sum of responseTime of all servers - responseTime)
            // so that the longer the response time, the less the weight and the less likely to be chosen
            Double weightSoFar = 0.0;

            // 每个服务的权重计数
            List<Double> finalWeights = new ArrayList<Double>();
            for (Server server : nlb.getAllServers()) {
                ServerStats ss = stats.getSingleServerStat(server);
                double weight = totalResponseTime - ss.getResponseTimeAvg();
                weightSoFar += weight;
                finalWeights.add(weightSoFar);   
            }
            // 重置权重
            setWeights(finalWeights);
        } catch (Exception e) {
            logger.error("Error calculating server weights", e);
        } finally {
            serverWeightAssignmentInProgress.set(false);
        }
    }
}

权重保存在一个列表中

// 权重累计
private volatile List<Double> accumulatedWeights = new ArrayList<Double>();

计算方式：
假设服务有4个节点，每个节点的平均响应时间是A(10ms)，B(20ms)，C(30ms)，D(40ms)
totalResponseTime=10+20+30+40 = 100
A权重：wt_A=100-10 = 90
B权重：wt_B=100-20 + wt_A = 100-20+90=170
C权重：wt_C=100-30 + wt_B = 100-30+170=240
D权重：wt_D=100-30 + wt_C = 100-40+240=300
权重区间：A，B，C，D分别是
【[0-90)，[90-170)，[170-240]，[240-300)】
选择服务时，先生成一个300以内的随机数然后对300取模，得到的数在那个区间就调用哪个服务。
accumulatedWeights中存放的是累计权重accumulatedWeights=[90, 170, 240, 300]

public Server choose(ILoadBalancer lb, Object key) {
    if (lb == null) {
        return null;
    }
    Server server = null;
    while (server == null) {
        List<Double> currentWeights = accumulatedWeights;
        if (Thread.interrupted()) {
            return null;
        }
        List<Server> allList = lb.getAllServers();
        int serverCount = allList.size();
        if (serverCount == 0) {
            return null;
        }

        int serverIndex = 0;
        // 最后一个是累计权重之和
        double maxTotalWeight = currentWeights.size() == 0 ? 0 : currentWeights.get(currentWeights.size() - 1); 
        // 还没有权重规则，回退到轮训算法
        if (maxTotalWeight < 0.001d || serverCount != currentWeights.size()) {
            server =  super.choose(getLoadBalancer(), key);
            if(server == null) {
                return server;
            }
        } else {
            // 生成随机数
            double randomWeight = random.nextDouble() * maxTotalWeight;
            // 找到区间
            int n = 0;
            for (Double d : currentWeights) {
                if (d >= randomWeight) {
                    // 找到
                    serverIndex = n;
                    break;
                } else {
                    n++;
                }
            }

            server = allList.get(serverIndex);
        }
        if (server == null) {
            // 重新找
            /* Transient. */
            Thread.yield();
            continue;
        }
        if (server.isAlive()) {
            //返回节点
            return (server);
        }
        // Next.
        server = null;
    }
    return server;
}