HBase源码分析之compact请求发起时机、判断条件等详情（一）

一般说来，任何一个比较复杂的分布式系统，针对能够使得其性能得到大幅提升的某一内部处理流程，必然有一个定期检查机制，使得该流程在满足一定条件的情况下，能够自发的进行，这样才能够很好的体现出复杂系统的自我适应与自我调节能力。我们知道，HBase内部的compact处理流程是为了解决MemStore Flush之后，文件数目太多，导致读数据性能大大下降的一种自我调节手段，它会将文件按照某种策略进行合并，大大提升HBase的数据读性能。那么，基于我刚才的陈述，compact流程是否有一个定期检查机制呢？在满足什么条件的情况下，会触发compact请求呢？

针对第一个问题，回答当然是肯定的。在HRegionServer内部，有一个成员变量，定义如下：

/*
   * Check for compactions requests.
   * 检查合并请求
   */
  Chore compactionChecker;

单从注释，我们就可以看出，这个compactionChecker成员变量就是一个检查合并请求的Chore，那么什么是Chore呢？先来看下它的定义、成员变量以及构造函数。先来看下类的定义，代码如下：

/**
 * Chore is a task performed on a period in hbase.  The chore is run in its own
 * thread. This base abstract class provides while loop and sleeping facility.
 * If an unhandled exception, the threads exit is logged.
 * Implementers just need to add checking if there is work to be done and if
 * so, do it.  Its the base of most of the chore threads in hbase.
 *
 * <p>Don't subclass Chore if the task relies on being woken up for something to
 * do, such as an entry being added to a queue, etc.
 * 
 * Chore是定期在HBase中执行的一个任务。Chore在它所在的线程内执行。这个基础抽象类提供了loop循环和sleep机制。
 */
@InterfaceAudience.Private
public abstract class Chore extends HasThread {
 
}

首先，从类的定义我们可以看到，Chore继承自HasThread类，而HasThread类是一个实现了Runnable接口的抽象类，并且定义了一个抽象的run()方法。自然，Chore就是一个线程了。而通过注释，我们可以很清晰的知道以下三点：1、Chore是定期在HBase中执行的一个任务；2、Chore在它所在的线程内执行；3、这个基础抽象类提供了loop循环和sleep机制。

再来看下它的成员变量，主要包含以下几个：

private final Sleeper sleeper;// 睡眠器
protected final Stoppable stopper;

上面提到，Chore提供了sleep机制，那么这个机制就是依靠Sleeper类型的sleeper这个成员变量来实现的，而stopper则是实现了Stoppable接口的任何实例，实际上是工作线程所依附的可停止运行的载体，比如HRegionServer，载体停止运行后，工作线程。等到分析其run()方法时，我们再具体分析这两个变量。

然后，我们再看下Chore的构造方法，代码如下：

/**
   * @param p Period at which we should run.  Will be adjusted appropriately
   * should we find work and it takes time to complete.
   * @param stopper When {@link Stoppable#isStopped()} is true, this thread will
   * cleanup and exit cleanly.
   * 
   * 构造方法，需要name、p和stopper三个参数
   * p为run方法循环的周期
   * 
   */
  public Chore(String name, final int p, final Stoppable stopper) {
    super(name);
    if (stopper == null){
      throw new NullPointerException("stopper cannot be null");
    }
    this.sleeper = new Sleeper(p, stopper);
    this.stopper = stopper;
  }

它需要name、p和stopper三个参数，name很简单，String类型的线程名字而已，关键在于这个int类型的p和Stoppable类型的stopper，构造函数利用p和stopper生成了一个睡眠期sleeper，并将stopper赋值给其同名成员变量。

下面，我们来看下这个sleeper的实现吧！Sleeper类中定义了4个关键变量和两个关键方法，实现了一个简单的睡眠器，其4个关键成员变量如下：

private final int period;
  private final Stoppable stopper;
  private final Object sleepLock = new Object();
  private boolean triggerWake = false;

其中，period代表了睡眠周期，它是由上诉参数p赋值的，而stopper的含义与Chore中同名变量一样。sleepLock仅仅是一个Object对象，依靠它的wait()方法，我们可以实现对象等待一段时间；triggerWake是一个标志位，依靠它被设置为true，我们可以跳出睡眠，重新复苏。

再来看下它的最终要的两个方法，第一个便是睡眠器最主要的功能性方法--睡眠sleep()，代码如下：

/**
   * Sleep for period adjusted by passed <code>startTime<code>
   * @param startTime Time some task started previous to now.  Time to sleep
   * will be docked current time minus passed <code>startTime<code>.
   */
  public void sleep(final long startTime) {
    
	// 如果stopper已停止，直接返回
	if (this.stopper.isStopped()) {
      return;
    }
    
    // 当前时间now
    long now = System.currentTimeMillis();
    
    // 计算最新的需要等待的时间，循环周期减去已过去的时间
    long waitTime = this.period - (now - startTime);
    
    // 如果等待时间waitTime已超过周期period，那么直接将period赋值给waitTime，并记录警告信息
    if (waitTime > this.period) {
      LOG.warn("Calculated wait time > " + this.period +
        "; setting to this.period: " + System.currentTimeMillis() + ", " +
        startTime);
      waitTime = this.period;
    }
    
    // 当等待时间waitTime大于0时，一直循环
    while (waitTime > 0) {
      long woke = -1;
      try {
    	  
    	// 判断标志位triggerWake，如果为true，
    	// 即如果其他线程已唤醒该睡眠期，跳出循环，复位triggerWake为fale，直接返回，不再睡眠
        synchronized (sleepLock) {
          if (triggerWake) break;
          // 否则，依靠sleepLock等待waitTime时间
          sleepLock.wait(waitTime);
        }
        
        // 计算已睡眠时间slept
        woke = System.currentTimeMillis();
        long slept = woke - now;
        
        // 如果slept时间已超出周期10s，记录警告信息
        if (slept - this.period > MINIMAL_DELTA_FOR_LOGGING) {
          LOG.warn("We slept " + slept + "ms instead of " + this.period +
              "ms, this is likely due to a long " +
              "garbage collecting pause and it's usually bad, see " +
              "http://hbase.apache.org/book.html#trouble.rs.runtime.zkexpired");
        }
      } catch(InterruptedException iex) {
        // We we interrupted because we're meant to stop?  If not, just
        // continue ignoring the interruption
        if (this.stopper.isStopped()) {
          return;
        }
      }
      
      // 重新计算等待时间：等待周期减去已睡眠时间
      // Recalculate waitTime.
      woke = (woke == -1)? System.currentTimeMillis(): woke;
      waitTime = this.period - (woke - startTime);
    }
    
    // 标志位triggerWake复位为false，需要在sleepLock上用synchronized关键字进行同步
    synchronized(sleepLock) {
      triggerWake = false;
    }
  }

这个方法会根据传入的参数睡眠的起始时间startTime，结合睡眠器构造时设定好的睡眠周期period，以及当前时间now，计算出等待时间waitTime。而后，在一个等待时间waitTime大于0的while循环内，首先判断标志位triggerWake，如果其为true，则break，复位triggerWake并停止休眠，否则，利用sleepLock的wait()方法休眠指定时间waitTime，直到时间结束或者有其他线程设置triggerWake标志位为true并通过sleepLock的notifyAll()方法唤醒sleepLock对象，让其wait()方法抛出InterruptedException异常，继而重新计算等待时间，并进入下一个循环。此时，标志位triggerWake已设置为true，则直接跳出循环，结束休眠。而在休眠时间未到的情况下结束休眠的一种手段，就是通过调用另外一个很关键的方法skipSleepCycle()来实现的，代码很简单，不做解释：

/**
   * If currently asleep, stops sleeping; if not asleep, will skip the next
   * sleep cycle.
   */
  public void skipSleepCycle() {
    synchronized (sleepLock) {
      // 标志位triggerWake设置为true
      triggerWake = true;
      // 唤醒等待在sleepLock上的其它线程
      sleepLock.notifyAll();
    }
  }

接下来，再看下Chore中最重要的run()方法，定义如下：

/**
   * @see java.lang.Thread#run()
   */
  @Override
  public void run() {
    try {
      boolean initialChoreComplete = false;
      
      // 只要stopper不停止，while循环就继续啊
      while (!this.stopper.isStopped()) {
    	// 开始时间
        long startTime = System.currentTimeMillis();
        try {
          // 如果是第一次循环，完成初始化工作
          if (!initialChoreComplete) {
            initialChoreComplete = initialChore();
          } else {
        	// 第一次后的每次循环，则周期性的调用chore()方法
            chore();
          }
        } catch (Exception e) {
          LOG.error("Caught exception", e);
          if (this.stopper.isStopped()) {
            continue;
          }
        }
        
        // 睡眠期睡眠一定的时间，然后再去调用chore()方法
        this.sleeper.sleep(startTime);
      }
    } catch (Throwable t) {
      LOG.fatal(getName() + "error", t);
    } finally {
      LOG.info(getName() + " exiting");
      cleanup();
    }
  }

这个run()方法的执行逻辑非常简单，只要stopper不停止，while循环就持续进行，首先，第一次进入run()方法时，标志位initialChoreComplete初始化为false，标志着Chore尚未初始化完毕，此时调用initialChore()做初始化工作，并返回初始化结果赋值给标志位initialChoreComplete，这个initialChore()目前是一个空方法，只返回true，而 第一次后的每次循环，则周期性的调用chore()方法，每次调用完chore()方法后，都通过睡眠器sleeper的sleep()方法，从每次进入while循环时获取的时刻startTime开始，休眠Chore构造函数传入的p时间，休眠过后再次执行chore()方法。如果stopper已停止，或者发生Throwable异常，则Chore调用cleanup()完成清理工作。

好了，Chore的运行机制到这里，已经给大家讲解清楚了。那么，再回到文章的初始，HRegionServer中名为compactionChecker的这个Chore，到底是如何初始化，并且都做了哪些事情呢？让我们继续往下看。

在前面讲解compact合并线程CompactSplitThread的文章中，我们了解过HRegionServer的initializeThreads()方法，它负责初始化工作在HRegionServer上的各种线程，包括CompactSplitThread，当然也就包括CompactionChecker。代码如下：

this.compactionChecker = new CompactionChecker(this, this.threadWakeFrequency, this);

它是通过构造一个CompactionChecker对象来完成初始化的。其构造方法如下：

// 构造函数
    CompactionChecker(final HRegionServer h, final int sleepTime,
        final Stoppable stopper) {
      
      // 调用父类Chore的构造方法
      super("CompactionChecker", sleepTime, h);
      
      // 将载体HRegionServer赋值给instance变量
      this.instance = h;
      LOG.info(this.getName() + " runs every " + StringUtils.formatTime(sleepTime));

      /* MajorCompactPriority is configurable.
       * If not set, the compaction will use default priority.
       */
      // 设置major合并优先级，取参数hbase.regionserver.compactionChecker.majorCompactPriority，默认为Integer.MAX_VALUE
      this.majorCompactPriority = this.instance.conf.
        getInt("hbase.regionserver.compactionChecker.majorCompactPriority",
        DEFAULT_PRIORITY);
    }

很简单，调用父类Chore的构造方法，设置上面提到的线程工作周期period和stopper，而这个工作周期period就是HRegionServer的threadWakeFrequency变量，它取自参数hbase.server.thread.wakefrequency，默认为10s，它是HBase上众多后台工作线程通用的工作频率，比如周期性MemStore刷新线程等。然后，构造方法还会将载体HRegionServer赋值给instance变量，并设置major合并优先级，取参数hbase.regionserver.compactionChecker.majorCompactPriority，默认为Integer.MAX_VALUE。

不止如此，在HRegionServer上的startServiceThreads()方法中，会将该线程设置为一个后台线程，目的就是为了方便虚拟机管理，当所有用户线程退出后，该后台线程也会自动退出，代码如下：

Threads.setDaemonThreadRunning(this.compactionChecker.getThread(), getName() +
      ".compactionChecker", uncaughtExceptionHandler);

至此，compactionChecker的初始化已完成。那么它是如何工作的呢？换句话，为了确保回答问题的全面性，也就是上面我们提到的第二个还没回答的问题：在满足什么条件的情况下，会触发compact请求呢？既然是个Chore，我们看下CompactionChecker的chore()方法，代码如下：

@Override
    // 线程的run方法会一直调用的函数chore()
    protected void chore() {
      
      // 循环检测HRegionServer的onlineRegions中的每个HRegion
      for (HRegion r : this.instance.onlineRegions.values()) {
        
    	// 对应HRegion为null的话，进入下一个HRegion的循环
    	if (r == null)
          continue;
        
    	// 取出每个Region中的Store
        for (Store s : r.getStores().values()) {
          try {
        	
        	// 调用Store的getCompactionCheckMultiplier()方法，获取合并检查倍增器multiplier
            long multiplier = s.getCompactionCheckMultiplier();
            
            // 合并检查倍增器multiplier必须确保大于0
            assert multiplier > 0;
            
            // 未到整数倍，跳过，每当迭代因子iteration为合并检查倍增器multiplier的整数倍时，才会发起检查
            if (iteration % multiplier != 0) continue;
            
            if (s.needsCompaction()) {// 需要合并的话，发起SystemCompaction请求
            	
              // Queue a compaction. Will recognize if major is needed.
              this.instance.compactSplitThread.requestSystemCompaction(r, s, getName()
                  + " requests compaction");
            } else if (s.isMajorCompaction()) {// 如果是Major合并的话，根据配置的major合并优先级majorCompactPriority确定发起合并请求
              
              // 如果工作线程中设置的合并优先级为Integer.MAX_VALUE，即默认，或者HRegion的合并优先级小于设置值的话
              if (majorCompactPriority == DEFAULT_PRIORITY
                  || majorCompactPriority > r.getCompactPriority()) {
            	  
            	// 使用默认优先级发起合并请求
                this.instance.compactSplitThread.requestCompaction(r, s, getName()
                    + " requests major compaction; use default priority", null);
              } else {
            	  
            	// 使用设置的优先级发起合并请求
                this.instance.compactSplitThread.requestCompaction(r, s, getName()
                    + " requests major compaction; use configured priority",
                  this.majorCompactPriority, null);
              }
            }
          } catch (IOException e) {
            LOG.warn("Failed major compaction check on " + r, e);
          }
        }
      }
      
      // 迭代计数器设置，累加1
      iteration = (iteration == Long.MAX_VALUE) ? 0 : (iteration + 1);
    }

整个工作流程很简单，chore()方法周期性的检测HRegionServer中所有在线Region的每个HStore，调用Store的getCompactionCheckMultiplier()方法，获取合并检查倍增器multiplier，当迭代因子iteration为合并检查倍增器multiplier的整数倍时，发起针对该HStore是否需要compact的检查，如果需要合并，则根据合并的种类，确定发起何种合并请求，并且如果是Major合并的话，则需要确定优先级。毕竟Major是最耗费资源的compact，为了合理有效的利用资源，也为了防止系统性能瓶颈，增加优先级就显得十分有必要了。整个流程比较清晰，而且上述代码注释也很详细，读者可自行补脑。

下面，我们针对几个要点进行简要说明：

1、onlineRegions是HRegionServer上存储的所有能够提供有效服务的在线Region集合；

2、整个检查过程是先轮询HRegion，然后针对HRegion上每个HStore进行的。并且，非常重要的是，它并不是对每个HRegion上所有HStore挨个检查，而是利用取余算法，对Region上的HStore进行检查。而这个过程的关键，就是上述代码中的合并检查倍增器multiplier，该值如果配置为1的话，则是挨个检查，如果配置成2的话，则是隔一个检查一个，依次类推。这个multiplier的获取，是通过HStore的getCompactionCheckMultiplier()方法获取的，它实际上是获取的HStore的compactionCheckMultiplier变量，而其初始化，则是取参数hbase.server.compactchecker.interval.multiplier，默认为1000。代码如下：

@Override
  public long getCompactionCheckMultiplier() {
    return this.compactionCheckMultiplier;
  }

// 取参数hbase.server.compactchecker.interval.multiplier，默认为1000
    this.compactionCheckMultiplier = conf.getInt(
        COMPACTCHECKER_INTERVAL_MULTIPLIER_KEY, DEFAULT_COMPACTCHECKER_INTERVAL_MULTIPLIER);
    if (this.compactionCheckMultiplier <= 0) {
      LOG.error("Compaction check period multiplier must be positive, setting default: "
          + DEFAULT_COMPACTCHECKER_INTERVAL_MULTIPLIER);
      this.compactionCheckMultiplier = DEFAULT_COMPACTCHECKER_INTERVAL_MULTIPLIER;
    }

3、对于是否需要合并，则是通过HStore的needsCompaction()方法判断的，代码如下：

@Override
  public boolean needsCompaction() {
    return this.storeEngine.needsCompaction(this.filesCompacting);
  }

而通过StoreEngine的一种实现DefaultStoreEngine，还有CompactionPolicy的一种实现RatioBasedCompactionPolicy等一系列调用，最终实现为如下代码：

public boolean needsCompaction(final Collection<StoreFile> storeFiles,
      final List<StoreFile> filesCompacting) {
	 // storeFile的总数减去正在合并的文件的数目
    int numCandidates = storeFiles.size() - filesCompacting.size();
    // 如果这个数目超过配置中合并文件的最小值
    return numCandidates >= comConf.getMinFilesToCompact();
  }

很简单，storeFile的总数减去正在合并的文件的数目，如果这个数目超过配置中合并文件的最小值，则视为需要发起合并请求。这个配置中合并文件的最小值，就是通过如下代码设置的：

// 先取新参数hbase.hstore.compaction.min，未配置的话，再去旧参数hbase.hstore.compactionThreshold，
    // 再未配置的话则默认为3，但是最终不能小于2
    minFilesToCompact = Math.max(2, conf.getInt(HBASE_HSTORE_COMPACTION_MIN_KEY,
          /*old name*/ conf.getInt("hbase.hstore.compactionThreshold", 3)));

4、需要合并的话，则调用CompactSplitThread的requestSystemCompaction()方法发起SystemCompaction请求，而如果是Major合并的话，则需要根据配置的major合并优先级majorCompactPriority确定发起合并请求，继而调用CompactSplitThread的requestCompaction()方法发起合并请求。

那么，如何认定一个合并为Major合并呢？它的判断需要以下几个条件：

4.1、HStore下全部存储文件的Reader必须不为null，也就是全部文件必须处于打开状态，否则直接返回false；

4.2、根据合并策略来确定，以RatioBasedCompactionPolicy为例：

4.2.1、获取下一次需要Major合并的时间mcTime；

4.2.2、如果待合并的全部文件为空，或者下一次需要Major合并的时间为0，直接返回false；

4.2.3、获取待合并文件中最小的时间戳lowTimestamp，并获取当前时间now；

4.2.4、如果最小时间戳lowTimestamp大于0，且小于当前时间now-减去下一次需要Major合并的时间：

4.2.4.1、获取列簇的TTL，即cfTtl；

4.2.4.2、如果存在多个待合并文件：直接返回true；

4.2.4.3、如果只存在一个待合并文件：则首先获取文件的最小时间戳minTimestamp，然后计算文件存留时间oldest，如果该文件不是元数据相关文件，且如果列簇的TTL为FOREVER，且文件保留时间仍在TTL内，那么我们需要根据数据块的位置索引与参数hbase.hstore.min.locality.to.skip.major.compact大小来判断是否只针对一个文件做compact，此时的这个compact理解为压缩比合并更好点，这部分后面再讲合并策略时再着重描述。

至此，我们把HRegionServer内部一个合并检查线程的初始化、工作方式及compact检查机制等统统讲完了。那么是否只要有这个定期检查工作线程就可以保证compact及时、正常运行，就能保证HBase的高性能了呢？

No，No，No，等着HBase源码分析之compact请求发起时机、判断条件等详情（二）吧！O(∩_∩)O哈哈~