HBase源代码分析之MemStore的flush发起时机、推断条件等详情

前面的几篇文章。我们具体介绍了HBase中HRegion上MemStore的flsuh流程，以及HRegionServer上MemStore的flush处理流程。那么，flush究竟是在什么情况下触发的呢？本文我们将具体探究下HBase中MemStore的flush流程的发起时机。看看究竟都有哪些操作。或者哪些后台服务进程会触发MemStore的flush。

首先。在《HBase源代码分析之HRegionServer上MemStore的flush处理流程（一）》和《HBase源代码分析之HRegionServer上MemStore的flush处理流程（二）》两篇文章中，我们了解了HRegionServer上MemStore的flush处理流程，知道了这么一个事实：flush的请求会通过requestFlush()或requestDelayedFlush()方法被加入到MemStoreFlusher的flushQueue队列中。然后由其内部的FlushHandler线程组消费。对须要flush的HRegion进行处理。所以。我们首先能够知道，调用MemStoreFlusher这两个方法的地方。肯定就会是MemStore发起flush的部分时机，另外。也许会存在部分操作或者内部流程直接调用HRegion的flushcache()方法而触发flush。以下，我们開始总结下，都有哪些操作或者内部流程触发MemStore的flush，以及须要flush的推断条件，包含其它一些方面的详情。

一、通过将RequestFlush请求加入到MemStoreFlusher的flushQueue队列

（一）单个Put操作

在HRegion中处理Put操作的put(Put put)方法中。在開始运行操作前，首先会调用checkResources()方法检查资源，这个checkResources()实际上就是检查HRegion的MemStore大小是否超过一定的阈值，假设超过，则会调用requestFlush()方法发起对该HRegion的MemStore进行flush的请求。并抛出RegionTooBusyException异常。阻止该操作继续，兴许将要讲的Delete、Append等数据更新操作也是如此，在開始运行操作前都会调用这个checkResources()方法来检查资源。checkResources()方法代码例如以下：

/*

   * Check if resources to support an update.

   * 检測是否有足够的资源支持一个Put、Append等数据更新操作

   *

   * We throw RegionTooBusyException if above memstore limit

   * and expect client to retry using some kind of backoff

   * 假设超过memstore的限制，我们抛出RegionTooBusyException这个异常。而且期望客户端使用某种补偿进行重试

  */

  private void checkResources()

    throws RegionTooBusyException {

    // If catalog region, do not impose resource constraints or block updates.

	// 假设是Meta Region。不实施资源约束或堵塞更新

    if (this.getRegionInfo().isMetaRegion()) return;

    // 假设Region当前内存大小超过阈值

    // 这个memstoreSize是当前时刻HRegion上MemStore的大小。它是在Put、Append等操作中调用addAndGetGlobalMemstoreSize()方法实时更新的。

    // 而blockingMemStoreSize是HRegion上设定的MemStore的一个阈值。当MemStore的大小超过这个阈值时，将会堵塞数据更新操作

    if (this.memstoreSize.get() > this.blockingMemStoreSize) {

      // 更新堵塞请求计数器

      blockedRequestsCount.increment();

      // 请求刷新Region

      requestFlush();

      // 抛出RegionTooBusyException异常

      throw new RegionTooBusyException("Above memstore limit, " +

          "regionName=" + (this.getRegionInfo() == null ? "unknown" :

          this.getRegionInfo().getRegionNameAsString()) +

          ", server=" + (this.getRegionServerServices() == null ? "unknown" :

          this.getRegionServerServices().getServerName()) +

          ", memstoreSize=" + memstoreSize.get() +

          ", blockingMemStoreSize=" + blockingMemStoreSize);

    }

  }

首先。假设是Meta Region，不实施资源约束或堵塞更新；

然后，假设Region当前内存大小memstoreSize超过阈值blockingMemStoreSize，则更新堵塞请求计数器。发起刷新MemStore请求，并抛出RegionTooBusyException异常。堵塞数据更新操作。

我们首先来看下memstoreSize，这个memstoreSize是当前时刻HRegion上MemStore的大小，它是在Put、Append等操作中调用addAndGetGlobalMemstoreSize()方法实时更新的。代码例如以下：

  /**

   * Increase the size of mem store in this region and the size of global mem

   * store

   * @return the size of memstore in this region

   */

  public long addAndGetGlobalMemstoreSize(long memStoreSize) {

    if (this.rsAccounting != null) {

      rsAccounting.addAndGetGlobalMemstoreSize(memStoreSize);

    }

    return this.memstoreSize.addAndGet(memStoreSize);

  }

而blockingMemStoreSize是HRegion上设定的MemStore的一个阈值。当MemStore的大小超过这个阈值时。将会堵塞数据更新操作。

其定义在HRegion上线被构造时须要调用的一个setHTableSpecificConf()中，部分代码例如以下：

    // blockingMemStoreSize是HRegion上设定的MemStore的一个阈值，当MemStore的大小超过这个阈值时，将会堵塞数据更新操作

    // 它的计算是由memstoreFlushSize乘以一个比例。这个比例取自參数hbase.hregion.memstore.block.multiplier，

    // 该參数未配置的话。则默觉得4

    this.blockingMemStoreSize = this.memstoreFlushSize *

        conf.getLong(HConstants.HREGION_MEMSTORE_BLOCK_MULTIPLIER,

                HConstants.DEFAULT_HREGION_MEMSTORE_BLOCK_MULTIPLIER);

我们能够知道。它的计算是由memstoreFlushSize乘以一个比例，这个比例取自參数hbase.hregion.memstore.block.multiplier，该參数未配置的话。则默觉得4。

那么memstoreFlushSize是什么呢？memstoreFlushSize为HRegion上设定的一个阈值，当MemStore的大小超过这个阈值时。将会发起flush请求。它的计算首先是由Table决定的。即每一个表能够设定自己的memstoreFlushSize，通过keywordMEMSTORE_FLUSHSIZE来设定，假设表中未设定，则取參数hbase.hregion.memstore.flush.size，假设參数再无配置的话，则默觉得1024*1024*128L，即128M。

代码相同在setHTableSpecificConf()这种方法中。例如以下：

    if (this.htableDescriptor == null) return;

    long flushSize = this.htableDescriptor.getMemStoreFlushSize();

    if (flushSize <= 0) {

      flushSize = conf.getLong(HConstants.HREGION_MEMSTORE_FLUSH_SIZE,

        HTableDescriptor.DEFAULT_MEMSTORE_FLUSH_SIZE);

    }

    // memstoreFlushSize为HRegion上设定的一个阈值，当MemStore的大小超过这个阈值时，将会发起flush请求

    // 它的计算首先是由Table决定的，即每一个表能够设定自己的memstoreFlushSize，通过keywordMEMSTORE_FLUSHSIZE来设定，

    // 假设未设定，则取參数hbase.hregion.memstore.flush.size。參数未配置的话。则默觉得1024*1024*128L，即128M

    this.memstoreFlushSize = flushSize;

接下来。我们看下requestFlush()是怎样发送的。代码例如以下：

private void requestFlush() {

	// 推断HRegion上的rsServices是否为空，rsServices为HRegionServer提供的服务类。

	// HRegion通过持有它，才干够获得能够发起运行诸如flush、compact、split等部分操作的工具对象。

	if (this.rsServices == null) {

      return;

    }

	// synchronized同步检查writestate的状态。假设writestate的状态为flushRequested。则直接返回，避免反复请求。

	// 否则将writestate的flushRequested设置为true，并继续发起flush请求

    synchronized (writestate) {

      if (this.writestate.isFlushRequested()) {

        return;

      }

      writestate.flushRequested = true;

    }

    // Make request outside of synchronize block; HBASE-818.

    // 通过rsServices获得FlushRequester。继而调用其requestFlush()方法，将HRegion自身传入。发起flush请求。

    // 这个FlushRequester就是HRegionServer上的cacheFlusher。它的requestFlush()就会将flush请求增加到请求队列中，利用内部工作线程去处理

    this.rsServices.getFlushRequester().requestFlush(this);

    if (LOG.isDebugEnabled()) {

      LOG.debug("Flush requested on " + this);

    }

  }

首先。须要做一些必要的推断和状态设置。例如以下：

1、推断HRegion上的rsServices是否为空，rsServices为HRegionServer提供的服务类。

HRegion通过持有它，才可以获得可以发起运行诸如flush、compact、split等部分操作的工具对象。

2、检查writestate的状态，假设writestate的状态为flushRequested，则直接返回。避免反复请求，否则将writestate的flushRequested设置为true。并继续发起flush请求。

检查通过后，就会通过rsServices获得FlushRequester，继而调用其requestFlush()方法。将HRegion自身传入，发起flush请求。

这个FlushRequester就是HRegionServer上的cacheFlusher，它的requestFlush()就会将flush请求增加到请求队列中，利用内部工作线程去处理。getFlushRequester()方法代码在实现了RegionServerServices()接口的HRegionServer中，代码例如以下：

  /** @return reference to FlushRequester */

  @Override

  public FlushRequester getFlushRequester() {

    return this.cacheFlusher;

  }

继续回到put()方法，继而依次调用doBatchMutate()、batchMutate()、batchMutate()方法，运行Put操作。

而batchMutate()方法的定义，是针对批量的Put、Delete等操作而专门设计的一个方法，仅仅只是单个的Put等操作传入的是一个仅仅包括一个操作的数组。

batchMutate()方法会在批量操作未所有完毕前一直循环，每次循环时，都会调用checkResources()检測MemStore，并调用doMiniBatchMutation()方法完毕操作并同步更新HRegion的MemStore大小，获取其值为newSize。最后通过isFlushSize()方法推断是否须要发起一个flush请求来决定是否调用requestFlush()方法。代码例如以下：

**

   * Perform a batch of mutations.

   * It supports only Put and Delete mutations and will ignore other types passed.

   *

   * 完毕一批变化操作。

* 它仅支持Put和Delete操作，将忽略其它类型的操作。

   *

   * @param batchOp contains the list of mutations

   * @return an array of OperationStatus which internally contains the

   *         OperationStatusCode and the exceptionMessage if any.

   * @throws IOException

   */

  OperationStatus[] batchMutate(BatchOperationInProgress<?> batchOp) throws IOException {

    boolean initialized = false;

    // 操作是否为回放，回放的话为REPLAY_BATCH_MUTATE。否则为BATCH_MUTATE

    Operation op = batchOp.isInReplay() ?

Operation.REPLAY_BATCH_MUTATE : Operation.BATCH_MUTATE;

    // 開始Region上的操作

    startRegionOperation(op);

    try {

      // 操作未所有完毕前一直循环

      while (!batchOp.isDone()) {

        // 假设是回放操作，则检測ReadOnly

    	if (!batchOp.isInReplay()) {

          checkReadOnly();

        }

    	// 检測相关资源

        checkResources();

        // 未初始化的话要先初始化

        if (!initialized) {

          // 更新写请求计数器

          this.writeRequestsCount.add(batchOp.operations.length);

          // 假设不是日志回放，运行pre钩子方法

          if (!batchOp.isInReplay()) {

            doPreMutationHook(batchOp);

          }

          initialized = true;

        }

        // 运行操作，并返回添加的内存大小

        long addedSize = doMiniBatchMutation(batchOp);

        // 以原子操作的方式添加Region上的MemStore内存大小

        long newSize = this.addAndGetGlobalMemstoreSize(addedSize);

        // 内存超过阈值时，请求flush

        if (isFlushSize(newSize)) {

          requestFlush();

        }

      }

    } finally {

      // 关闭Region上的操作

      closeRegionOperation(op);

    }

    // 返回批操作的状态

    return batchOp.retCodeDetails;

  }

我们来看下isFlushSize()方法。非常easy。它就是通过推断当前MemStore大小newSize是否超过memstoreFlushSize来决定。

代码例如以下：

/*

   * @param size

   * @return True if size is over the flush threshold

   */

  private boolean isFlushSize(final long size) {

    return size > this.memstoreFlushSize;

  }

至此，单个Put操作讲完了。

（二）单个Delete操作

单个Delete操作与单个Put操作一样，代码例如以下：

//////////////////////////////////////////////////////////////////////////////

  // set() methods for client use.

  //////////////////////////////////////////////////////////////////////////////

  /**

   * @param delete delete object

   * @throws IOException read exceptions

   */

  public void delete(Delete delete)

  throws IOException {

    checkReadOnly();

    checkResources();

    startRegionOperation(Operation.DELETE);

    try {

      delete.getRow();

      // All edits for the given row (across all column families) must happen atomically.

      doBatchMutate(delete);

    } finally {

      closeRegionOperation(Operation.DELETE);

    }

  }

也是先调用checkResources()检查MemStore，再调用doBatchMutate()进行处理，同单个Put操作是一样。

读者可自行分析。

（三）checkAndMutate/checkAndRowMutate操作

checkAndMutate()/checkAndRowMutate()方法中。也是先调用checkResources()检查MemStore，再调用doBatchMutate()进行处理，同单个Put操作是一样。checkAndMutate/checkAndRowMutate操作的特点就是保证改动数据的原子性，也是属于数据更新的操作。

（四）单个Append操作

在HBase中的append()方法中。相同是先调用checkResources()方法检測HRegion的MemStore。然后在处理完append操作后。调用addAndGetGlobalMemstoreSize()方法更新并获取最新的MemStore大小size。继而调用isFlushSize()方法推断是否须要调用requestFlush()方法发起flush请求。处理模式也是同上述的单个Put、单个Delete操作大致相同。

（五）单个Increment操作

同单个Append操作。在HBase中的increment()方法中，对于MemStore的检查及推断是否须要flush等全然一致，不再赘述。

（六）批量操作

既然单个操作可能会引起flush。那么批量操作更不用说了。批量操作是通过batchMutate()方法实现的，上面已经分析过这种方法了。这里不再赘述。

综上所述，凡是涉及到数据更新的操作，比方Put、Delete、Append、increment等操作，均是先检查MemStore。假设其高于某个阈值。将会发送flush请求。并抛出异常，堵塞数据更新操作。

另外，在操作运行完成后。也会依据MemStore增长情况。推断是否达到了该触发flush的条件。假设条件满足。则会发送flush请求。

二、通过直接调用HRegion的flushcache()方法

（一）外部触发（命令行等）

这样的场景的是通过RegionServer上RSRpcServices的flushRegion()方法发起的。flushRegion()的代码例如以下：

  /**

   * Flush a region on the region server.

   * 刷新RegionServer上的 一个Region

   *

   * @param controller the RPC controller

   * @param request the request

   * @throws ServiceException

   */

  @Override

  @QosPriority(priority=HConstants.ADMIN_QOS)

  public FlushRegionResponse flushRegion(final RpcController controller,

      final FlushRegionRequest request) throws ServiceException {

    try {

      // 检測RegionServer的状态

      checkOpen();

      // 请求计数器加1

      requestCount.increment();

      // 获取须要flush的Region

      HRegion region = getRegion(request.getRegion());

      LOG.info("Flushing " + region.getRegionNameAsString());

      boolean shouldFlush = true;

      // 推断是否须要flush。推断的根据是region上次flush时间小于请求中getIfOlderThanTs()方法的返回值

      if (request.hasIfOlderThanTs()) {

        shouldFlush = region.getLastFlushTime() < request.getIfOlderThanTs();

      }

      FlushRegionResponse.Builder builder = FlushRegionResponse.newBuilder();

      // 假设应该刷新

      if (shouldFlush) {

    	// 開始时间

        long startTime = EnvironmentEdgeManager.currentTime();

        // 调用Region的flushcache()方法刷新

        HRegion.FlushResult flushResult = region.flushcache();

        if (flushResult.isFlushSucceeded()) {// 刷新成功的话，记录刷新耗时

          long endTime = EnvironmentEdgeManager.currentTime();

          regionServer.metricsRegionServer.updateFlushTime(endTime - startTime);

        }

        boolean result = flushResult.isCompactionNeeded();

        if (result) {// 须要合并的话。通过RegionServer的compactSplitThread发起合并请求

          regionServer.compactSplitThread.requestSystemCompaction(region,

            "Compaction through user triggered flush");

        }

        builder.setFlushed(result);

      }

      builder.setLastFlushTime(region.getLastFlushTime());

      return builder.build();

    } catch (DroppedSnapshotException ex) {

      // Cache flush can fail in a few places. If it fails in a critical

      // section, we get a DroppedSnapshotException and a replay of wal

      // is required. Currently the only way to do this is a restart of

      // the server.

      regionServer.abort("Replay of WAL required. Forcing server shutdown", ex);

      throw new ServiceException(ex);

    } catch (IOException ie) {

      throw new ServiceException(ie);

    }

  }

flushRegion()方法内部在推断应该进行flush后，会调用HRegion的flushcache()方法对其MemStore进行flush处理，代码非常easy，这里就不多说了。

（二）Region合并（不是compact。而是两个Region的merge）

在RSRpcServices中存在Region合并时调用的mergeRegions()方法。在其内部会先后调用regionA和regionB的flushcache()方法去flush每一个Region上的MemStore，然后再运行Region合并。关键代码例如以下：

      long startTime = EnvironmentEdgeManager.currentTime();

      // 刷新regionA的MemStore

      HRegion.FlushResult flushResult = regionA.flushcache();

      if (flushResult.isFlushSucceeded()) {

        long endTime = EnvironmentEdgeManager.currentTime();

        regionServer.metricsRegionServer.updateFlushTime(endTime - startTime);

      }

      startTime = EnvironmentEdgeManager.currentTime();

      // 刷新regionB的MemStore

      flushResult = regionB.flushcache();

      if (flushResult.isFlushSucceeded()) {

        long endTime = EnvironmentEdgeManager.currentTime();

        regionServer.metricsRegionServer.updateFlushTime(endTime - startTime);

      }

      regionServer.compactSplitThread.requestRegionsMerge(regionA, regionB, forcible,

          masterSystemTime);

（三）Region分裂

在RSRpcServices中存在Region分裂时调用的splitRegion()方法，也是先调用flushcache()将Region上的memstore刷新。

代码例如以下：

/**

   * Split a region on the region server.

   * 分裂RegionServer上的一个Region

   *

   * @param controller the RPC controller

   * @param request the request

   * @throws ServiceException

   */

  @Override

  @QosPriority(priority=HConstants.ADMIN_QOS)

  public SplitRegionResponse splitRegion(final RpcController controller,

      final SplitRegionRequest request) throws ServiceException {

    try {

      // 检測RegionServer状态

      checkOpen();

      // 请求计数器加1

      requestCount.increment();

      // 获取相应的Region

      HRegion region = getRegion(request.getRegion());

      // Region上开启一个SPLIT_REGION操作

      region.startRegionOperation(Operation.SPLIT_REGION);

      LOG.info("Splitting " + region.getRegionNameAsString());

      // Split開始时间

      long startTime = EnvironmentEdgeManager.currentTime();

      // 先调用flushcache()将Region上的memstore刷新

      HRegion.FlushResult flushResult = region.flushcache();

      if (flushResult.isFlushSucceeded()) {

    	// flush成功的话，RegionServer记录flush耗时

        long endTime = EnvironmentEdgeManager.currentTime();

        regionServer.metricsRegionServer.updateFlushTime(endTime - startTime);

      }

      byte[] splitPoint = null;

      // 假设请求中有切分点，则取请求中的切分点

      if (request.hasSplitPoint()) {

        splitPoint = request.getSplitPoint().toByteArray();

      }

      // 设置标志位splitRequest为tue，并设置明白的分裂点explicitSplitPoint

      region.forceSplit(splitPoint);

      // 调用RegionServer上compactSplitThread的requestSplit()方法。请求分裂region

      // 调用前，先调用Region的checkSplit()方法。确定能否够切分，并确定切分点

      regionServer.compactSplitThread.requestSplit(region, region.checkSplit());

      return SplitRegionResponse.newBuilder().build();

    } catch (DroppedSnapshotException ex) {

      regionServer.abort("Replay of WAL required. Forcing server shutdown", ex);

      throw new ServiceException(ex);

    } catch (IOException ie) {

      throw new ServiceException(ie);

    }

  }

（四）利用Bulk载入HFile

Bulk是HBase直接载入HFile存储数据的一种快速、有用的手段或工具。在HRegion中的bulkLoadHFiles()方法中，也会调用flushcache()方法刷新HRegion上的MemStore内存。相关代码例如以下：

if (assignSeqId) {

        FlushResult fs = this.flushcache();

        if (fs.isFlushSucceeded()) {

          seqId = fs.flushSequenceId;

        } else if (fs.result == FlushResult.Result.CANNOT_FLUSH_MEMSTORE_EMPTY) {

          seqId = fs.flushSequenceId;

        } else {

          throw new IOException("Could not bulk load with an assigned sequential ID because the " +

              "flush didn't run. Reason for not flushing: " + fs.failureReason);

        }

      }

（五）做Table的快照

在做表的快照时。会将相应Table中涉及到的Region的MemStore进行flush。做这项工作的是FlushSnapshotSubprocedure类中的RegionSnapshotTask。它在call()方法中。当snapshotSkipFlush为false时。会调用HRegion的flushcache()方法，对MemStore进行flush。相关代码例如以下：

 if (snapshotSkipFlush) {

        /*

         * This is to take an online-snapshot without force a coordinated flush to prevent pause

         * The snapshot type is defined inside the snapshot description. FlushSnapshotSubprocedure

         * should be renamed to distributedSnapshotSubprocedure, and the flush() behavior can be

         * turned on/off based on the flush type.

         * To minimized the code change, class name is not changed.

         */

          LOG.debug("take snapshot without flush memstore first");

        } else {

          LOG.debug("Flush Snapshotting region " + region.toString() + " started...");

          region.flushcache();

        }

综上所述，凡是涉及到Region个数改变（比方两个变一个的合并、一个变两个的分裂、一个拷贝一个的快照等）。大都须要先flush掉其MemStore。可是有个疑问，为什么非要做flush呢？将内存中的数据也合并、分裂或者拷贝。不是更快吗？难道是由于其不稳定？留待以后再深入分析吧！

好了。MemStore的flush发起时机、推断条件等详情基本上就这些了。假设有遗漏的，或者解说不深入的，以后再慢慢深入吧！

巴特西

HBase源代码分析之MemStore的flush发起时机、推断条件等详情

最新文章

热门文章