在《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;可以发现,cacheFlusher是MemStoreFlusher类型的一个对象,我们来看下类的注释及定义:
/** * 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上处理刷新缓存请求的线程。那么接下来的问题就是,cacheFlusher是如何被初始化的?它又是如何处理flush请求的?带着这两个问题,我们继续本文。
一、如何初始化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); }继续追踪preRegistrationInitialization()方法,在其内部,调用了initializeThreads()方法,如下:
if (!isStopped() && !isAborted()) { initializeThreads(); }而这个initializeThreads()方法,做的主要工作就是初始化HRegionServer内部的各种工作线程,其中就包括cacheFlusher,代码如下:
// Cache flushing thread. // 缓存刷新线程 this.cacheFlusher = new MemStoreFlusher(conf, this);接下来,我们在看看这个MemStoreFlusher类是如何定义及工作的。首先看下它最主要的几个成员变量:
首当其冲的便是flushQueue,其定义如下:
private final BlockingQueue<FlushQueueEntry> flushQueue = new DelayQueue<FlushQueueEntry>();flushQueue是MemStoreFlusher中非常重要的一个变量,它是一个存储了Region刷新缓存请求的队列。而与flushQueue同时被更新的是regionsInQueue,它存储的是HRegion到FlushRegionEntry映射关系的集合,FlushRegionEntry是对发起memstore刷新请求的HRegion的一个封装,不仅包含了HRegion实例,还包括HRegion刷新memstore请求的产生时间,到期时间,以及一种类似续约的处理方式,即延长该请求的到期时间等。regionsInQueue的定义如下:
private final Map<HRegion, FlushRegionEntry> regionsInQueue = new HashMap<HRegion, FlushRegionEntry>();flushQueue和regionsInQueue的更新是同步的,即如果在flushQueue中加入或删除一条记录,那么在regionsInQueue中也会同步加入或删除一条记录。
接下来比较重要的便是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)); }MemStoreFlusher的构造函数比较简单,我们重点分析下获取全局memstore所占堆内存的百分比globalMemStorePercent的HeapMemorySizeUtil类的getGlobalMemStorePercent()方法,和获取全局memstore大小限制值的低水平线百分比globalMemStoreLimitLowMarkPercent的HeapMemorySizeUtil类的getGlobalMemStoreLowerMark()方法。
首先,看下获取全局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; }这个方法的主要作用就是获取配置的全局memstore占整个heap内存的百分比。获取的逻辑如下:
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; }这个方法的主要作用就是获取配置的全局memstore内存占全部heap内存的低水平线百分比。获取的逻辑如下:
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); } } }requestFlush()方法的主要作用,就是添加一个flush region的请求至MemStoreFlusher内部队列。其主要逻辑如下:
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 { }一个集成了java的Delayed接口的无任何方法的空接口而已,那么它都有哪些实现类呢?答案就是WakeupFlushThread和FlushRegionEntry。在介绍这二者之前,我们首先介绍下flushQueue对应的队列类型---Java中的DelayQueue。
众所周知,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); } }它的主要作用是做为一个占位符或令牌插入到刷新队列flushQueue,以确保FlushHandler不会休眠。而且,其getDelay()方法返回值为0,说明其不存在延迟时间,入列后即可出列。而它的compareTo()方法返回的值是-1,说明它与其它WakeupFlushThread在队内的顺序是等价的,无前后之分,实际上WakeupFlushThread区分前后也没有意义,它本身也没有实质性的内容。
接下来,我们再看下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; } } }FlushRegionEntry类有几个很重要的对像:
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是如何发起的,我们下节再讲,敬请期待。