HBase源码分析之HRegionServer上MemStore的flush处理流程
Posted lipeng_bigdata
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了HBase源码分析之HRegionServer上MemStore的flush处理流程相关的知识，希望对你有一定的参考价值。
在《HBase源码分析之HRegion上MemStore的flsuh流程（一）》、《HBase源码分析之HRegion上MemStore的flsuh流程（二）》等文中，我们介绍了HRegion上Memstore flush的主体流程和主要细节。但是，HRegion只是HBase表中按照行的方向对一片连续的数据区域的抽象，它并不能对外提供单独的服务，供客户端或者HBase其它实体调用。而HRegion上MemStore的flush还是要通过HRegionServer来对外提供服务的。下面，我们就详细探究下HRegionServer上是如何实现这点的。
在HRegionServer中，有一个叫做cacheFlusher的东东，它是什么呢？我们先看一下它是如何被定义的：
// Cache flushing
// memstore内存刷新管理对象
protected MemStoreFlusher cacheFlusher;/**
* Thread that flushes cache on request
* 处理刷新缓存请求的线程
*
* NOTE: This class extends Thread rather than Chore because the sleep time
* can be interrupted when there is something to do, rather than the Chore
* sleep time which is invariant.
*
* @see FlushRequester
*/
@InterfaceAudience.Private
class MemStoreFlusher implements FlushRequester
一、如何初始化cacheFlusher
首先，我们发现HRegionServer继承自HasThread，而HasThread实现了Runnable接口，那么在其内部肯定会执行run()方法，而run()方法的开始，有如下代码：
try
// Do pre-registration initializations; zookeeper, lease threads, etc.
preRegistrationInitialization();
catch (Throwable e)
abort("Fatal exception during initialization", e);
if (!isStopped() && !isAborted())
initializeThreads();
// Cache flushing thread.
// 缓存刷新线程
this.cacheFlusher = new MemStoreFlusher(conf, this);首当其冲的便是flushQueue，其定义如下：
private final BlockingQueue<FlushQueueEntry> flushQueue =
new DelayQueue<FlushQueueEntry>();private final Map<HRegion, FlushRegionEntry> regionsInQueue =
new HashMap<HRegion, FlushRegionEntry>();接下来比较重要的便是flushHandlers，它是FlushHandler类型的一个数组，定义如下：
private final FlushHandler[] flushHandlers;FlushHandler是什么呢？它是处理缓存刷新的线程类，线程一旦启动后，在其run()方法内，就会不停的从flushQueue队列中拉取flush请求进行处理，其类的定义如下：
/**
* 处理缓存刷新的线程类
*/
private class FlushHandler extends HasThread 以上就是MemStoreFlusher内执行flush流程最重要的几个成员变量，其他的变量都是一些辅助性的，这里不再做详细介绍。
下面，我们来看下MemStoreFlusher的构造及成员变量的初始化，构造函数如下：
/**
* @param conf
* @param server
*/
public MemStoreFlusher(final Configuration conf,
final HRegionServer server)
super();
// 赋值RegionServer实例server
this.server = server;
// 线程唤醒频率threadWakeFrequency，取参数hbase.server.thread.wakefrequency配置的值，默认为10s，即线程的工作频率
this.threadWakeFrequency =
conf.getLong(HConstants.THREAD_WAKE_FREQUENCY, 10 * 1000);
// 获取最大可用堆内存max
long max = ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getMax();
// 获取全局memstore所占堆内存的百分比globalMemStorePercent
float globalMemStorePercent = HeapMemorySizeUtil.getGlobalMemStorePercent(conf, true);
// 获取全局memstore大小限制值globalMemStoreLimit
this.globalMemStoreLimit = (long) (max * globalMemStorePercent);
// 获取全局memstore大小限制值的低水平线百分比globalMemStoreLimitLowMarkPercent
this.globalMemStoreLimitLowMarkPercent =
HeapMemorySizeUtil.getGlobalMemStoreLowerMark(conf, globalMemStorePercent);
// 获取全局memstore大小限制值的低水平线globalMemStoreLimitLowMark
this.globalMemStoreLimitLowMark =
(long) (this.globalMemStoreLimit * this.globalMemStoreLimitLowMarkPercent);
// 获取阻塞等待时间blockingWaitTime，取参数hbase.hstore.blockingWaitTime，默认为90000
this.blockingWaitTime = conf.getInt("hbase.hstore.blockingWaitTime",
90000);
// 获取flush处理线程数目handlerCount，取参数hbase.hstore.flusher.count，默认为2
int handlerCount = conf.getInt("hbase.hstore.flusher.count", 2);
// 构造handlerCount个flush处理线程数组，默认为2个，可通过hbase.hstore.flusher.count设置
this.flushHandlers = new FlushHandler[handlerCount];
// 记录日志信息
LOG.info("globalMemStoreLimit=" +
StringUtils.humanReadableInt(this.globalMemStoreLimit) +
", globalMemStoreLimitLowMark=" +
StringUtils.humanReadableInt(this.globalMemStoreLimitLowMark) +
", maxHeap=" + StringUtils.humanReadableInt(max));
首先，看下获取全局memstore所占堆内存的百分比globalMemStorePercent的HeapMemorySizeUtil类的getGlobalMemStorePercent()方法，代码如下：
/**
* Retrieve global memstore configured size as percentage of total heap.
* 获取配置的全局memstore占整个heap内存的百分比
* @param c
* @param logInvalid
*/
public static float getGlobalMemStorePercent(final Configuration c, final boolean logInvalid)
// 获取全局memstore的大小，优先取参数hbase.regionserver.global.memstore.size，
// 未配置的话再取参数hbase.regionserver.global.memstore.upperLimit，
// 如果还未配置的话，默认为0.4
float limit = c.getFloat(MEMSTORE_SIZE_KEY,
c.getFloat(MEMSTORE_SIZE_OLD_KEY, DEFAULT_MEMSTORE_SIZE));
// 如果limit的值在区间(0,0.8]之外的话
if (limit > 0.8f || limit <= 0.0f)
if (logInvalid) // 根据参数logInvalid确定是否记录警告日志
LOG.warn("Setting global memstore limit to default of " + DEFAULT_MEMSTORE_SIZE
+ " because supplied value outside allowed range of (0 -> 0.8]");
// 将limit设置为0.4
limit = DEFAULT_MEMSTORE_SIZE;
// 返回limit
return limit;
1、获取配置的全局memstore占整个heap内存的百分比limit：优先取参数hbase.regionserver.global.memstore.size，未配置的话再取参数hbase.regionserver.global.memstore.upperLimit，如果还未配置的话，默认为0.4；
2、判断limit是否在区间(0,0.8]之外，根据参数logInvalid确定是否记录警告日志，并将limit设置为默认值0.4；
3、返回limit。
下面，我们再看下获取全局memstore大小限制值的低水平线百分比globalMemStoreLimitLowMarkPercent的HeapMemorySizeUtil类的getGlobalMemStoreLowerMark()方法，代码如下：
/**
* Retrieve configured size for global memstore lower water mark as percentage of total heap.
* 获取配置的全局memstore内存占全部heap内存的低水平线百分比
* @param c
* @param globalMemStorePercent
*/
public static float getGlobalMemStoreLowerMark(final Configuration c, float globalMemStorePercent)
// 取新参数hbase.regionserver.global.memstore.size.lower.limit
String lowMarkPercentStr = c.get(MEMSTORE_SIZE_LOWER_LIMIT_KEY);
// 如果新参数配置了的话，直接转化为double并返回
if (lowMarkPercentStr != null)
return Float.parseFloat(lowMarkPercentStr);
// 取旧参数hbase.regionserver.global.memstore.lowerLimit"
String lowerWaterMarkOldValStr = c.get(MEMSTORE_SIZE_LOWER_LIMIT_OLD_KEY);
// 如果旧参数配置的话，记录警告日志信息
if (lowerWaterMarkOldValStr != null)
LOG.warn(MEMSTORE_SIZE_LOWER_LIMIT_OLD_KEY + " is deprecated. Instead use "
+ MEMSTORE_SIZE_LOWER_LIMIT_KEY);
// 转化为double类型lowerWaterMarkOldVal
float lowerWaterMarkOldVal = Float.parseFloat(lowerWaterMarkOldValStr);
// 如果参数值大于计算得到的全局memstore所占堆内存的百分比，赋值为globalMemStorePercent，并记录日志信息
if (lowerWaterMarkOldVal > globalMemStorePercent)
lowerWaterMarkOldVal = globalMemStorePercent;
LOG.info("Setting globalMemStoreLimitLowMark == globalMemStoreLimit " + "because supplied "
+ MEMSTORE_SIZE_LOWER_LIMIT_OLD_KEY + " was > " + MEMSTORE_SIZE_OLD_KEY);
// 返回lowerWaterMarkOldVal / globalMemStorePercent
return lowerWaterMarkOldVal / globalMemStorePercent;
// 如果新旧参数均未配置的话，默认为0.95
return DEFAULT_MEMSTORE_SIZE_LOWER_LIMIT;
1、取新参数hbase.regionserver.global.memstore.size.lower.limit配置的值，如果新参数配置了的话，直接转化为double并返回；
2、如果新参数未配置的话，取旧参数hbase.regionserver.global.memstore.lowerLimit配置的值，如果旧参数配置的话，记录警告日志信息，并：
2.1、将旧参数配置的值转化为double类型lowerWaterMarkOldVal；
2.2、如果旧参数值大于计算得到的全局memstore所占堆内存的百分比，赋值为globalMemStorePercent，并记录日志信息；
2.3、返回lowerWaterMarkOldVal / globalMemStorePercent；
3、如果新旧参数均未配置的话，默认为0.95。
二、cacheFlusher如何处理flush请求
通过如何初始化cacheFlusher部分的介绍，我们已经知道，在MemStoreFlusher内部，存在两个存储flush请求及其HRegion封装类的队列和集合，即flushQueue和regionsInQueue，而MemStoreFlusher对外提供了一个requestFlush()方法，我们大体看下这个方法：
/**
* 请求刷新，
* 即将需要刷新MemStore的HRegion放置到regionsInQueue中，
* 同时根据HRegion构造FlushRegionEntry实例，添加到flushQueue中
*/
public void requestFlush(HRegion r)
synchronized (regionsInQueue) // 使用synchronized关键字对regionsInQueue进行线程同步
if (!regionsInQueue.containsKey(r)) // 如果regionsInQueue中不存在对应HRegion
// This entry has no delay so it will be added at the top of the flush
// queue. It'll come out near immediately.
// 将HRegion类型的r封装成FlushRegionEntry类型的fqe
// 这个fqe没有delay，即延迟执行时间，所以它被添加到flush队列的顶部。不久它将出列被处理。
FlushRegionEntry fqe = new FlushRegionEntry(r);
// 将HRegion->FlushRegionEntry的对应关系添加到regionsInQueue集合
// 将flush请求FlushRegionEntry添加到flushQueue队列
// 从这里可以看出regionsInQueue、flushQueue这两个成员变量go together
this.regionsInQueue.put(r, fqe);
this.flushQueue.add(fqe);
1、首先需要使用synchronized关键字对regionsInQueue进行线程同步，这么做是为了防止多线程的并发；
2、然后判断regionsInQueue中是否存在对应的HRegion，如果regionsInQueue集合中不存在对应HRegion的话继续，否则直接返回；
3、既然regionsInQueue集合中不存在对应HRegion，将HRegion类型的r封装成FlushRegionEntry类型的fqe；
4、将HRegion->FlushRegionEntry的对应关系添加到regionsInQueue集合；
5、将flush请求FlushRegionEntry添加到flushQueue队列。
从上述4、5步就可以看出regionsInQueue、flushQueue这两个成员变量go together，并且这个fqe没有delay，即延迟执行时间，所以它被添加到flush队列的顶部，不久它将出列被处理。这个该怎么理解呢？我们还是回到flushQueue的定义，flushQueue是一个存储了Region刷新缓存请求的队列，里面存储的是实现了FlushQueueEntry接口的对象，FlushQueueEntry没有定义任何行为，但是继承了java.util.concurrent.Delayed接口，故flushQueue是java中的DelayQueue，队列里存储的对象有一个过期时间的概念。
既然flush的请求已经被添加至flushQueue队列，相当于生产者已经把产品生产出来了，那么谁来消费呢？这个消费者的角色就是由FlushHandler线程来担任的。既然是线程，那么处理的逻辑肯定在其run()方法内，但是在研究其run()方法前，我们先看下flushQueue中存储的都是什么东西？
我们再回顾下flushQueue的定义，它是一个存储了FlushQueueEntry的队列DelayQueue。我们先看下FlushQueueEntry的定义：
interface FlushQueueEntry extends Delayed
众所周知，DelayQueue是一个无界的BlockingQueue，其内部存储的必然是实现了Delayed接口的对象。所以，FlushQueueEntry必须实现java的Delayed接口。而这种队列中的成员有一个最大特点，就是只有在其到期后才能出列，并且该队列内的成员都是有序的，从头至尾按照延迟到期时间的长短来排序。那么如何判断成员是否到期呢？对应成员对象的getDelay()方法返回一个小于等于0的值，就说明对应对象在队列中已到期，可以被取走。
既然DelayQueue中存储的成员对象都是有序的，那么实现了Delayed接口的类，必须提供compareTo()方法，用以排序，并且需要实现上述getDelay()方法，判断队内成员是否到期可以被取走。
接下来，我们分别来研究下WakeupFlushThread和FlushRegionEntry。
首先，WakeupFlushThread非常简单，没有任何实质内容，代码如下：
/**
* Token to insert into the flush queue that ensures that the flusher does not sleep
* 加入到刷新队列的确保刷新器不睡眠的令牌
*/
static class WakeupFlushThread implements FlushQueueEntry
@Override
public long getDelay(TimeUnit unit)
return 0;
@Override
public int compareTo(Delayed o)
return -1;
@Override
public boolean equals(Object obj)
return (this == obj);
接下来，我们再看下FlushRegionEntry类，其定义如下：
/**
* Datastructure used in the flush queue. Holds region and retry count.
* Keeps tabs on how old this object is. Implements @link Delayed. On
* construction, the delay is zero. When added to a delay queue, we'll come
* out near immediately. Call @link #requeue(long) passing delay in
* milliseconds before readding to delay queue if you want it to stay there
* a while.
*
* 用在刷新队列里的数据结构。保存region和重试次数。
* 跟踪对象多大（ps.即时间）
* 实现了java的Delayed接口。
* 在构造方法里，delay为0。
* 如果你想要它在队列中保持在在被重新加入delay队列之前
*
*
*/
static class FlushRegionEntry implements FlushQueueEntry
// 待flush的HRegion
private final HRegion region;
// 创建时间
private final long createTime;
// 何时到期
private long whenToExpire;
// 重入队列次数
private int requeueCount = 0;
FlushRegionEntry(final HRegion r)
// 待flush的HRegion
this.region = r;
// 创建时间为当前时间
this.createTime = EnvironmentEdgeManager.currentTime();
// 何时到期也为当前时间，意味着首次入队列时是没有延迟时间的，入列即可出列
this.whenToExpire = this.createTime;
/**
* @param maximumWait
* @return True if we have been delayed > <code>maximumWait</code> milliseconds.
*/
public boolean isMaximumWait(final long maximumWait)
return (EnvironmentEdgeManager.currentTime() - this.createTime) > maximumWait;
/**
* @return Count of times @link #requeue(long) was called; i.e this is
* number of times we've been requeued.
*/
public int getRequeueCount()
return this.requeueCount;
/**
* 类似重新入列的处理方法，重新入列次数requeueCount加1，何时到期未当前时间加参数when
*
* @param when When to expire, when to come up out of the queue.
* Specify in milliseconds. This method adds EnvironmentEdgeManager.currentTime()
* to whatever you pass.
* @return This.
*/
public FlushRegionEntry requeue(final long when)
this.whenToExpire = EnvironmentEdgeManager.currentTime() + when;
this.requeueCount++;
return this;
/**
* 判断何时到期的方法
*/
@Override
public long getDelay(TimeUnit unit)
// 何时到期减去当前时间
return unit.convert(this.whenToExpire - EnvironmentEdgeManager.currentTime(),
TimeUnit.MILLISECONDS);
/**
* 排序比较方法，根据判断何时到期的getDelay()方法来决定顺序
*/
@Override
public int compareTo(Delayed other)
// Delay is compared first. If there is a tie, compare region's hash code
int ret = Long.valueOf(getDelay(TimeUnit.MILLISECONDS) -
other.getDelay(TimeUnit.MILLISECONDS)).intValue();
if (ret != 0)
return ret;
// 何时到期时间一直的话，根据hashCode()来排序，其实也就是根据HRegion的hashCode()方法返回值来排序
FlushQueueEntry otherEntry = (FlushQueueEntry) other;
return hashCode() - otherEntry.hashCode();
@Override
public String toString()
return "[flush region " + Bytes.toStringBinary(region.getRegionName()) + "]";
@Override
public int hashCode()
int hash = (int) getDelay(TimeUnit.MILLISECONDS);
return hash ^ region.hashCode();
@Override
public boolean equals(Object obj)
if (this == obj)
return true;
if (obj == null || getClass() != obj.getClass())
return false;
Delayed other = (Delayed) obj;
return compareTo(other) == 0;
1、HRegion region：待flush的HRegion；
2、long createTime：创建时间；
3、long whenToExpire：何时到期；
4、int requeueCount = 0：重入队列次数。
而它的对象在初始化时，创建时间createTime设置为当前时间，何时到期whenToExpire也为当前时间，它判断是否到期的getDelay()方法为何时到期减去当前时间，也就意味着首次入队列时是没有延迟时间的，入列即可出列。另外，它在队列内部用于排序的compareTo()方法，也是首先根据判断何时到期的getDelay()方法来决定顺序，何时到期时间一致的话，根据hashCode()来排序，其实也就是根据HRegion的hashCode()方法返回值来排序。比较特别的是，这个类还提供了类似重新入列的处理方法，重新入列次数requeueCount加1，何时到期未当前时间加参数when，那么就相当于延期的了when时间变量。
说了那么多，接下来我们看下flush请求的实际处理流程，即FlushHandler的run()方法，其代码为：
@Override
public void run()
while (!server.isStopped()) // HRegionServer未停止的话，run()方法一直运行
FlushQueueEntry fqe = null;
try
// 标志位AtomicBoolean类型的wakeupPending设置为false
wakeupPending.set(false); // allow someone to wake us up again
// 从flushQueue队列中拉取一个FlushQueueEntry，即fqe
fqe = flushQueue.poll(threadWakeFrequency, TimeUnit.MILLISECONDS);
if (fqe == null || fqe instanceof WakeupFlushThread) // 如果fqe为空，或者为WakeupFlushThread
if (isAboveLowWaterMark())
// 由于内存高于低阈值，flush线程唤醒
LOG.debug("Flush thread woke up because memory above low water="
+ StringUtils.humanReadableInt(globalMemStoreLimitLowMark));
// 调用flushOneForGlobalPressure()方法，flush一个HRegion的MemStore，
// 降低MemStore的大小，预防OOM等异常情况的发生
if (!flushOneForGlobalPressure())
// Wasn't able to flush any region, but we're above low water mark
// This is unlikely to happen, but might happen when closing the
// 这是不可能发生的，但是当关闭全部服务器时可能发生，另外一个线程正在flush region；
// entire server - another thread is flushing regions. We'll just
// sleep a little bit to avoid spinning, and then pretend that
// we flushed one, so anyone blocked will check again
// 我们将会休眠一段时间，以避免旋转，然后假装我们flush了一个region，以使得被阻塞线程再次检查
Thread.sleep(1000);
wakeUpIfBlocking();// 唤醒其他阻塞线程
// Enqueue another one of these tokens so we'll wake up again
// 入列另一个令牌，以使我们之后再次被唤醒
wakeupFlushThread();
continue;
// fre不为空，且不为WakeupFlushThread的话，转化为FlushRegionEntry类型的fre
FlushRegionEntry fre = (FlushRegionEntry) fqe;
// 调用flushRegion()方法，并且如果结果为false的话，跳出循环
if (!flushRegion(fre))
break;
catch (InterruptedException ex)
continue;
catch (ConcurrentModificationException ex)
continue;
catch (Exception ex)
LOG.error("Cache flusher failed for entry " + fqe, ex);
if (!server.checkFileSystem())
break;
// 同时清空regionsInQueue和flushQueue
// 又是在一起啊
synchronized (regionsInQueue)
regionsInQueue.clear();
flushQueue.clear();
// Signal anyone waiting, so they see the close flag
// 唤醒所有的等待着，使得它们能够看到close标志
wakeUpIfBlocking();
// 记录日志信息
LOG.info(getName() + " exiting");
1、首先HRegionServer未停止的话，run()方法一直运行；
2、将标志位AtomicBoolean类型的wakeupPending设置为false；
3、从flushQueue队列中拉取一个FlushQueueEntry，即fqe：
3.1、如果fqe为空，或者为WakeupFlushThread：
3.1.1、如果通过isAboveLowWaterMark()方法判断全局MemStore的大小高于限制值得低水平线，调用flushOneForGlobalPressure()方法，按照一定策略，flush一个HRegion的MemStore，降低MemStore的大小，预防OOM等异常情况的发生，并入列另一个令牌，以使该线程之后再次被唤醒；
3.2、fre不为空，且不为WakeupFlushThread的话，转化为FlushRegionEntry类型的fre：调用flushRegion()方法，并且如果结果为false的话，跳出循环；
4、如果循环结束，同时清空regionsInQueue和flushQueue（ps：又是在一起啊O(∩_∩)O~）
5、唤醒所有的等待着，使得它们能够看到close标志；
6、记录日志。
我们注意到，WakeupFlushThread的主要作用是做为一个占位符或令牌插入到刷新队列flushQueue，以确保FlushHandler不会休眠，实际上WakeupFlushThread起到的作用不仅仅是这个，在FlushHandler线程不断的poll刷新队列flushQueue中的元素时，如果获取到的是一个WakeupFlushThread，它会发起 一个检测，即RegionServer的全局MemStore大小是否超过低水平线，如果未超过，WakeupFlushThread仅仅起到了一个占位符的作用，否则，WakeupFlushThread不仅做为占位符，保证刷新线程不休眠，还按照一定策略选择该RegionServer上的一个Region刷新memstore，以缓解RegionServer内存压力。
至于，如果全局MemStore的大小高于限制值得低水平线时，如何选择一个HRegion进行flush以缓解MemStore压力，还有HRegion的flush是如何发起的，我们下节再讲，敬请期待。
以上是关于HBase源码分析之HRegionServer上MemStore的flush处理流程的主要内容，如果未能解决你的问题，请参考以下文章
黑猴子的家：HBase 之HRegionserver挂死，日志出现Session Expired异常排查