一、简介
BlockCache是HBase中的一个重要特性,相比于写数据时缓存为Memstore,读数据时的缓存则为BlockCache。
LruBlockCache是HBase中BlockCache的默认实现,它采用严格的LRU算法来淘汰Block。
二、缓存级别
目前有三种缓存级别,定义在BlockPriority中,如下:
public enum BlockPriority { /** * Accessed a single time (used for scan-resistance) */ SINGLE, /** * Accessed multiple times */ MULTI, /** * Block from in-memory store */ MEMORY }
1、SINGLE:主要用于scan等,避免大量的这种一次的访问导致缓存替换;
2、MULTI:多次缓存;
3、MEMORY:常驻缓存的,比如meta信息等。
三、缓存实现分析
LruBlockCache缓存的实现在方法cacheBlock()中,实现逻辑如下:
1、首先需要判断需要缓存的数据大小是否超过最大块大小,按照2%的频率记录warn类型log并返回;
2、从缓存map中根据cacheKey尝试获取已缓存数据块cb;
3、如果已经缓存过,比对下内容,如果内容不一样,抛出异常,否则记录warn类型log并返回;
4、获取当前缓存大小currentSize,获取可以接受的缓存大小currentAcceptableSize,计算硬性限制大小hardLimitSize;
5、如果当前大小超过硬性限制,当回收没在执行时,执行回收并返回,否则直接返回;
6、利用cacheKey、数据buf等构造Lru缓存数据块实例cb;
7、将cb放置入map缓存中;
8、元素个数原子性增1;
9、如果新大小超过当前可以接受的大小,且未执行回收过程中,执行内存回收。
详细代码如下,可自行阅读分析:
// BlockCache implementation /** * Cache the block with the specified name and buffer. * <p> * It is assumed this will NOT be called on an already cached block. In rare cases (HBASE-8547) * this can happen, for which we compare the buffer contents. * @param cacheKey block's cache key * @param buf block buffer * @param inMemory if block is in-memory * @param cacheDataInL1 */ @Override public void cacheBlock(BlockCacheKey cacheKey, Cacheable buf, boolean inMemory, final boolean cacheDataInL1) { // 首先需要判断需要缓存的数据大小是否超过最大块大小 if (buf.heapSize() > maxBlockSize) { // If there are a lot of blocks that are too // big this can make the logs way too noisy. // So we log 2% if (stats.failInsert() % 50 == 0) { LOG.warn("Trying to cache too large a block " + cacheKey.getHfileName() + " @ " + cacheKey.getOffset() + " is " + buf.heapSize() + " which is larger than " + maxBlockSize); } return; } // 从缓存map中根据cacheKey尝试获取已缓存数据块 LruCachedBlock cb = map.get(cacheKey); if (cb != null) {// 如果已经缓存过 // compare the contents, if they are not equal, we are in big trouble if (compare(buf, cb.getBuffer()) != 0) {// 比对缓存内容,如果不相等,抛出异常,否则记录warn日志 throw new RuntimeException("Cached block contents differ, which should not have happened." + "cacheKey:" + cacheKey); } String msg = "Cached an already cached block: " + cacheKey + " cb:" + cb.getCacheKey(); msg += ". This is harmless and can happen in rare cases (see HBASE-8547)"; LOG.warn(msg); return; } // 获取当前缓存大小 long currentSize = size.get(); // 获取可以接受的缓存大小 long currentAcceptableSize = acceptableSize(); // 计算硬性限制大小 long hardLimitSize = (long) (hardCapacityLimitFactor * currentAcceptableSize); if (currentSize >= hardLimitSize) {// 如果当前大小超过硬性限制,当回收没在执行时,执行回收并返回 stats.failInsert(); if (LOG.isTraceEnabled()) { LOG.trace("LruBlockCache current size " + StringUtils.byteDesc(currentSize) + " has exceeded acceptable size " + StringUtils.byteDesc(currentAcceptableSize) + " too many." + " the hard limit size is " + StringUtils.byteDesc(hardLimitSize) + ", failed to put cacheKey:" + cacheKey + " into LruBlockCache."); } if (!evictionInProgress) {// 当回收没在执行时,执行回收并返回 runEviction(); } return; } // 利用cacheKey、数据buf等构造Lru缓存数据块实例 cb = new LruCachedBlock(cacheKey, buf, count.incrementAndGet(), inMemory); long newSize = updateSizeMetrics(cb, false); // 放置入map缓存中 map.put(cacheKey, cb); // 元素个数原子性增1 long val = elements.incrementAndGet(); if (LOG.isTraceEnabled()) { long size = map.size(); assertCounterSanity(size, val); } // 如果新大小超过当前可以接受的大小,且未执行回收过程中 if (newSize > currentAcceptableSize && !evictionInProgress) { runEviction();// 执行内存回收 } }四、淘汰缓存实现分析
淘汰缓存的实现方式有两种:
1、第一种是在主线程中执行缓存淘汰;
2、第二种是在一个专门的淘汰线程中通过持有对外部类LruBlockCache的弱引用WeakReference来执行缓存淘汰。
应用那种方式,取决于构造函数中的boolean参数evictionThread,如下:
if(evictionThread) { this.evictionThread = new EvictionThread(this); this.evictionThread.start(); // FindBugs SC_START_IN_CTOR } else { this.evictionThread = null; }而在执行淘汰缓存的runEviction()方法中,有如下判断:
/** * Multi-threaded call to run the eviction process. * 多线程调用以执行回收过程 */ private void runEviction() { if(evictionThread == null) {// 如果未指定回收线程 evict(); } else {// 如果执行了回收线程 evictionThread.evict(); } }而EvictionThread的evict()实现如下:
@edu.umd.cs.findbugs.annotations.SuppressWarnings(value="NN_NAKED_NOTIFY", justification="This is what we want") public void evict() { synchronized(this) { this.notifyAll(); } }通过synchronized获取EvictionThread线程的对象锁,然后主线程通过回收线程对象的notifyAll唤醒EvictionThread线程,那么这个线程是何时wait的呢?答案就在其run()方法中,notifyAll()之后,线程run()方法得以继续执行:
@Override public void run() { enteringRun = true; while (this.go) { synchronized(this) { try { this.wait(1000 * 10/*Don't wait for ever*/); } catch(InterruptedException e) { LOG.warn("Interrupted eviction thread ", e); Thread.currentThread().interrupt(); } } LruBlockCache cache = this.cache.get(); if (cache == null) break; cache.evict(); } }线程会wait10s,放弃对象锁,在notifyAll()后,继续执行后面的淘汰流程,即:
/** * Eviction method. */ void evict() { // Ensure only one eviction at a time // 通过可重入互斥锁ReentrantLock确保同一时刻只有一个回收在执行 if(!evictionLock.tryLock()) return; try { // 标志位,是否正在进行回收过程 evictionInProgress = true; // 当前缓存大小 long currentSize = this.size.get(); // 计算应该释放的缓冲大小bytesToFree long bytesToFree = currentSize - minSize(); if (LOG.isTraceEnabled()) { LOG.trace("Block cache LRU eviction started; Attempting to free " + StringUtils.byteDesc(bytesToFree) + " of total=" + StringUtils.byteDesc(currentSize)); } // 如果需要回收的大小小于等于0,直接返回 if(bytesToFree <= 0) return; // Instantiate priority buckets // 实例化优先级队列:single、multi、memory BlockBucket bucketSingle = new BlockBucket("single", bytesToFree, blockSize, singleSize()); BlockBucket bucketMulti = new BlockBucket("multi", bytesToFree, blockSize, multiSize()); BlockBucket bucketMemory = new BlockBucket("memory", bytesToFree, blockSize, memorySize()); // Scan entire map putting into appropriate buckets // 扫描缓存,分别加入上述三个优先级队列 for(LruCachedBlock cachedBlock : map.values()) { switch(cachedBlock.getPriority()) { case SINGLE: { bucketSingle.add(cachedBlock); break; } case MULTI: { bucketMulti.add(cachedBlock); break; } case MEMORY: { bucketMemory.add(cachedBlock); break; } } } long bytesFreed = 0; if (forceInMemory || memoryFactor > 0.999f) {// 如果memoryFactor或者InMemory缓存超过99.9%, long s = bucketSingle.totalSize(); long m = bucketMulti.totalSize(); if (bytesToFree > (s + m)) {// 如果需要回收的缓存超过则全部回收Single、Multi中的缓存大小和,则全部回收Single、Multi中的缓存,剩余的则从InMemory中回收 // this means we need to evict blocks in memory bucket to make room, // so the single and multi buckets will be emptied bytesFreed = bucketSingle.free(s); bytesFreed += bucketMulti.free(m); if (LOG.isTraceEnabled()) { LOG.trace("freed " + StringUtils.byteDesc(bytesFreed) + " from single and multi buckets"); } // 剩余的则从InMemory中回收 bytesFreed += bucketMemory.free(bytesToFree - bytesFreed); if (LOG.isTraceEnabled()) { LOG.trace("freed " + StringUtils.byteDesc(bytesFreed) + " total from all three buckets "); } } else {// 否则,不需要从InMemory中回收,按照如下策略回收Single、Multi中的缓存:尝试让single-bucket和multi-bucket的比例为1:2 // this means no need to evict block in memory bucket, // and we try best to make the ratio between single-bucket and // multi-bucket is 1:2 long bytesRemain = s + m - bytesToFree; if (3 * s <= bytesRemain) {// single-bucket足够小,从multi-bucket中回收 // single-bucket is small enough that no eviction happens for it // hence all eviction goes from multi-bucket bytesFreed = bucketMulti.free(bytesToFree); } else if (3 * m <= 2 * bytesRemain) {// multi-bucket足够下,从single-bucket中回收 // multi-bucket is small enough that no eviction happens for it // hence all eviction goes from single-bucket bytesFreed = bucketSingle.free(bytesToFree); } else { // single-bucket和multi-bucket中都回收,且尽量满足回收后比例为1:2 // both buckets need to evict some blocks bytesFreed = bucketSingle.free(s - bytesRemain / 3); if (bytesFreed < bytesToFree) { bytesFreed += bucketMulti.free(bytesToFree - bytesFreed); } } } } else {// 否则,从三个队列中循环回收 PriorityQueue<BlockBucket> bucketQueue = new PriorityQueue<BlockBucket>(3); bucketQueue.add(bucketSingle); bucketQueue.add(bucketMulti); bucketQueue.add(bucketMemory); int remainingBuckets = 3; BlockBucket bucket; while((bucket = bucketQueue.poll()) != null) { long overflow = bucket.overflow(); if(overflow > 0) { long bucketBytesToFree = Math.min(overflow, (bytesToFree - bytesFreed) / remainingBuckets); bytesFreed += bucket.free(bucketBytesToFree); } remainingBuckets--; } } if (LOG.isTraceEnabled()) { long single = bucketSingle.totalSize(); long multi = bucketMulti.totalSize(); long memory = bucketMemory.totalSize(); LOG.trace("Block cache LRU eviction completed; " + "freed=" + StringUtils.byteDesc(bytesFreed) + ", " + "total=" + StringUtils.byteDesc(this.size.get()) + ", " + "single=" + StringUtils.byteDesc(single) + ", " + "multi=" + StringUtils.byteDesc(multi) + ", " + "memory=" + StringUtils.byteDesc(memory)); } } finally { // 重置标志位,释放锁等 stats.evict(); evictionInProgress = false; evictionLock.unlock(); } }
逻辑比较清晰,如下:
1、通过可重入互斥锁ReentrantLock确保同一时刻只有一个回收在执行;
2、设置标志位evictionInProgress,是否正在进行回收过程为true;
3、获取当前缓存大小currentSize;
4、计算应该释放的缓冲大小bytesToFree:currentSize - minSize();
5、如果需要回收的大小小于等于0,直接返回;
6、实例化优先级队列:single、multi、memory;
7、扫描缓存,分别加入上述三个优先级队列;
8、如果forceInMemory或者InMemory缓存超过99.9%:
8.1、如果需要回收的缓存超过则全部回收Single、Multi中的缓存大小和,则全部回收Single、Multi中的缓存,剩余的则从InMemory中回收(this means we need to evict blocks in memory bucket to make room,so the single and multi buckets will be emptied):
8.2、否则,不需要从InMemory中回收,按照如下策略回收Single、Multi中的缓存:尝试让single-bucket和multi-bucket的比例为1:2:
8.2.1、 single-bucket足够小,从multi-bucket中回收;
8.2.2、 multi-bucket足够小,从single-bucket中回收;
8.2.3、single-bucket和multi-bucket中都回收,且尽量满足回收后比例为1:2;
9、否则,从三个队列中循环回收;
10、最后,重置标志位,释放锁等。
四、实例化
参见《HBase-1.2.4 Allow block cache to be external分析》最后。