HBase LRU源码解析

先来看下LruBlockCache的构造，关键是看清每个参数的作用:

  /**
   * Configurable constructor.  Use this constructor if not using defaults.
   * @param maxSize maximum size of this cache, in bytes
   * @param blockSize expected average size of blocks, in bytes
   * @param evictionThread whether to run evictions in a bg thread or not
   * @param mapInitialSize initial size of backing ConcurrentHashMap
   * @param mapLoadFactor initial load factor of backing ConcurrentHashMap
   * @param mapConcurrencyLevel initial concurrency factor for backing CHM
   * @param minFactor percentage of total size that eviction will evict until
   * @param acceptableFactor percentage of total size that triggers eviction
   * @param singleFactor percentage of total size for single -access blocks
   * @param multiFactor percentage of total size for multiple -access blocks
   * @param memoryFactor percentage of total size for in -memory blocks
   */
  public LruBlockCache( long maxSize, long blockSize, boolean evictionThread,
      int mapInitialSize, float mapLoadFactor, int mapConcurrencyLevel,
      float minFactor , float acceptableFactor,
      float singleFactor, float multiFactor, float memoryFactor) {
    if(singleFactor + multiFactor + memoryFactor != 1) {
      throw new IllegalArgumentException("Single, multi, and memory factors " +
          " should total 1.0");
    }
    if(minFactor >= acceptableFactor) {
      throw new IllegalArgumentException("minFactor must be smaller than acceptableFactor");
    }
    if(minFactor >= 1.0f || acceptableFactor >= 1.0f) {
      throw new IllegalArgumentException("all factors must be < 1" );
    }
    this. maxSize = maxSize;
    this. blockSize = blockSize;
    map = new ConcurrentHashMap<String,CachedBlock>(mapInitialSize,
        mapLoadFactor, mapConcurrencyLevel);
    this. minFactor = minFactor;
    this. acceptableFactor = acceptableFactor;
    this. singleFactor = singleFactor;
    this. multiFactor = multiFactor;
    this. memoryFactor = memoryFactor;
    this. stats = new CacheStats();
    this. count = new AtomicLong(0);
    this. elements = new AtomicLong(0);
    this. overhead = calculateOverhead(maxSize, blockSize, mapConcurrencyLevel);
    this. size = new AtomicLong(this.overhead);
    if(evictionThread) {
      this. evictionThread = new EvictionThread(this);
      this. evictionThread.start(); // FindBugs SC_START_IN_CTOR
    } else {
      this. evictionThread = null ;
    }
    this. scheduleThreadPool.scheduleAtFixedRate(new StatisticsThread(this),
        statThreadPeriod, statThreadPeriod , TimeUnit.SECONDS);
  }

接下来我们还需要了解几个相关的类：

public class CachedBlock implements HeapSize, Comparable<CachedBlock >

这个类代表了LruBlockCache中的一个条目，它里面有个非常关键的枚举：

  static enum BlockPriority {
    /**
     * Accessed a single time (used for scan -resistance)
     */
    SINGLE,
    /**
     * Accessed multiple times
     */
    MULTI,
    /**
     * Block from in -memory store
     */
    MEMORY
  };

通过以下代码可以更好的解释：

  public CachedBlock(String blockName, ByteBuffer buf, long accessTime,
      boolean inMemory ) {
    this. blockName = blockName;
    this.buf = buf;
    this. accessTime = accessTime;
    this. size = ClassSize. align(blockName.length()) +
    ClassSize.align(buf.capacity()) + PER_BLOCK_OVERHEAD;
    //第一次缓存一个block时，假设inMemory为false(默认)，那么会把这个CachedBlock的BlockPriority 设置为SINGLE, 否则为MEMORY。
    if(inMemory ) {
      this. priority = BlockPriority. MEMORY;
    } else {
      this. priority = BlockPriority. SINGLE;
    }
  }

 /**
   * Block has been accessed.  Update its local access time.
   */
  public void access(long accessTime) {
    this. accessTime = accessTime;
    // 当再次访问到时，假如此时CacheedBlock的BlockPriority的值是SINGLE,则把它变为MULTI 
    if(this. priority == BlockPriority. SINGLE) {
      this. priority = BlockPriority. MULTI;
    }
  }

另一方面，因为是LRU算法的实现，该类也实现了一个比较器：

  public int compareTo(CachedBlock that) {
    if(this. accessTime == that.accessTime ) return 0;
    return this.accessTime < that.accessTime ? 1 : -1;
  }

因为它实现了HeapSize这个接口，所以它能返回这个条目所占用的heap大小。

另一个关键的类是LruBlcokCache的内部类：

private class BlockBucket implements Comparable<BlockBucket >

这个类的作用是把所有的block分到不同的priority bucket中，每个BlockPriority都会有自己的一个bucket

我们可以开始看将一个新的block加入缓存：

  public void cacheBlock(String blockName, ByteBuffer buf, boolean inMemory ) {
    //private final ConcurrentHashMap<String,CachedBlock> map, 维护了缓存映射
    CachedBlock cb = map.get(blockName);
    //如果这个block已经被缓存了，那么就抛出一个运行时异常
    if(cb != null) {
      throw new RuntimeException("Cached an already cached block" );
    }
    //初始化一个新的CachedBlock
    cb = new CachedBlock(blockName, buf, count.incrementAndGet(), inMemory);
    //得到最新的heapsize
    long newSize = size.addAndGet(cb.heapSize());
    //将新增的block放到map中
    map.put(blockName, cb);
    //elements记录了目前缓存的数目
    elements.incrementAndGet();
    //假如最新的heapsize大于了acceptableSize(见下面的方法)，那么就需要进行evict动作
    if(newSize > acceptableSize() && ! evictionInProgress) {
      runEviction();
    }
  }
 //-----------------------------
  //假如没有特定的清理线程，那么就使用目前的线程来进行evict，这显然不是一个好主意，会造成阻塞，假如有清理线程，那么调用其evict方法 
  private void runEviction() {
    if(evictionThread == null) {
      evict();
    } else {
      evictionThread.evict(); //事实上是触发了清理线程的notify
    }
  }

有必要来看一下这个清理线程，在初始化LruBlockCache的时候就已经将其启动：

  private static class EvictionThread extends Thread {
    private WeakReference<LruBlockCache> cache;

    public EvictionThread(LruBlockCache cache) {
      super( "LruBlockCache.EvictionThread" );
      setDaemon( true);
      this. cache = new WeakReference<LruBlockCache>(cache);
    }

    @Override
    //这里使用了wait和notify机制，线程将一直等待，知道有notify消息过来说需要进行清理了
    public void run() {
      while( true) {
        synchronized(this ) {
          try {
            this.wait();
          } catch(InterruptedException e) {}
        }
        //这里cache使用了弱引用
        LruBlockCache cache = this.cache .get();
        if(cache == null) break;
        cache.evict();
      }
    }
    public void evict() {
      synchronized( this) {
        this.notify(); // FindBugs NN_NAKED_NOTIFY
      }
    }
  }

看具体的evict方法:

 void evict () {

    // Ensure only one eviction at a time
    if(!evictionLock.tryLock()) return;

    try {
      evictionInProgress = true;
      long currentSize = this.size .get();
      //需要释放掉的heap大小
      long bytesToFree = currentSize - minSize();

      if (LOG.isDebugEnabled()) {
        LOG.debug("Block cache LRU eviction started; Attempting to free " +
          StringUtils. byteDesc(bytesToFree) + " of total=" +
          StringUtils. byteDesc(currentSize));
      }

      if(bytesToFree <= 0) return;

      // Instantiate priority buckets
      //初始化三个桶，来存放single,multi,和memory，比例分别为25%，50%，25%
      BlockBucket bucketSingle = new BlockBucket(bytesToFree, blockSize ,
          singleSize());
      BlockBucket bucketMulti = new BlockBucket(bytesToFree, blockSize ,
          multiSize());
      BlockBucket bucketMemory = new BlockBucket(bytesToFree, blockSize ,
          memorySize());

      // Scan entire map putting into appropriate buckets
      for(CachedBlock cachedBlock : map.values()) {
        switch(cachedBlock.getPriority()) {
          case SINGLE : {
            bucketSingle.add(cachedBlock);
            break;
          }
          case MULTI : {
            bucketMulti.add(cachedBlock);
            break;
          }
          case MEMORY : {
            bucketMemory.add(cachedBlock);
            break;
          }
        }
      }

      //接下来将三个桶放入PriorityQueue
      PriorityQueue<BlockBucket> bucketQueue =
        new PriorityQueue<BlockBucket>(3);
      
      //会调用到CachedBlockQueue的add方法，下面分析
      bucketQueue.add(bucketSingle);
      bucketQueue.add(bucketMulti);
      bucketQueue.add(bucketMemory);

      int remainingBuckets = 3;
      long bytesFreed = 0;

      BlockBucket bucket;
     //溢出的多的那个桶，会越先被清理， 参看BlockBucket的compareTo方法
       //这里也说明，三个桶本身没有优先级
      while((bucket = bucketQueue.poll()) != null) {
        long overflow = bucket.overflow();
        if(overflow > 0) {
          // 本次要释放掉的内存
          long bucketBytesToFree = Math.min(overflow, (bytesToFree - bytesFreed) / remainingBuckets);
          //free方法在下面解释
          bytesFreed += bucket.free(bucketBytesToFree);
        }
        remainingBuckets--;
      }

      if (LOG.isDebugEnabled()) {
        long single = bucketSingle.totalSize();
        long multi = bucketMulti.totalSize();
        long memory = bucketMemory.totalSize();
        LOG.debug("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();
    }
  }

public void add(CachedBlock cb) {
    //如果当前的heapsize小于maxsize，直接加到queue中，这边的queue也是一个PriorityQueue
    if(heapSize < maxSize) {
      queue.add(cb);
      heapSize += cb.heapSize();
    } else {
      // 否则先取出列表头
      CachedBlock head = queue.peek();
      //判断假如的cb是不是比head大，实际的意义就是看新加入的cb是不是比head新，参看CachedBlock的compareTo方法，假如新，则继续
      if(cb.compareTo(head) > 0) {
        heapSize += cb.heapSize();
        heapSize -= head.heapSize();
        if(heapSize > maxSize ) {
          //取出head
          queue.poll();
        } else {
          heapSize += head.heapSize();
        }
        queue.add(cb);
      }
    }
  }

 public long free(long toFree) {
      //这边的queue是CacheBlockQueue类型，这个get方法很重要，它对PriorityQueue做了反序，这样的话就把时间最早的放在队列头
      LinkedList<CachedBlock> blocks = queue.get();
      long freedBytes = 0;
      for(CachedBlock cb: blocks) {
        freedBytes += evictBlock(cb);
        if(freedBytes >= toFree) {
          return freedBytes;
        }
      }
      return freedBytes;
    }

  //最后调用这个方法将block从map中移除：
  protected long evictBlock(CachedBlock block) {
    map.remove(block.getName());
    size.addAndGet(-1 * block.heapSize());
    elements.decrementAndGet();
    stats.evicted();
    return block.heapSize();
  }