Hbase负载均衡流程以及源码

public BalancerChore(HMaster master) {

    super(master.getServerName() + "-BalancerChore",

        master.getConfiguration().getInt("hbase.balancer.period", 300000),

        master);//hbase.balancer.period 为负载均衡方法执行的周期，毫秒为单位，hbaser-site.xml中可以设置

    this.master = master;

  }
 
  @Override

  protected void chore() {
    try {

      master.balance();//执行负载均衡方法

    } catch (IOException e) {

      LOG.error("Failed to balance.", e);

    }
  }


//执行负载均衡的入口。

  public void run() {
    try {
      boolean initialChoreComplete = false;
      while (!this.stopper.isStopped()) {//stopper是Hmaster service,这里判断Hmaster是否是正常状态。
        long startTime = System.currentTimeMillis();
        try {
          if (!initialChoreComplete) {
            initialChoreComplete = initialChore();//在循环开始前，执行初始化方法，这里默认返回true;
          } else {
            chore();//执行负载均衡方法
          }
        } catch (Exception e) {
          if (this.stopper.isStopped()) {
            continue;
          }
        }
        this.sleeper.sleep(startTime);
      }
  ...
  }

org.apache.hadoop.hbase.master.HMaster

 public boolean balance() throws IOException {
    // 如果master没有被初始化，则不执行初始化操作
    if (!this.initialized) {
      LOG.debug("Master has not been initialized, don‘t run balancer.");
      return false;
    }

   //只能同时跑一个负载均衡方法
      if (!this.loadBalancerTracker.isBalancerOn()) return false;
    // Do this call outside of synchronized block.
    int maximumBalanceTime = getBalancerCutoffTime();
    synchronized (this.balancer) {
    //如果有region处于splitting状态，则不跑负载均衡方法。
      if (this.assignmentManager.getRegionStates().isRegionsInTransition()) {
        Map<String, RegionState> regionsInTransition =
          this.assignmentManager.getRegionStates().getRegionsInTransition();
    ...
        return false;
      }

      if (this.serverManager.areDeadServersInProgress()) {//如果有挂掉的region server则不执行负载均衡。
        LOG.debug("Not running balancer because processing dead regionserver(s): " +
          this.serverManager.getDeadServers());
        return false;
      }
...
      Map<TableName, Map<ServerName, List<HRegionInfo>>> assignmentsByTable =
        this.assignmentManager.getRegionStates().getAssignmentsByTable();//获取table下面的region server 和region。
      List<RegionPlan> plans = new ArrayList<RegionPlan>();
      //Give the balancer the current cluster state.
      this.balancer.setClusterStatus(getClusterStatus());//设置当前集群的状态
      for (Map<ServerName, List<HRegionInfo>> assignments : assignmentsByTable.values()) {//可以看到，负载均衡方法是以每个table作为负载均衡的依据的。
        List<RegionPlan> partialPlans = this.balancer.balanceCluster(assignments);//获取负载均衡计划表
        if (partialPlans != null) plans.addAll(partialPlans);
      }
...

      if (plans != null && !plans.isEmpty()) {
        for (RegionPlan plan: plans) {
        ...
          this.assignmentManager.balance(plan);//根据执行计划表的迁移内容。
         ...
        }
    }
    }

    // If LoadBalancer did not generate any plans, it means the cluster is already balanced.
    // Return true indicating a success.
    return true;
  }

protected boolean needsBalance(ClusterLoadState cs) {
    ...
    float average = cs.getLoadAverage(); // for logging 获取cluster中region server平均拥有的region数目
    int floor = (int) Math.floor(average * (1 - slop));//slop默认是0.2，可接受范围的最低值
    int ceiling = (int) Math.ceil(average * (1 + slop));//最高值
    if (!(cs.getMaxLoad() > ceiling || cs.getMinLoad() < floor)) {//如果cluster的最多和最少region的region server不在范围内，返回false表明需要进行负载均衡算法。
       ...
      return false;
    }
    return true;
}

/* 计算cluster的总得分*/

  protected double computeCost(Cluster cluster, double previousCost) {
    double total = 0;for (CostFunction c:costFunctions) {//CostFunction 根据某个维度计算分数 ，costFunctions的实现见下面代码。
      if (c.getMultiplier() <= 0) {//multiplier是权重。
        continue;
      }
      total += c.getMultiplier() * c.cost(cluster);//权重*当前维度的评分
      if (total > previousCost) {
        return total;
      }
    }
    return total;
  }

//costFunctions 初始化
    regionLoadFunctions = new CostFromRegionLoadFunction[] {
      new ReadRequestCostFunction(conf),//读请求维度评分
      new WriteRequestCostFunction(conf),//写请求维度评分
      new MemstoreSizeCostFunction(conf),//memstore 大小维度评分
      new StoreFileCostFunction(conf)//StoreFile 维度评分
    };

    costFunctions = new CostFunction[]{
      new RegionCountSkewCostFunction(conf),//region 数目 维度评分
      new MoveCostFunction(conf),//迁移region 维度评分
      localityCost,//本地相关 维度评分
      new TableSkewCostFunction(conf), //表 维度评分
      regionLoadFunctions[0],
      regionLoadFunctions[1],
      regionLoadFunctions[2],
      regionLoadFunctions[3],
    };

public static class RegionCountSkewCostFunction extends CostFunction {
    private static final String REGION_COUNT_SKEW_COST_KEY =
        "hbase.master.balancer.stochastic.regionCountCost";
    private static final float DEFAULT_REGION_COUNT_SKEW_COST = 500;//默认权重为500
    private double[] stats = null;
    RegionCountSkewCostFunction(Configuration conf) {
      super(conf);
      // Load multiplier should be the greatest as it is the most general way to balance data.
      this.setMultiplier(conf.getFloat(REGION_COUNT_SKEW_COST_KEY, DEFAULT_REGION_COUNT_SKEW_COST));//设置权重
    }

    @Override
    double cost(Cluster cluster) {
      if (stats == null || stats.length != cluster.numServers) {
        stats = new double[cluster.numServers];
      }
      for (int i =0; i < cluster.numServers; i++) {
        stats[i] = cluster.regionsPerServer[i].length;//当前维度是根据每个region server 的region数目作为评分标准。
      }
      return costFromArray(stats);
    }
  }

    protected double costFromArray(double[] stats) {//根据某一维度，每个region server计算出来的评分
      double totalCost = 0;
      double total = getSum(stats);//计算总分
      double mean = total/((double)stats.length);//获取每个region server的平均评分
      double count = stats.length;//region server的总数
      // Compute max as if all region servers had 0 and one had the sum of all costs.  This must be
      // a zero sum cost for this to make sense.
      //这里假设最坏的情况为（count-1）的region server的评分为0，剩下的一个region server 占有了所有的分数，也就是负载非常不均衡，全部压力都压到同一台region server上面了。计算出最大的差值max。

      double max = ((count - 1) * mean) + (total - mean);
      for (double n : stats) {//计算当前的差值
        double diff = Math.abs(mean - n);
        totalCost += diff;
      }

      double scaled =  scale(0, max, totalCost);//（最大差值/（当前差值-最小差值））
      return scaled;
    }

RegionPlan rp = new RegionPlan(region, initialServer, newServer); //initialServer的region需要迁移到newServer