SpringBoot2.x 官方推荐缓存框架-Caffeine高性能设计剖析

private static LoadingCache<String, String> cache = Caffeine.newBuilder()  
            //最大个数限制  
            .maximumSize(256L)  
            //初始化容量  
            .initialCapacity(1)  
            //访问后过期（包括读和写）  
            .expireAfterAccess(2, TimeUnit.DAYS)  
            //写后过期  
            .expireAfterWrite(2, TimeUnit.HOURS)  
            //写后自动异步刷新  
            .refreshAfterWrite(1, TimeUnit.HOURS)  
            //记录下缓存的一些统计数据，例如命中率等  
            .recordStats()  
            //cache对缓存写的通知回调  
            .writer(new CacheWriter<Object, Object>() {  
                @Override  
                public void write(@NonNull Object key, @NonNull Object value) {  
                    log.info("key={}, CacheWriter write", key);  
                }  
  
                @Override  
                public void delete(@NonNull Object key, @Nullable Object value, @NonNull RemovalCause cause) {  
                    log.info("key={}, cause={}, CacheWriter delete", key, cause);  
                }  
            })  
            //使用CacheLoader创建一个LoadingCache  
            .build(new CacheLoader<String, String>() {  
                //同步加载数据  
                @Nullable  
                @Override  
                public String load(@NonNull String key) throws Exception {  
                    return "value_" + key;  
                }  
  
                //异步加载数据  
                @Nullable  
                @Override  
                public String reload(@NonNull String key, @NonNull String oldValue) throws Exception {  
                    return "value_" + key;  
                }  
            });  

frequencySketch().increment(key);  

//FrequencySketch的一些属性  
  
//种子数  
static final long[] SEED = { // A mixture of seeds from FNV-1a, CityHash, and Murmur3  
    0xc3a5c85c97cb3127L, 0xb492b66fbe98f273L, 0x9ae16a3b2f90404fL, 0xcbf29ce484222325L};  
static final long RESET_MASK = 0x7777777777777777L;  
static final long ONE_MASK = 0x1111111111111111L;  
  
int sampleSize;  
//为了快速根据hash值得到table的index值的掩码  
//table的长度size一般为2的n次方，而tableMask为size-1，这样就可以通过&操作来模拟取余操作，速度快很多，老司机都知道  
int tableMask;  
//存储数据的一维long数组  
long[] table;  
int size;  
  
/**  
 * Increments the popularity of the element if it does not exceed the maximum (15). The popularity  
 * of all elements will be periodically down sampled when the observed events exceeds a threshold.  
 * This process provides a frequency aging to allow expired long term entries to fade away.  
 *  
 * @param e the element to add  
 */  
public void increment(@NonNull E e) {  
  if (isNotInitialized()) {  
    return;  
  }  
  
  //根据key的hashCode通过一个哈希函数得到一个hash值  
  //本来就是hashCode了，为什么还要再做一次hash？怕原来的hashCode不够均匀分散，再打散一下。  
  int hash = spread(e.hashCode());  
  //这句光看有点难理解  
  //就如我刚才说的，Caffeine把一个long的64bit划分成16个等分，每一等分4个bit。  
  //这个start就是用来定位到是哪一个等分的，用hash值低两位作为随机数，再左移2位，得到一个小于16的值  
  int start = (hash & 3) << 2;  
  
  //indexOf方法的意思就是，根据hash值和不同种子得到table的下标index  
  //这里通过四个不同的种子，得到四个不同的下标index  
  int index0 = indexOf(hash, 0);  
  int index1 = indexOf(hash, 1);  
  int index2 = indexOf(hash, 2);  
  int index3 = indexOf(hash, 3);  
  
  //根据index和start(+1, +2, +3)的值，把table[index]对应的等分追加1  
  //这个incrementAt方法有点难理解，看我下面的解释  
  boolean added = incrementAt(index0, start);  
  added |= incrementAt(index1, start + 1);  
  added |= incrementAt(index2, start + 2);  
  added |= incrementAt(index3, start + 3);  
  
  //这个reset等下说  
  if (added && (++size == sampleSize)) {  
    reset();  
  }  
}  
  
/**  
 * Increments the specified counter by 1 if it is not already at the maximum value (15).  
 *  
 * @param i the table index (16 counters)  
 * @param j the counter to increment  
 * @return if incremented  
 */  
boolean incrementAt(int i, int j) {  
  //这个j表示16个等分的下标，那么offset就是相当于在64位中的下标（这个自己想想）  
  int offset = j << 2;  
  //上面提到Caffeine把频率统计最大定为15，即0xfL  
  //mask就是在64位中的掩码，即1111后面跟很多个0  
  long mask = (0xfL << offset);  
  //如果&的结果不等于15，那么就追加1。等于15就不会再加了  
  if ((table[i] & mask) != mask) {  
    table[i] += (1L << offset);  
    return true;  
  }  
  return false;  
}  
  
/**  
 * Returns the table index for the counter at the specified depth.  
 *  
 * @param item the element's hash  
 * @param i the counter depth  
 * @return the table index  
 */  
int indexOf(int item, int i) {  
  long hash = SEED[i] * item;  
  hash += hash >>> 32;  
  return ((int) hash) & tableMask;  
}  
  
/**  
 * Applies a supplemental hash function to a given hashCode, which defends against poor quality  
 * hash functions.  
 */  
int spread(int x) {  
  x = ((x >>> 16) ^ x) * 0x45d9f3b;  
  x = ((x >>> 16) ^ x) * 0x45d9f3b;  
  return (x >>> 16) ^ x;  
}  

/**  
 * Returns the estimated number of occurrences of an element, up to the maximum (15).  
 *  
 * @param e the element to count occurrences of  
 * @return the estimated number of occurrences of the element; possibly zero but never negative  
 */  
@NonNegative  
public int frequency(@NonNull E e) {  
  if (isNotInitialized()) {  
    return 0;  
  }  
  
  //得到hash值，跟上面一样  
  int hash = spread(e.hashCode());  
  //得到等分的下标，跟上面一样  
  int start = (hash & 3) << 2;  
  int frequency = Integer.MAX_VALUE;  
  //循环四次，分别获取在table数组中不同的下标位置  
  for (int i = 0; i < 4; i++) {  
    int index = indexOf(hash, i);  
    //这个操作就不多说了，其实跟上面incrementAt是一样的，定位到table[index] + 等分的位置，再根据mask取出计数值  
    int count = (int) ((table[index] >>> ((start + i) << 2)) & 0xfL);  
    //取四个中的较小值  
    frequency = Math.min(frequency, count);  
  }  
  return frequency;  
}  

//size变量就是所有记录的频率统计之，即每个记录加1，这个size都会加1  
//sampleSize一个阈值，从FrequencySketch初始化可以看到它的值为maximumSize的10倍  
if (added && (++size == sampleSize)) {  
      reset();  
}  

/** Reduces every counter by half of its original value. */  
void reset() {  
  int count = 0;  
  for (int i = 0; i < table.length; i++) {  
    count += Long.bitCount(table[i] & ONE_MASK);  
    table[i] = (table[i] >>> 1) & RESET_MASK;  
  }  
  size = (size >>> 1) - (count >>> 2);  
}  

/**  
   * Performs the pending maintenance work and sets the state flags during processing to avoid  
   * excess scheduling attempts. The read buffer, write buffer, and reference queues are  
   * drained, followed by expiration, and size-based eviction.  
   *  
   * @param task an additional pending task to run, or {@code null} if not present  
   */  
  @GuardedBy("evictionLock")  
  void maintenance(@Nullable Runnable task) {  
    lazySetDrainStatus(PROCESSING_TO_IDLE);  
  
    try {  
      drainReadBuffer();  
  
      drainWriteBuffer();  
      if (task != null) {  
        task.run();  
      }  
  
      drainKeyReferences();  
      drainValueReferences();  
  
      expireEntries();  
      //把符合条件的记录淘汰掉  
      evictEntries();  
      //动态调整window区和protected区的大小  
      climb();  
    } finally {  
      if ((drainStatus() != PROCESSING_TO_IDLE) || !casDrainStatus(PROCESSING_TO_IDLE, IDLE)) {  
        lazySetDrainStatus(REQUIRED);  
      }  
    }  
  }  

//最大的个数限制  
long maximum;  
//当前的个数  
long weightedSize;  
//window区的最大限制  
long windowMaximum;  
//window区当前的个数  
long windowWeightedSize;  
//protected区的最大限制  
long mainProtectedMaximum;  
//protected区当前的个数  
long mainProtectedWeightedSize;  
//下一次需要调整的大小（还需要进一步计算）  
double stepSize;  
//window区需要调整的大小  
long adjustment;  
//命中计数  
int hitsInSample;  
//不命中的计数  
int missesInSample;  
//上一次的缓存命中率  
double previousSampleHitRate;  
  
final FrequencySketch<K> sketch;  
//window区的LRU queue（FIFO）  
final AccessOrderDeque<Node<K, V>> accessOrderWindowDeque;  
//probation区的LRU queue（FIFO）  
final AccessOrderDeque<Node<K, V>> accessOrderProbationDeque;  
//protected区的LRU queue（FIFO）  
final AccessOrderDeque<Node<K, V>> accessOrderProtectedDeque;  

/** The initial percent of the maximum weighted capacity dedicated to the main space. */  
static final double PERCENT_MAIN = 0.99d;  
/** The percent of the maximum weighted capacity dedicated to the main's protected space. */  
static final double PERCENT_MAIN_PROTECTED = 0.80d;  
/** The difference in hit rates that restarts the climber. */  
static final double HILL_CLIMBER_RESTART_THRESHOLD = 0.05d;  
/** The percent of the total size to adapt the window by. */  
static final double HILL_CLIMBER_STEP_PERCENT = 0.0625d;  
/** The rate to decrease the step size to adapt by. */  
static final double HILL_CLIMBER_STEP_DECAY_RATE = 0.98d;  
/** The maximum number of entries that can be transfered between queues. */  

/** Evicts entries if the cache exceeds the maximum. */  
@GuardedBy("evictionLock")  
void evictEntries() {  
  if (!evicts()) {  
    return;  
  }  
  //淘汰window区的记录  
  int candidates = evictFromWindow();  
  //淘汰Main区的记录  
  evictFromMain(candidates);  
}  
  
/**  
 * Evicts entries from the window space into the main space while the window size exceeds a  
 * maximum.  
 *  
 * @return the number of candidate entries evicted from the window space  
 */  
//根据W-TinyLFU，新的数据都会无条件的加到admission window  
//但是window是有大小限制，所以要“定期”做一下“维护”  
@GuardedBy("evictionLock")  
int evictFromWindow() {  
  int candidates = 0;  
  //查看window queue的头部节点  
  Node<K, V> node = accessOrderWindowDeque().peek();  
  //如果window区超过了最大的限制，那么就要把“多出来”的记录做处理  
  while (windowWeightedSize() > windowMaximum()) {  
    // The pending operations will adjust the size to reflect the correct weight  
    if (node == null) {  
      break;  
    }  
    //下一个节点  
    Node<K, V> next = node.getNextInAccessOrder();  
    if (node.getWeight() != 0) {  
      //把node定位在probation区  
      node.makeMainProbation();  
      //从window区去掉  
      accessOrderWindowDeque().remove(node);  
      //加入到probation queue，相当于把节点移动到probation区（晋升了）  
      accessOrderProbationDeque().add(node);  
      candidates++;  
      //因为移除了一个节点，所以需要调整window的size  
      setWindowWeightedSize(windowWeightedSize() - node.getPolicyWeight());  
    }  
    //处理下一个节点  
    node = next;  
  }  
  
  return candidates;  
}  

/**  
 * Evicts entries from the main space if the cache exceeds the maximum capacity. The main space  
 * determines whether admitting an entry (coming from the window space) is preferable to retaining  
 * the eviction policy's victim. This is decision is made using a frequency filter so that the  
 * least frequently used entry is removed.  
 *  
 * The window space candidates were previously placed in the MRU position and the eviction  
 * policy's victim is at the LRU position. The two ends of the queue are evaluated while an  
 * eviction is required. The number of remaining candidates is provided and decremented on  
 * eviction, so that when there are no more candidates the victim is evicted.  
 *  
 * @param candidates the number of candidate entries evicted from the window space  
 */  
//根据W-TinyLFU，从window晋升过来的要跟probation区的进行“PK”，胜者才能留下  
@GuardedBy("evictionLock")  
void evictFromMain(int candidates) {  
  int victimQueue = PROBATION;  
  //victim是probation queue的头部  
  Node<K, V> victim = accessOrderProbationDeque().peekFirst();  
  //candidate是probation queue的尾部，也就是刚从window晋升来的  
  Node<K, V> candidate = accessOrderProbationDeque().peekLast();  
  //当cache不够容量时才做处理  
  while (weightedSize() > maximum()) {  
    // Stop trying to evict candidates and always prefer the victim  
    if (candidates == 0) {  
      candidate = null;  
    }  
  
    //对candidate为null且victim为bull的处理  
    if ((candidate == null) && (victim == null)) {  
      if (victimQueue == PROBATION) {  
        victim = accessOrderProtectedDeque().peekFirst();  
        victimQueue = PROTECTED;  
        continue;  
      } else if (victimQueue == PROTECTED) {  
        victim = accessOrderWindowDeque().peekFirst();  
        victimQueue = WINDOW;  
        continue;  
      }  
  
      // The pending operations will adjust the size to reflect the correct weight  
      break;  
    }  
  
    //对节点的weight为0的处理  
    if ((victim != null) && (victim.getPolicyWeight() == 0)) {  
      victim = victim.getNextInAccessOrder();  
      continue;  
    } else if ((candidate != null) && (candidate.getPolicyWeight() == 0)) {  
      candidate = candidate.getPreviousInAccessOrder();  
      candidates--;  
      continue;  
    }  
  
    // Evict immediately if only one of the entries is present  
    if (victim == null) {  
      @SuppressWarnings("NullAway")  
      Node<K, V> previous = candidate.getPreviousInAccessOrder();  
      Node<K, V> evict = candidate;  
      candidate = previous;  
      candidates--;  
      evictEntry(evict, RemovalCause.SIZE, 0L);  
      continue;  
    } else if (candidate == null) {  
      Node<K, V> evict = victim;  
      victim = victim.getNextInAccessOrder();  
      evictEntry(evict, RemovalCause.SIZE, 0L);  
      continue;  
    }  
  
    // Evict immediately if an entry was collected  
    K victimKey = victim.getKey();  
    K candidateKey = candidate.getKey();  
    if (victimKey == null) {  
      @NonNull Node<K, V> evict = victim;  
      victim = victim.getNextInAccessOrder();  
      evictEntry(evict, RemovalCause.COLLECTED, 0L);  
      continue;  
    } else if (candidateKey == null) {  
      candidates--;  
      @NonNull Node<K, V> evict = candidate;  
      candidate = candidate.getPreviousInAccessOrder();  
      evictEntry(evict, RemovalCause.COLLECTED, 0L);  
      continue;  
    }  
  
    //放不下的节点直接处理掉  
    if (candidate.getPolicyWeight() > maximum()) {  
      candidates--;  
      Node<K, V> evict = candidate;  
      candidate = candidate.getPreviousInAccessOrder();  
      evictEntry(evict, RemovalCause.SIZE, 0L);  
      continue;  
    }  
  
    //根据节点的统计频率frequency来做比较，看看要处理掉victim还是candidate  
    //admit是具体的比较规则，看下面  
    candidates--;  
    //如果candidate胜出则淘汰victim  
    if (admit(candidateKey, victimKey)) {  
      Node<K, V> evict = victim;  
      victim = victim.getNextInAccessOrder();  
      evictEntry(evict, RemovalCause.SIZE, 0L);  
      candidate = candidate.getPreviousInAccessOrder();  
    } else {  
      //如果是victim胜出，则淘汰candidate  
      Node<K, V> evict = candidate;  
      candidate = candidate.getPreviousInAccessOrder();  
      evictEntry(evict, RemovalCause.SIZE, 0L);  
    }  
  }  
}  
  
/**  
 * Determines if the candidate should be accepted into the main space, as determined by its  
 * frequency relative to the victim. A small amount of randomness is used to protect against hash  
 * collision attacks, where the victim's frequency is artificially raised so that no new entries  
 * are admitted.  
 *  
 * @param candidateKey the key for the entry being proposed for long term retention  
 * @param victimKey the key for the entry chosen by the eviction policy for replacement  
 * @return if the candidate should be admitted and the victim ejected  
 */  
@GuardedBy("evictionLock")  
boolean admit(K candidateKey, K victimKey) {  
  //分别获取victim和candidate的统计频率  
  //frequency这个方法的原理和实现上面已经解释了  
  int victimFreq = frequencySketch().frequency(victimKey);  
  int candidateFreq = frequencySketch().frequency(candidateKey);  
  //谁大谁赢  
  if (candidateFreq > victimFreq) {  
    return true;  
      
    //如果相等，candidate小于5都当输了  
  } else if (candidateFreq <= 5) {  
    // The maximum frequency is 15 and halved to 7 after a reset to age the history. An attack  
    // exploits that a hot candidate is rejected in favor of a hot victim. The threshold of a warm  
    // candidate reduces the number of random acceptances to minimize the impact on the hit rate.  
    return false;  
  }  
  //如果相等且candidate大于5，则随机淘汰一个  
  int random = ThreadLocalRandom.current().nextInt();  
  return ((random & 127) == 0);  
}  

//与上次命中率之差的阈值  
static final double HILL_CLIMBER_RESTART_THRESHOLD = 0.05d;  
//步长（调整）的大小（跟最大值maximum的比例）  
static final double HILL_CLIMBER_STEP_PERCENT = 0.0625d;  
//步长的衰减比例  
static final double HILL_CLIMBER_STEP_DECAY_RATE = 0.98d;  
  /** Adapts the eviction policy to towards the optimal recency / frequency configuration. */  
//climb方法的主要作用就是动态调整window区的大小（相应的，main区的大小也会发生变化，两个之和为100%）。  
//因为区域的大小发生了变化，那么区域内的数据也可能需要发生相应的移动。  
@GuardedBy("evictionLock")  
void climb() {  
  if (!evicts()) {  
    return;  
  }  
  //确定window需要调整的大小  
  determineAdjustment();  
  //如果protected区有溢出，把溢出部分移动到probation区。因为下面的操作有可能需要调整到protected区。  
  demoteFromMainProtected();  
  long amount = adjustment();  
  if (amount == 0) {  
    return;  
  } else if (amount > 0) {  
    //增加window的大小  
    increaseWindow();  
  } else {  
    //减少window的大小  
    decreaseWindow();  
  }  
}  

/** Calculates the amount to adapt the window by and sets {@link #adjustment()} accordingly. */  
@GuardedBy("evictionLock")  
void determineAdjustment() {  
  //如果frequencySketch还没初始化，则返回  
  if (frequencySketch().isNotInitialized()) {  
    setPreviousSampleHitRate(0.0);  
    setMissesInSample(0);  
    setHitsInSample(0);  
    return;  
  }  
  //总请求量 = 命中 + miss  
  int requestCount = hitsInSample() + missesInSample();  
  //没达到sampleSize则返回  
  //默认下sampleSize = 10 * maximum。用sampleSize来判断缓存是否足够”热“。  
  if (requestCount < frequencySketch().sampleSize) {  
    return;  
  }  
    
  //命中率的公式 = 命中 / 总请求  
  double hitRate = (double) hitsInSample() / requestCount;  
  //命中率的差值  
  double hitRateChange = hitRate - previousSampleHitRate();  
  //本次调整的大小，是由命中率的差值和上次的stepSize决定的  
  double amount = (hitRateChange >= 0) ? stepSize() : -stepSize();  
  //下次的调整大小：如果命中率的之差大于0.05，则重置为0.065 * maximum，否则按照0.98来进行衰减  
  double nextStepSize = (Math.abs(hitRateChange) >= HILL_CLIMBER_RESTART_THRESHOLD)  
      ? HILL_CLIMBER_STEP_PERCENT * maximum() * (amount >= 0 ? 1 : -1)  
      : HILL_CLIMBER_STEP_DECAY_RATE * amount;  
  setPreviousSampleHitRate(hitRate);  
  setAdjustment((long) amount);  
  setStepSize(nextStepSize);  
  setMissesInSample(0);  
  setHitsInSample(0);  
}  
  
/** Transfers the nodes from the protected to the probation region if it exceeds the maximum. */  
  
//这个方法比较简单，减少protected区溢出的部分  
@GuardedBy("evictionLock")  
void demoteFromMainProtected() {  
  long mainProtectedMaximum = mainProtectedMaximum();  
  long mainProtectedWeightedSize = mainProtectedWeightedSize();  
  if (mainProtectedWeightedSize <= mainProtectedMaximum) {  
    return;  
  }  
  
  for (int i = 0; i < QUEUE_TRANSFER_THRESHOLD; i++) {  
    if (mainProtectedWeightedSize <= mainProtectedMaximum) {  
      break;  
    }  
  
    Node<K, V> demoted = accessOrderProtectedDeque().poll();  
    if (demoted == null) {  
      break;  
    }  
    demoted.makeMainProbation();  
    accessOrderProbationDeque().add(demoted);  
    mainProtectedWeightedSize -= demoted.getPolicyWeight();  
  }  
  setMainProtectedWeightedSize(mainProtectedWeightedSize);  
}  
  
/**  
 * Increases the size of the admission window by shrinking the portion allocated to the main  
 * space. As the main space is partitioned into probation and protected regions (80% / 20%), for  
 * simplicity only the protected is reduced. If the regions exceed their maximums, this may cause  
 * protected items to be demoted to the probation region and probation items to be demoted to the  
 * admission window.  
 */  
  
//增加window区的大小，这个方法比较简单，思路就像我上面说的  
@GuardedBy("evictionLock")  
void increaseWindow() {  
  if (mainProtectedMaximum() == 0) {  
    return;  
  }  
  
  long quota = Math.min(adjustment(), mainProtectedMaximum());  
  setMainProtectedMaximum(mainProtectedMaximum() - quota);  
  setWindowMaximum(windowMaximum() + quota);  
  demoteFromMainProtected();  
  
  for (int i = 0; i < QUEUE_TRANSFER_THRESHOLD; i++) {  
    Node<K, V> candidate = accessOrderProbationDeque().peek();  
    boolean probation = true;  
    if ((candidate == null) || (quota < candidate.getPolicyWeight())) {  
      candidate = accessOrderProtectedDeque().peek();  
      probation = false;  
    }  
    if (candidate == null) {  
      break;  
    }  
  
    int weight = candidate.getPolicyWeight();  
    if (quota < weight) {  
      break;  
    }  
  
    quota -= weight;  
    if (probation) {  
      accessOrderProbationDeque().remove(candidate);  
    } else {  
      setMainProtectedWeightedSize(mainProtectedWeightedSize() - weight);  
      accessOrderProtectedDeque().remove(candidate);  
    }  
    setWindowWeightedSize(windowWeightedSize() + weight);  
    accessOrderWindowDeque().add(candidate);  
    candidate.makeWindow();  
  }  
  
  setMainProtectedMaximum(mainProtectedMaximum() + quota);  
  setWindowMaximum(windowMaximum() - quota);  
  setAdjustment(quota);  
}  
  
/** Decreases the size of the admission window and increases the main's protected region. */  
//同上increaseWindow差不多，反操作  
@GuardedBy("evictionLock")  
void decreaseWindow() {  
  if (windowMaximum() <= 1) {  
    return;  
  }  
  
  long quota = Math.min(-adjustment(), Math.max(0, windowMaximum() - 1));  
  setMainProtectedMaximum(mainProtectedMaximum() + quota);  
  setWindowMaximum(windowMaximum() - quota);  
  
  for (int i = 0; i < QUEUE_TRANSFER_THRESHOLD; i++) {  
    Node<K, V> candidate = accessOrderWindowDeque().peek();  
    if (candidate == null) {  
      break;  
    }  
  
    int weight = candidate.getPolicyWeight();  
    if (quota < weight) {  
      break;  
    }  
  
    quota -= weight;  
    setMainProtectedWeightedSize(mainProtectedWeightedSize() + weight);  
    setWindowWeightedSize(windowWeightedSize() - weight);  
    accessOrderWindowDeque().remove(candidate);  
    accessOrderProbationDeque().add(candidate);  
    candidate.makeMainProbation();  
  }  
  
  setMainProtectedMaximum(mainProtectedMaximum() - quota);  
  setWindowMaximum(windowMaximum() + quota);  
  setAdjustment(-quota);  
}  

final Buffer<Node<K, V>> readBuffer;  
  
readBuffer = evicts() || collectKeys() || collectValues() || expiresAfterAccess()  
        ? new BoundedBuffer<>()  
        : Buffer.disabled();  

/**  
 * Performs the post-processing work required after a read.  
 *  
 * @param node the entry in the page replacement policy  
 * @param now the current time, in nanoseconds  
 * @param recordHit if the hit count should be incremented  
 */  
void afterRead(Node<K, V> node, long now, boolean recordHit) {  
  if (recordHit) {  
    statsCounter().recordHits(1);  
  }  
  //把记录加入到readBuffer  
  //判断是否需要立即处理readBuffer  
  //注意这里无论offer是否成功都可以走下去的，即允许写入readBuffer丢失，因为这个  
  boolean delayable = skipReadBuffer() || (readBuffer.offer(node) != Buffer.FULL);  
  if (shouldDrainBuffers(delayable)) {  
    scheduleDrainBuffers();  
  }  
  refreshIfNeeded(node, now);  
}  
  
 /**  
   * Returns whether maintenance work is needed.  
   *  
   * @param delayable if draining the read buffer can be delayed  
   */  
  
  //caffeine用了一组状态来定义和管理“维护”的过程  
  boolean shouldDrainBuffers(boolean delayable) {  
    switch (drainStatus()) {  
      case IDLE:  
        return !delayable;  
      case REQUIRED:  
        return true;  
      case PROCESSING_TO_IDLE:  
      case PROCESSING_TO_REQUIRED:  
        return false;  
      default:  
        throw new IllegalStateException();  
    }  
  }  

/**  
 * A striped, non-blocking, bounded buffer.  
 *  
 * @author ben.manes@gmail.com (Ben Manes)  
 * @param <E> the type of elements maintained by this buffer  
 */  
final class BoundedBuffer<E> extends StripedBuffer<E>  

/**  
 * A base class providing the mechanics for supporting dynamic striping of bounded buffers. This  
 * implementation is an adaption of the numeric 64-bit {@link java.util.concurrent.atomic.Striped64}  
 * class, which is used by atomic counters. The approach was modified to lazily grow an array of  
 * buffers in order to minimize memory usage for caches that are not heavily contended on.  
 *  
 * @author dl@cs.oswego.edu (Doug Lea)  
 * @author ben.manes@gmail.com (Ben Manes)  
 */  
  
abstract class StripedBuffer<E> implements Buffer<E>  

/** Table of buffers. When non-null, size is a power of 2. */  
//RingBuffer数组  
transient volatile Buffer<E> @Nullable[] table;  
  
//当进行resize时，需要整个table锁住。tableBusy作为CAS的标记。  
static final long TABLE_BUSY = UnsafeAccess.objectFieldOffset(StripedBuffer.class, "tableBusy");  
static final long PROBE = UnsafeAccess.objectFieldOffset(Thread.class, "threadLocalRandomProbe");  
  
/** Number of CPUS. */  
static final int NCPU = Runtime.getRuntime().availableProcessors();  
  
/** The bound on the table size. */  
//table最大size  
static final int MAXIMUM_TABLE_SIZE = 4 * ceilingNextPowerOfTwo(NCPU);  
  
/** The maximum number of attempts when trying to expand the table. */  
//如果发生竞争时（CAS失败）的尝试次数  
static final int ATTEMPTS = 3;  
  
/** Table of buffers. When non-null, size is a power of 2. */  
//核心数据结构  
transient volatile Buffer<E> @Nullable[] table;  
  
/** Spinlock (locked via CAS) used when resizing and/or creating Buffers. */  
transient volatile int tableBusy;  
  
/** CASes the tableBusy field from 0 to 1 to acquire lock. */  
final boolean casTableBusy() {  
  return UnsafeAccess.UNSAFE.compareAndSwapInt(this, TABLE_BUSY, 0, 1);  
}  
  
/**  
 * Returns the probe value for the current thread. Duplicated from ThreadLocalRandom because of  
 * packaging restrictions.  
 */  
static final int getProbe() {  
  return UnsafeAccess.UNSAFE.getInt(Thread.currentThread(), PROBE);  
}  

@Override  
public int offer(E e) {  
  int mask;  
  int result = 0;  
  Buffer<E> buffer;  
  //是否不存在竞争  
  boolean uncontended = true;  
  Buffer<E>[] buffers = table  
  //是否已经初始化  
  if ((buffers == null)  
      || (mask = buffers.length - 1) < 0  
      //用thread的随机值作为hash值，得到对应位置的RingBuffer  
      || (buffer = buffers[getProbe() & mask]) == null  
      //检查追加到RingBuffer是否成功  
      || !(uncontended = ((result = buffer.offer(e)) != Buffer.FAILED))) {  
    //其中一个符合条件则进行扩容  
    expandOrRetry(e, uncontended);  
  }  
  return result;  
}  
  
/**  
 * Handles cases of updates involving initialization, resizing, creating new Buffers, and/or  
 * contention. See above for explanation. This method suffers the usual non-modularity problems of  
 * optimistic retry code, relying on rechecked sets of reads.  
 *  
 * @param e the element to add  
 * @param wasUncontended false if CAS failed before call  
 */  
  
//这个方法比较长，但思路还是相对清晰的。  
@SuppressWarnings("PMD.ConfusingTernary")  
final void expandOrRetry(E e, boolean wasUncontended) {  
  int h;  
  if ((h = getProbe()) == 0) {  
    ThreadLocalRandom.current(); // force initialization  
    h = getProbe();  
    wasUncontended = true;  
  }  
  boolean collide = false; // True if last slot nonempty  
  for (int attempt = 0; attempt < ATTEMPTS; attempt++) {  
    Buffer<E>[] buffers;  
    Buffer<E> buffer;  
    int n;  
    if (((buffers = table) != null) && ((n = buffers.length) > 0)) {  
      if ((buffer = buffers[(n - 1) & h]) == null) {  
        if ((tableBusy == 0) && casTableBusy()) { // Try to attach new Buffer  
          boolean created = false;  
          try { // Recheck under lock  
            Buffer<E>[] rs;  
            int mask, j;  
            if (((rs = table) != null) && ((mask = rs.length) > 0)  
                && (rs[j = (mask - 1) & h] == null)) {  
              rs[j] = create(e);  
              created = true;  
            }  
          } finally {  
            tableBusy = 0;  
          }  
          if (created) {  
            break;  
          }  
          continue; // Slot is now non-empty  
        }  
        collide = false;  
      } else if (!wasUncontended) { // CAS already known to fail  
        wasUncontended = true;      // Continue after rehash  
      } else if (buffer.offer(e) != Buffer.FAILED) {  
        break;  
      } else if (n >= MAXIMUM_TABLE_SIZE || table != buffers) {  
        collide = false; // At max size or stale  
      } else if (!collide) {  
        collide = true;  
      } else if (tableBusy == 0 && casTableBusy()) {  
        try {  
          if (table == buffers) { // Expand table unless stale  
            table = Arrays.copyOf(buffers, n << 1);  
          }  
        } finally {  
          tableBusy = 0;  
        }  
        collide = false;  
        continue; // Retry with expanded table  
      }  
      h = advanceProbe(h);  
    } else if ((tableBusy == 0) && (table == buffers) && casTableBusy()) {  
      boolean init = false;  
      try { // Initialize table  
        if (table == buffers) {  
          @SuppressWarnings({"unchecked", "rawtypes"})  
          Buffer<E>[] rs = new Buffer[1];  
          rs[0] = create(e);  
          table = rs;  
          init = true;  
        }  
      } finally {  
        tableBusy = 0;  
      }  
      if (init) {  
        break;  
      }  
    }  
  }  
}  

/** The maximum number of elements per buffer. */  
static final int BUFFER_SIZE = 16;  
  
// Assume 4-byte references and 64-byte cache line (16 elements per line)  
//256长度，但是是以16为单位，所以最多存放16个元素  
static final int SPACED_SIZE = BUFFER_SIZE << 4;  
static final int SPACED_MASK = SPACED_SIZE - 1;  
static final int OFFSET = 16;   
//RingBuffer数组  
final AtomicReferenceArray<E> buffer;  
  
 //插入方法  
 @Override  
 public int offer(E e) {  
   long head = readCounter;  
   long tail = relaxedWriteCounter();  
   //用head和tail来限制个数  
   long size = (tail - head);  
   if (size >= SPACED_SIZE) {  
     return Buffer.FULL;  
   }  
   //tail追加16  
   if (casWriteCounter(tail, tail + OFFSET)) {  
     //用tail“取余”得到下标  
     int index = (int) (tail & SPACED_MASK);  
     //用unsafe.putOrderedObject设值  
     buffer.lazySet(index, e);  
     return Buffer.SUCCESS;  
   }  
   //如果CAS失败则返回失败  
   return Buffer.FAILED;  
 }  
  
 //用consumer来处理buffer的数据  
 @Override  
 public void drainTo(Consumer<E> consumer) {  
   long head = readCounter;  
   long tail = relaxedWriteCounter();  
   //判断数据多少  
   long size = (tail - head);  
   if (size == 0) {  
     return;  
   }  
   do {  
     int index = (int) (head & SPACED_MASK);  
     E e = buffer.get(index);  
     if (e == null) {  
       // not published yet  
       break;  
     }  
     buffer.lazySet(index, null);  
     consumer.accept(e);  
     //head也跟tail一样，每次递增16  
     head += OFFSET;  
   } while (head != tail);  
   lazySetReadCounter(head);  
 }  

private static LoadingCache<String, String> cache = Caffeine.newBuilder()  
        .maximumSize(256L)  
        .initialCapacity(1)  
        //.expireAfterAccess(2, TimeUnit.DAYS)  
        //.expireAfterWrite(2, TimeUnit.HOURS)  
        .refreshAfterWrite(1, TimeUnit.HOURS)  
        //自定义过期时间  
        .expireAfter(new Expiry<String, String>() {  
            //返回创建后的过期时间  
            @Override  
            public long expireAfterCreate(@NonNull String key, @NonNull String value, long currentTime) {  
                return 0;  
            }  
  
            //返回更新后的过期时间  
            @Override  
            public long expireAfterUpdate(@NonNull String key, @NonNull String value, long currentTime, @NonNegative long currentDuration) {  
                return 0;  
            }  
  
            //返回读取后的过期时间  
            @Override  
            public long expireAfterRead(@NonNull String key, @NonNull String value, long currentTime, @NonNegative long currentDuration) {  
                return 0;  
            }  
        })  
        .recordStats()  
        .build(new CacheLoader<String, String>() {  
            @Nullable  
            @Override  
            public String load(@NonNull String key) throws Exception {  
                return "value_" + key;  
            }  
        });  

/**  
 * Schedules a timer event for the node.  
 *  
 * @param node the entry in the cache  
 */  
public void schedule(@NonNull Node<K, V> node) {  
  Node<K, V> sentinel = findBucket(node.getVariableTime());  
  link(sentinel, node);  
}  
  
/**  
 * Determines the bucket that the timer event should be added to.  
 *  
 * @param time the time when the event fires  
 * @return the sentinel at the head of the bucket  
 */  
Node<K, V> findBucket(long time) {  
  long duration = time - nanos;  
  int length = wheel.length - 1;  
  for (int i = 0; i < length; i++) {  
    if (duration < SPANS[i + 1]) {  
      long ticks = (time >>> SHIFT[i]);  
      int index = (int) (ticks & (wheel[i].length - 1));  
      return wheel[i][index];  
    }  
  }  
  return wheel[length][0];  
}  
  
/** Adds the entry at the tail of the bucket's list. */  
void link(Node<K, V> sentinel, Node<K, V> node) {  
  node.setPreviousInVariableOrder(sentinel.getPreviousInVariableOrder());  
  node.setNextInVariableOrder(sentinel);  
  
  sentinel.getPreviousInVariableOrder().setNextInVariableOrder(node);  
  sentinel.setPreviousInVariableOrder(node);  
}