MemStore刷写线程—MemStoreFlusher源代码分析，hbasememstore

MemStore刷写线程—MemStoreFlusher源代码分析，hbasememstore

     在HBase中表由一个或多个Region组成，而Region由一个或者多个Store组成，Store又由一个MenStore和若干个StoreFile组成。无论是向HBase写入数据还是请求读数据，都首先经过MemStore，对于写请求来说就是将数据直接写入MemStore，对于读请求来说就是先检查MenStore中是否包含相应的数据，如果有则直接读取该数据，否则在StoreFile中检索并读取数据。当写入MemStore中的数据达到一定的数量时，就需要将其中的数据刷写到StoreFile中，这是由后台线程自动完成的，负责该任务的主要Java类是MemStoreFlusher，本篇文章将结合该类的源代码学习HBase如何决定是否刷写数据到StoreFile中的，而不关注MemStoreFlusher如何被实例化及启动的（相应的代码位于HRegionServer.java中）以及写入StoreFile的代码，而仅仅关注HBase在满足什么条件的情况将触发刷写。

     官方文档将MenStoreFlusher解释为一个线程，而实际上该类既没有继承自Thread也没有实现Runnable接口，但定义了FlushHandler内部类，该类继承自HasThread，而HasThread实现了Runnable接口，在稍后的学习中将会看到FlushHandler就是负责刷写数据到StoreFile中的线程，并会根据参数hbase.hstore.flusher.count的值（默认为2）决定启动该线程的数量。该参数具有比较重要的意义，如果该值较小，则会导致刷写队列中包含过多的待刷写的Region，而如果该值较大的话，则会有较多并行执行刷写的线程，则会增加HDFS的负担，进而引起HBase更加频繁地执行Compaction。在MenStoreFlusher中刷写队列是由下面的数据结构定义的：

private final BlockingQueue<FlushQueueEntry> flushQueue = new DelayQueue<FlushQueueEntry>();
private final Map<HRegion, FlushRegionEntry> regionsInQueue = new HashMap<HRegion, FlushRegionEntry>();

    这两个数据结构必须一起使用，若某个FlushQueueEntry在二者中的一个，则在另一个中也必须能够找到该FlushQueueEntry。注意FlushRegionEntry实现了FlushQueueEntry接口，而后者继承自java.util.concurrent.Delayed接口。另一个实现了FlushQueueEntry的类为WakeupFlushThread类，该类的主要用作占位符插入到刷写队列中以确保刷写线程不会休眠。FlushRegionEntry类保存了请求刷写的Region、重试次数以及在刷写队列中存在的时间。刷写队列flushQueue的类型为DelayQueue，保存在该队列中的对象必须实现了Delayed接口，且位于该队列中的对象只有在其延迟过期后才可以被取出，位于队列头部的是延迟过期最长的对象，如果没有延迟过期，队列没有头部，poll操作将返回null。当调用对象的getDelay(TimeUnit.NANOSECONDS)方法返回值小于等于0时，该对象的延迟过期。由于DelayQueue实现了BlockingQueue接口，该接口是线程安全的，因此该队列也就支持BlockingQueue具有的阻塞入列和出列，即当队列为空时，将等待队列直到队列不为空再提取对象，当队列为满时，将等待队列有空闲位置时再插入对象。FlushRegionEntry的该方法实现如下：

@Override
 public long getDelay(TimeUnit unit) {
      return unit.convert(this.whenToExpire - EnvironmentEdgeManager.currentTime(), TimeUnit.MILLISECONDS);
 }

    FlushRegionEntry除了实现getDelay方法外，还定义了requeue()方法：

 public FlushRegionEntry requeue(final long when) {
      this.whenToExpire = EnvironmentEdgeManager.currentTime() + when;
      this.requeueCount++;
      return this;
 }

    稍后会介绍为什么会使用DelayQueue这样的数据结构定义刷写队列，现在继续看MemStoreFlusher的源代码。在MemStoreFlusher的构造函数中，读取hbase-site.xml中设置的与刷写相关的参数并赋值给相应的变量，并实例化了刷写线程FlushHandler：

public MemStoreFlusher(final Configuration conf, final HRegionServer server) {
    super();
    this.server = server;
    //hbase.server.thread.wakefrequency,默认值10s
    this.threadWakeFrequency =conf.getLong(HConstants.THREAD_WAKE_FREQUENCY, 10 * 1000);
    long max = ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getMax();
    //hbase.regionserver.global.memstore.size的值，默认为0.4，且该值必须小于等于0.8，大于0，即(0.0,0.8]
    float globalMemStorePercent = HeapMemorySizeUtil.getGlobalMemStorePercent(conf, true);
    this.globalMemStoreLimit = (long) (max * globalMemStorePercent);
    //hbase.regionserver.global.memstore.size.lower.limit
    this.globalMemStoreLimitLowMarkPercent =  HeapMemorySizeUtil.getGlobalMemStoreLowerMark(conf,globalMemStorePercent);
    //RS Xmx * hbase.regionserver.global.memstore.size * hbase.regionserver.global.memstore.size.lower.limit
    this.globalMemStoreLimitLowMark =  (long) (this.globalMemStoreLimit * this.globalMemStoreLimitLowMarkPercent);
    //默认值为90s，如果任何一个Store总的StoreFile数量大于hbase.hstore.blockingStoreFiles值，HRegion将阻塞更新直到该参数
    //设置的时间到达或者Compaction完成。在该参数设置的时间到期后，即使Compaction没有完成，HRegion也将不再阻塞更新
    this.blockingWaitTime = conf.getInt("hbase.hstore.blockingWaitTime", 90000);
    int handlerCount = conf.getInt("hbase.hstore.flusher.count", 2);
    this.flushHandlers = new FlushHandler[handlerCount];
  }

    MemStoreFlusher定义了三个刷写方法，分别为：flushOneForGlobalPressure、flushRegion(final FlushRegionEntry fqe)和flushRegion(finalHRegion region, final boolean emergencyFlush)其中前两个方法最终调用了最后一个方法，也就是只有最后一个方法执行了实际的刷写操作，前两个方法都在FlushHandler刷写线程中调用，分别用于当RegionServer中的所有MemStore的大小超过了RS Xmx *hbase.regionserver.global.memstore.size *hbase.regionserver.global.memstore.size.lower.limit进行的刷写和当某个Region需要执行刷写操作。

    首先看看flushOneForGlobalPressure方法的作用，从该方法的名称就可以看出，该方法用于缓解全局MemStore缓存的压力，在FlushHandler刷写线程的run方法中，当刷写队列中没有要求立刻执行刷写的Region时，刷写线程会判断所有MemStore的大小是否大参数设置的阈值，如果是，则执行该方法。在该方法中，刷写是按照Region的大小进行的，但不一定对最大的Region执行刷写，还需要进一步判断。判断主要是由方法getBiggestMemstoreRegion(SortedMap<Long, HRegion> regionsBySize, Set<HRegion>excludedRegions, boolean checkStoreFileCount)完成的，该方法的第三个参数决定是否检查当前Region的StoreFile文件数量（决定是否执行Compact操作），如果当前Region的StoreFile数量超过了参数hbase.hstore.blockingStoreFiles设置的值（默认值为10），且checkStoreFileCount的值为true，则当前Region不执行刷写，其它不执行刷写的情况还包括当前Region正在执行刷写或者当前Region不可写。在flushOneForGlobalPressure中，对getBiggestMemstoreRegion连续调用两次，一次检查StoreFile的数量，一次不检查，这样的话，不检查StoreFile数量的调用返回的Region的大小总是大于等于检查StoreFile数量的调用返回的Region，在源代码中，前者声明为bestAnyRegion，后者声明为bestFlushableRegion。如果当前Region包含有过多的StoreFile，则bestFlushableRegion的大小小于当前Region，否则bestAnyRegion和bestFlushableRegion都为当前Region，因而二者也相等。如果bestFlushableRegion不为null（bestAnyRegion肯定不为null），且拥有过多StoreFile的bestAnyRegion的大小比bestFlushableRegion大2倍多，则选择bestAnyRegion为要刷写的Region，这是因为如果选择较小的bestFlushableRegion为要刷写的Region则会导致产生较多较小的StoreFile，从而引起Compaction。如果bestFlushableRegion为null，则bestAnyRegion为要刷写的Region，如果bestFlushableRegion且其大小大于等于bestAnyRegion的一半，则bestFlushableRegion为要刷写的Region。在确定了要执行刷写的Region后，将会调用flushRegion(regionToFlush, true)方法执行刷写操作。如果刷写不成功，则重复上述的步骤，并将当前执行刷写的Region排除在判断挑选之外。

    在刷写线程中调用的另一个刷写方法是flushRegion(final FlushRegionEntry fqe)，该方法用于检查指定的Region是否有过多的StoreFile，如果否的话则调用flushRegion(finalHRegion region, final boolean emergencyFlush)执行刷写，否则的话再根据该Region是否等待了足够长的时间来决定执行刷写还是重新入列。这里足够长的时间指的是参数hbase.hstore.blockingWaitTime设置的值（默认为90s），该参数用于设置阻塞Region更新的时间，当Region的StoreFile数量超过了参数hbase.hstore.blockingStoreFiles设置的值，Region将阻塞更新而执行Compaction操作，当Compaction完成或者阻塞的时间超过了hbase.hstore.blockingWaitTime设置的值，Region将不再阻塞更新。如果该Region等待了足够的时间，即使该Region拥有过多的StoreFile，也将会对该Region执行刷写操作。如果该Region还没有等待hbase.hstore.blockingWaitTime设置的值，则先判断该Region重新入列的次数，如果该Region是第一次从刷写队列中取出（即该队列没有重复入列），再判断该Region是否需要执行split操作，如果不需要执行split操作，则请求执行Compaction（由于存在太多StoreFile）。对于没有等待足够长时间的Region，将该队列重新入列（此时重新入列的次数将会加1），并设置延迟时间为百分之一hbase.hstore.blockingWaitTime的值，这样该Region至少要在刷写队列中等待百分之一hbase.hstore.blockingWaitTime才能够被再次取出，进而给合并线程时间来完成对StoreFile的合并，当该Region再次有机会执行刷写时，就有可能StoreFile的数量降低到hbase.hstore.blockingStoreFiles之下，进而可刷写数据到StoreFile中。从对该方法的描述中，可以知道为什么使用DelayQueue做为刷写队列的实现，就是为了让刷写队列中的Region必须等待设置的时间，以让合并线程有时间完成合并StoreFile的任务。

    上述两个方法都没有实现实际的刷写任务，而将刷写任务交给了flushRegion(final HRegion region, finalboolean emergencyFlush)方法。该方法的第二个参数的值由调用者决定，在上述的第一个方法中为true（因为所有MemStore的大小已经超过了设置的阈值），第二个方法中为false（因为仅仅是某个Region需要执行刷写操作）。如果第二个方法的参数为true，则将该Region从刷写队列中移除，如果为false，则DelayQueue的poll方法移除该Region。在该方法中调用了HRegion的flushcache方法（此处不关心具体是如何刷写数据到StoreFile，仅关注MemStoreFlusher中的实现，后续会详细研究flushcache的具体实现），flushcache的返回结果为HRegion.FlushResult，可能的值为：FLUSHED_NO_COMPACTION_NEEDED、FLUSHED_COMPACTION_NEEDED、CANNOT_FLUSH_MEMSTORE_EMPTY、CANNOT_FLUSH，接着根据返回结果判断是否要执行Compaction和Split，只能执行二者中的一种操作，无论Compaction还是Split都是由CompactSplitThread类完成的。当刷写成功完成后，将会记录刷写的完成时间。

    在MemStoreFlusher中还有一个方法用于阻塞Region的更新，该方法就是reclaimMemStoreMemory。该方法将检查当前RegionServer中所有MemStore的大小是否超过了参数hbase.regionserver.global.memstore.size设置的值，如果超过了该值，则会阻塞更新5秒钟，当但并不影响刷写线程的执行。由于当所有MemStore的值高于MemStore的最高水位时自然大于MemStore的低水位，所以在刷写线程的run方法中同样会调用flushOneForGlobalPressure方法以选择合适的Region并对其所拥有的MemStore进行刷写。每当对Region进行修改操作时都将调用reclaimMemStoreMemory方法，这将导致检查所有MemStore的大小，并在条件满足时阻塞更新操作。