Caffeine高性能缓存设计

川长思鸟来 2023-05-22 12:29 6阅读 0赞

### Caffeine高性能缓存设计 ###

*  是否需要缓存
 *  选择合适的缓存
 *  Caffeine 的使用
 *  Caffeine高性能设计
 *   *  W-TinyLFU 淘汰算法的整体设计
     *   *  LRU和LRU的缺点
         *  TinyLFU
         *   *  统计频率Count–Min Sketch算法
             *  频率统计Count–Min Sketch的保新机制
         *  增加一个小window
         *  淘汰策略
         *   *  expireEntries 方法
             *  evictEntries 方法
         *  Pacer in TinyLFU
     *  异步的高性能读写
     *   *  异步读 ReadBuffer
         *  异步写 WriteBuffer
         *  异步PerformCleanupTask
     *  TimeWheel设计
     *  整体设计架构

Caffeine是一个高性能，高命中率，低内存占用，near optimal 的本地缓存。Caffeine被普遍称为“现代缓存之王”。本文将重点讲解Caffeine的高性能设计，以及对应部分的源码分析。

本文基于 2.8.1 源码分析

# 是否需要缓存 #

在使用缓存之前，首先需要确认你的项目是否真的需要缓存。使用缓存会引入的一定的技术复杂度，后文也将会一一介绍这些复杂度。一般来说从两个方面来个是否需要使用缓存:

1.  CPU占用: 如果你有某些应用需要消耗大量的cpu去计算，比如正则表达式，如果你使用正则表达式比较频繁，而其又占用了很多CPU的话，那你就应该使用缓存将正则表达式的结果给缓存下来。
2.  数据库或则网络IO占用: 如果你发现你的数据库连接池比较空闲，那么不应该用缓存。但是如果数据库连接池比较繁忙，甚至经常报出连接不够的报警，那么是时候应该考虑缓存了。

如果并没有上述两个问题，那么你不必为了增加缓存而缓存。

# 选择合适的缓存 #

缓存分为本地缓存和分布式缓存两种。

对于本地缓存来说，如果不需要淘汰算法则选择ConcurrentHashMap，如果需要淘汰算法和一些丰富的API，这里推荐选择Caffeine。

分布式缓存，这里就不介绍了。

实际应用系统一般都会有多级缓存。

# Caffeine 的使用 #

Caffeine的API和Guava非常的相似，下面给出一个创建cache的例子：

package com.example.caffeine;
    
    import java.util.concurrent.TimeUnit;
    
    import com.github.benmanes.caffeine.cache.Cache;
    import com.github.benmanes.caffeine.cache.CacheLoader;
    import com.github.benmanes.caffeine.cache.CacheWriter;
    import com.github.benmanes.caffeine.cache.Caffeine;
    import com.github.benmanes.caffeine.cache.LoadingCache;
    import com.github.benmanes.caffeine.cache.RemovalCause;
    import com.sun.istack.internal.Nullable;
    import org.checkerframework.checker.nullness.qual.NonNull;
    
    public class Demo { 
        public static void main(String[] args) { 
            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) { 
                        System.out.printf("key={}, CacheWriter write", key);
                    }
    
                    @Override
                    public void delete(@NonNull Object key, @Nullable Object value, @NonNull RemovalCause cause) { 
                        System.out.printf("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;
                    }
                });
    
            cache.put("aaa", "abdunoias");
            cache.get("aaa");
            cache.getIfPresent("bnsodapo,");
        }
    }

这个API的设计和Guava非常像。

# Caffeine高性能设计 #

判断一个本地缓存的好坏最核心指标就是命中率和内存占用，影响命中率的因素有很多，比如业务场景，淘汰策略，清理策略，缓存容量

## W-TinyLFU 淘汰算法的整体设计 ##

淘汰策略是影响缓存命中率的很重要的因素，我们常用的有LRU或则LFU。W-TinyLFU很明显是一种变种的 LFU 的淘汰算法。

### LRU和LRU的缺点 ###

LRU实现非常简单，性能也非常好。LRU对突发的稀疏流量（sparse bursts）表现很好，但同时也会产生缓存污染，比如偶然性的要对全量数据进行遍历，那么“历史访问记录”就会被刷走，造成污染。也就是冷数据会顶掉热数据

如果数据的分布在一段时间内是固定的话，那么LFU可以达到最高的命中率。但是有两个很大的缺点：

1.  维护每个记录项的频率信息，这是个巨大的开销；
2.  对突发性的稀疏流量无力。

针对LRU和LFU都有很多改良算法，比如基于LRU的ARC等。

### TinyLFU ###

TinyLFU就是基于LFU的改良算法。

解决LFU的第一个缺点是采用了Count–Min Sketch算法。

解决LFU的第二个缺点是让记录尽量保持相对的“新鲜”（Freshness Mechanism），并且当有新的记录插入时，可以让它跟老的记录进行“PK”，输者就会被淘汰，这样一些老的、不再需要的记录就会被剔除。

下图是TinyLFU设计图：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA4NTMyNjE_size_16_color_FFFFFF_t_70]

#### 统计频率Count–Min Sketch算法 ####

统计频率的核心问题就是如果既可以对一个key进行统计，但是又可以节省空间。简单的hashmap肯定是不行的，这太消耗内存。对于缓存key的统计来说，这里的统计并不需要非常精确，只需要一个近似值就可以了。这个和Bloom Filter似乎非常相似，只不过Bloom Filter统计的是true或则false。Count–Min Sketch的原理跟Bloom Filter一样，只不过Bloom Filter只有0和1的值，那么你可以把Count–Min Sketch看作是“数值”版的Bloom Filter。

TODO：Count–Min Sketch实现的细节

#### 频率统计Count–Min Sketch的保新机制 ####

保新机制是为了让缓存保持“新鲜”，剔除掉过往频率很高但之后不经常使用的缓存，Caffeine有一个Freshness Mechanism。做法很简答，就是当整体的统计计数（当前所有记录的频率统计之和，这个数值内部维护）达到某一个值时，那么所有记录的频率统计除以2。

TODO：Count–Min Sketch关于reset的实现细节

### 增加一个小window ###

Caffeine通过测试发现 TinyLFU 在面对突发性的稀疏流量（sparse bursts）时表现很差，因为新的记录（new items）还没来得及建立足够的频率就被剔除出去了，这就使得命中率下降。

于是Caffeine设计出一种新的policy，即Window Tiny LFU（W-TinyLFU），并通过实验和实践发现W-TinyLFU比TinyLFU表现的更好。

W-TinyLFU的设计如下所示：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA4NTMyNjE_size_16_color_FFFFFF_t_70 1]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA4NTMyNjE_size_16_color_FFFFFF_t_70 2]  
它主要包括两个缓存模块，主缓存是SLRU（Segmented LRU，即分段LRU），SLRU包括一个名为protected和一个名为probation的缓存区。通过增加一个缓存区（即Window Cache），当有新的记录插入时，会先在window区呆一下，就可以避免上述说的sparse bursts问题。

### 淘汰策略 ###

Caffeine 中所有的缓存数据都存储在ConcurrentHashMap中。在caffeine中有三个记录引用的LRU队列:

1.  Eden队列(Window): caffeine中规定只能为缓存容量的%1，如果size=100，那这个队列的有效大小就等于1。这个队列中记录的是新到的数据，防止突发流量由于之前没有访问频率，而导致被淘汰。比如有一部新剧上线，在最开始其实是没有访问频率的，防止上线之后被其他缓存淘汰出去，而加入这个区域。Eden区最舒服最安逸的区域，在这里很难被其他数据淘汰。
2.  Probation队列: 叫做缓刑队列，在这个队列就代表你的数据相对比较冷，马上就要被淘汰了。这个有效大小为size减去eden减去protected。
3.  Protected队列: 在这个队列中，可以稍微放心一下了，你暂时不会被淘汰，但是别急，如果Probation队列没有数据了或者Protected数据满了，你也将会被面临淘汰的尴尬局面。当然想要变成这个队列，需要把Probation访问一次之后，就会提升为Protected队列。这个有效大小为(size减去eden) X 80% 如果size =100，就会是79。

经过实验测试，上面列出的Eden队列占比1%，剩余的99%当中的80%分给protected区，20%分给probation区时，这时整体性能和命中率表现得最好，所以Caffeine默认的比例设置就是这个。

不过这个比例Caffeine会在运行时根据统计数据（statistics）去动态调整，如果你的应用程序的缓存随着时间变化比较快的话，那么增加window区的比例可以提高命中率，相反缓存都是比较固定不变的话，增加Main Cache区（protected区 +probation区）的比例会有较好的效果。具体这块实现在后面的Pacer中介绍。

Caffeine的缓存淘汰策略就是基于这三个队列做的。Caffeine的淘汰策略都包含在函数 `maintenance` 中。

`maintenance`的调用大部分情况下都会在 `PerformCleanupTask` 里面run。提交`PerformCleanupTask` 这个 Runnable 的Task的场景很多，最主要是在 `afterRead` 或则 `afterWrite` 之后。

这里主要来分析一下`maintenance` 函数主要做了什么，这里只先说跟“淘汰策略”有关的`expireEntries`和`evictEntries` 函数。

@GuardedBy("evictionLock")
    void maintenance(@Nullable Runnable task) { 
    	// 异步读写处理
    	drainReadBuffer();
    	drainWriteBuffer();
    	drainKeyReferences();
    	drainValueReferences();
    	
    	// 淘汰策略
    	expireEntries();
    	evictEntries();
    	
    	// pacer
    	climb();
    }

根据注释，该函数会执行挂起的维护工作，并在处理期间设置状态标志，以避免过多的调度尝试。执行之后读缓冲区、写缓冲区和引用队列会被耗尽，然后是过期和基于大小的回收。

先介绍一下Caffeine对上面说到的W-TinyLFU策略的实现用到的数据结构：

//最大的个数限制
    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. */
    static final int QUEUE_TRANSFER_THRESHOLD = 1_000;

#### expireEntries 方法 ####

expireEntries 主要是基于Access、Write、或则是variable过期entries的。比如Access后一小时过期.

TODO

#### evictEntries 方法 ####

evictEntries用于window 区的size超过了其最大的capacity之后来 evict entries。

@GuardedBy("evictionLock")
    void evictEntries() { 
    	if (!evicts()) { 
    		return;
    	}
    	// 淘汰window区的记录, 返回的是淘汰的记录数。
    	int candidates = evictFromWindow();
    	// 淘汰Main区的记录
    	evictFromMain(candidates);
    }

当Window区域size超过了其最大值时候，从window区域淘汰元素到Main区域。

//根据W-TinyLFU，新的数据都会无条件的加到admission window
    //但是window是有大小限制，所以要“定期”做一下“维护”
    @GuardedBy("evictionLock")
    int evictFromWindow() { 
      int candidates = 0;
      //获取window区域的头部节点
      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.getPolicyWeight() != 0) { 
        	//把node定位在probation区
        	//然后从window区去掉，并加到probation区，相当于把节点移动到probation区（晋升了）
          node.makeMainProbation();
          accessOrderWindowDeque().remove(node);
          accessOrderProbationDeque().add(node);
          candidates++;
    //因为移除了一个节点，所以需要调整window的size
          setWindowWeightedSize(windowWeightedSize() - node.getPolicyWeight());
        }
        node = next;
      }
    
      return candidates;
    }

淘汰Main区的记录的`evictFromMain`方法：

//根据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为null的处理
        if ((candidate == null) && (victim == null)) { 
          if (victimQueue == PROBATION) { 
          	// 当PROBATION区为空的时候，就切换到从PROTECTED区去取元素做淘汰PK
            victim = accessOrderProtectedDeque().peekFirst();
            victimQueue = PROTECTED;
            continue;
          } else if (victimQueue == PROTECTED) { 
          	// Main区整个都空的时候，从WINDOW区去取元素做淘汰PK
            victim = accessOrderWindowDeque().peekFirst();
            victimQueue = WINDOW;
            continue;
          }
    
          // The pending operations will adjust the size to reflect the correct weight
          break;
        }
    
        //忽略weight为0的victim节点
        if ((victim != null) && (victim.getPolicyWeight() == 0)) { 
          victim = victim.getNextInAccessOrder();
          continue;
        } else if ((candidate != null) && (candidate.getPolicyWeight() == 0)) { 
          //对weight为0的candidate处理
          candidate = candidate.getPreviousInAccessOrder();
          candidates--;
          continue;
        }
    
        // Evict immediately if only one of the entries is present
        if (victim == null) { 
          @SuppressWarnings("NullAway")
          // 直接淘汰 candidate
          Node<K, V> previous = candidate.getPreviousInAccessOrder();
          Node<K, V> evict = candidate;
          candidate = previous;
          candidates--;
          evictEntry(evict, RemovalCause.SIZE, 0L);
          continue;
        } else if (candidate == null) { 
          // 直接淘汰 victim
          Node<K, V> evict = victim;
          victim = victim.getNextInAccessOrder();
          evictEntry(evict, RemovalCause.SIZE, 0L);
          continue;
        }
    
        // 当entry的key已经因为引用关系被回收时候，直接evict entry。
        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--;
        if (admit(candidateKey, victimKey)) { 
          //如果candidate胜出则淘汰victim
          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);
        }
      }
    }

`admit` 函数就是 `Count–Min Sketch` 获取key的频率然后根据频率做PK的逻辑。具体可以参考Count–Min Sketch的实现部分。

上面的函数 evict entry都是通过调用 `evictEntry` 函数来实现的。这里evictEntry的cause有两种，一个是因为GC已经回收没有引用关系的entry，一个是因为缓存的size。

@GuardedBy("evictionLock")
    // 尝试去 evict entry，如果entry已经有了一些update，这次更新可能会忽略。
    boolean evictEntry(Node<K, V> node, RemovalCause cause, long now) { 
      K key = node.getKey();
      V[] value = (V[]) new Object[1];
      // 标记entry是否已经被remove
      boolean[] removed = new boolean[1];
      // 标记entry是否已经复活。
      boolean[] resurrect = new boolean[1];
      RemovalCause[] actualCause = new RemovalCause[1];
    
      // 如果entry的key还存在于缓存中，就通过mappingFunction函数重新计算新值并替换旧值。
      // 如果entry已经复活了，就会直接return，否则就会将evict entry(从缓存中删除)。
      data.computeIfPresent(node.getKeyReference(), (k, n) -> { 
        if (n != node) { 
          return n;
        }
        synchronized (n) { 
          value[0] = n.getValue();
    
          actualCause[0] = (key == null) || (value[0] == null) ? RemovalCause.COLLECTED : cause;
          if (actualCause[0] == RemovalCause.EXPIRED) { 
            boolean expired = false;
            if (expiresAfterAccess()) { 
              expired |= ((now - n.getAccessTime()) >= expiresAfterAccessNanos());
            }
            if (expiresAfterWrite()) { 
              expired |= ((now - n.getWriteTime()) >= expiresAfterWriteNanos());
            }
            if (expiresVariable()) { 
              expired |= (n.getVariableTime() <= now);
            }
            if (!expired) { 
              resurrect[0] = true;
              return n;
            }
          } else if (actualCause[0] == RemovalCause.SIZE) { 
            int weight = node.getWeight();
            if (weight == 0) { 
              resurrect[0] = true;
              return n;
            }
          }
    
          if (key != null) { 
            writer.delete(key, value[0], actualCause[0]);
          }
          makeDead(n);
        }
        removed[0] = true;
        return null;
      });
    
      // entry已经复活，放弃这次evict
      if (resurrect[0]) { 
        return false;
      }
    
      // If the eviction fails due to a concurrent removal of the victim, that removal may cancel out
      // the addition that triggered this eviction. The victim is eagerly unlinked before the removal
      // task so that if an eviction is still required then a new victim will be chosen for removal.
      if (node.inWindow() && (evicts() || expiresAfterAccess())) { 
        accessOrderWindowDeque().remove(node);
      } else if (evicts()) { 
        if (node.inMainProbation()) { 
          accessOrderProbationDeque().remove(node);
        } else { 
          accessOrderProtectedDeque().remove(node);
        }
      }
      if (expiresAfterWrite()) { 
        writeOrderDeque().remove(node);
      } else if (expiresVariable()) { 
        timerWheel().deschedule(node);
      }
    
      if (removed[0]) { 
        statsCounter().recordEviction(node.getWeight(), actualCause[0]);
        if (hasRemovalListener()) { 
          // Notify the listener only if the entry was evicted. This must be performed as the last
          // step during eviction to safe guard against the executor rejecting the notification task.
          notifyRemoval(key, value[0], actualCause[0]);
        }
      } else { 
        // Eagerly decrement the size to potentially avoid an additional eviction, rather than wait
        // for the removal task to do it on the next maintenance cycle.
        makeDead(node);
      }
    
      return true;
    }

这里可以看出，当我们evict一个entry时候，会将其从cache中删除kv。

### Pacer in TinyLFU ###

`climb` 方法主要是用来调整window size的，使得Caffeine可以适应你的应用类型（如OLAP或OLTP）表现出最佳的命中率。这个也就是TinyLFU中的Pacer。

调整时用到的默认比例数据：

//与上次命中率之差的阈值
    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;

官方测试的数据：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA4NTMyNjE_size_16_color_FFFFFF_t_70 3]  
我们看看window size的调整是怎么实现的:

//根据淘汰策略，调整最优的recent/frequency的配置
    @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();
      }
    }

下面分别分析上面的各个函数实现：

`determineAdjustment`函数计算调整的数值；

@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;
      //当前命中率和前一次命中率的diff
      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);
    }

`demoteFromMainProtected`：如果Protected区weight超过最大值，则将节点从Protected区域转移到Probation区域。

@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);
    }

`increaseWindow` 和 `decreaseWindow` 分别是增加或则减少Window区域，代码很好理解，这里就不描述了。

这里需要强调的是，当increaseWindow时候，会优先从Probation区域取出entry，然后移动到window区域。当decreaseWindow时候，从window区域取出的entry也是会优先移动到Probation区域。

## 异步的高性能读写 ##

缓存进行读写之后（读的话，已经存在则直接返回，不存在则load数据，保存，再返回；写的话，则直接插入或更新），需要维护一些淘汰策略相关的额外数据，诸如：

1.  计算和比较数据的是否过期
2.  统计频率（像LFU或其变种）
3.  维护read queue和write queue
4.  淘汰符合条件的数据
5.  等等。。。

缓存数据的读写都伴随着缓存状态的变更，Guava Cache的做法是把缓存状态的变更和读写操作一起同步处理，在一个同步加锁的操作中完成。虽然Guava Cache巧妙地利用了JDK的ConcurrentHashMap（JDk7的分段锁或者JDK8的无锁CAS）来降低锁的密度，达到提高并发度的目的。但是，对于一些热点数据，这种做法还是避免不了频繁的锁竞争。

Caffeine借鉴了数据库系统的WAL（Write-Ahead Logging）思想，即先写日志再执行操作，这种思想同样适合缓存。在执行读写操作时，先把操作记录在缓冲区，然后在合适的时机异步、批量地执行缓冲区中的内容。但在执行缓冲区的内容时，也是需要在缓冲区加上同步锁的，不然存在并发问题，只不过这样就可以把 `对锁的竞争从缓存数据转移到对缓冲区上`。

cache的访问主要是读写。所以buffer主要包含：ReadBuffer和WriteBuffer。

### 异步读 ReadBuffer ###

Caffeine内部为了支持不同的Features（如Eviction，Removal，Refresh，Statistics，Cleanup，Policy等等），自动扩展了很多子类，它们共同的父类是BoundedLocalCache，而readBuffer就是作为它们共有的属性，即都是用一样的readBuffer，看定义：

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

`Buffer` 是一个`多producer/单consumer缓冲区`，如果它是满的或者由于争用而错误地失败，Buffer会拒绝新元素。与队列和堆栈不同，缓冲区不保证FIFO或LIFO顺序中的元素排序。请注意，调用方有责任确保consumer对缓冲区具有独占的读访问权。

`BoundedBuffer` 是一个分割的，非阻塞的，有界缓冲区，继承于StripedBuffer类。这个StripedBuffer设计的思想是跟Striped64类似的，通过扩展结构把竞争热点分离。

具体实现是这样的，StripedBuffer维护一个Buffer\[\]数组，每个元素就是一个RingBuffer，每个线程用自己threadLocalRandomProbe属性作为hash值，这样就相当于每个线程都有自己“专属”的RingBuffer，就不会产生竞争啦，而不是用key的hashCode作为hash值，因为会产生热点数据问题。

StripedBuffer 属性：

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();
    
    //table最大size
    /** The bound on the 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;
    
    //RingBuffer数组
    /** Table of buffers. When non-null, size is a power of 2. */
    transient volatile Buffer<E> @Nullable[] table;
    
    //当进行resize时，需要整个table锁住。tableBusy作为CAS的标记。
    /** Spinlock (locked via CAS) used when resizing and/or creating Buffers. */
    transient volatile int tableBusy;

看一下StripedBuffer最核心方法： `offer`：

@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;
    }

offer方法，当没初始化或存在竞争时，则扩容为2倍。实际是调用RingBuffer的offer方法，把数据追加到RingBuffer后面。

扩容的方法：

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;
    	    }
    	  }
    	}
    }

扩容方法比较简单，table的长度限制了就是2的次幂，所以这里扩容两倍。

最后看看RingBuffer，注意RingBuffer是BoundedBuffer的内部类。

注意，RingBuffer的size（固定是16个）是不变的，变的是head和tail而已。

/** 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;
    
    /** Enforces a memory layout to avoid false sharing by padding the read count. */
    abstract static class ReadCounterRef extends PadReadCounter { 
      static final long READ_OFFSET =
          UnsafeAccess.objectFieldOffset(ReadCounterRef.class, "readCounter");
    
      volatile long readCounter;
    
      void lazySetReadCounter(long count) { 
        UnsafeAccess.UNSAFE.putOrderedLong(this, READ_OFFSET, count);
      }
    }
    
    abstract static class PadWriteCounter extends ReadCounterRef { 
      long p20, p21, p22, p23, p24, p25, p26, p27;
      long p30, p31, p32, p33, p34, p35, p36;
    }
      
    abstract static class ReadAndWriteCounterRef extends PadWriteCounter { 
      static final long WRITE_OFFSET =
          UnsafeAccess.objectFieldOffset(ReadAndWriteCounterRef.class, "writeCounter");
    
      volatile long writeCounter;
    
      ReadAndWriteCounterRef() { 
        UnsafeAccess.UNSAFE.putOrderedLong(this, WRITE_OFFSET, OFFSET);
      }
    
      long relaxedWriteCounter() { 
        return UnsafeAccess.UNSAFE.getLong(this, WRITE_OFFSET);
      }
    
      boolean casWriteCounter(long expect, long update) { 
        return UnsafeAccess.UNSAFE.compareAndSwapLong(this, WRITE_OFFSET, expect, update);
      }
    }
      
      
    static final class RingBuffer<E> extends BBHeader.ReadAndWriteCounterRef implements Buffer<E> { 
      final AtomicReferenceArray<E> buffer;
    
      @SuppressWarnings({ "unchecked", "cast", "rawtypes"})
      public RingBuffer(E e) { 
        buffer = new AtomicReferenceArray<>(SPACED_SIZE);
        buffer.lazySet(0, e);
      }
    
      //插入方法
      @Override
      public int offer(E e) { 
        long head = readCounter;
        long tail = relaxedWriteCounter();
        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;
        }
        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);
      }
    
      @Override
      public int reads() { 
        return (int) readCounter / OFFSET;
      }
    
      @Override
      public int writes() { 
        return (int) writeCounter / OFFSET;
      }
    }

总的来说ReadBuffer有如下特点：

1.  使用 Striped-RingBuffer来提升对buffer的并发读写
2.  用thread的hash来避开热点key的竞争
3.  允许写入的丢失

Caffeine每次对缓存的读操作都会触发`afterRead`函数：

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 */
    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();
      }
    }

caffeine用了一组状态来定义和管理“维护”的过程：

/** A drain is not taking place. */
    static final int IDLE = 0;
    /** A drain is required due to a pending write modification. */
    static final int REQUIRED = 1;
    /** A drain is in progress and will transition to idle. */
    static final int PROCESSING_TO_IDLE = 2;
    /** A drain is in progress and will transition to required. */
    static final int PROCESSING_TO_REQUIRED = 3;
    
    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();
      }
    }

如果需要Drain Buffer，就会调用`scheduleDrainBuffers` 提交异步任务去执行Drain Buffer.

/**
    * Attempts to schedule an asynchronous task to apply the pending operations to the page
     * replacement policy. If the executor rejects the task then it is run directly.
     */
    void scheduleDrainBuffers() {
      if (drainStatus() >= PROCESSING_TO_IDLE) {
        return;
      }
      if (evictionLock.tryLock()) {
        try {
          int drainStatus = drainStatus();
          if (drainStatus >= PROCESSING_TO_IDLE) {
            return;
          }
          lazySetDrainStatus(PROCESSING_TO_IDLE);
          // 提交异步任务去执行Drain Buffer
          executor.execute(drainBuffersTask);
        } catch (Throwable t) {
          logger.log(Level.WARNING, "Exception thrown when submitting maintenance task", t);
          maintenance(/* ignored */ null);
        } finally {
          evictionLock.unlock();
        }
      }
    }

### 异步写 WriteBuffer ###

缓存一般都是读多写少的场景，读的并发会更高，且afterRead显得没那么重要，允许延迟甚至丢失。写不一样，写afterWrite不允许丢失，且要求尽量马上执行。Caffeine使用MPSC（Multiple Producer / Single Consumer）作为buffer数组，实现在MpscGrowableArrayQueue类，它是仿照JCTools的MpscGrowableArrayQueue来写的。

MPSC允许无锁的高并发写入，但只允许一个消费者，同时也牺牲了部分操作。MPSC的细节不是重点。

我们看一下 `afterWrite` 函数：

/** The number of attempts to insert into the write buffer before yielding. */
    static final int WRITE_BUFFER_RETRIES = 100;
    
    
    void afterWrite(Runnable task) { 
      if (buffersWrites()) { 
        for (int i = 0; i < WRITE_BUFFER_RETRIES; i++) { 
          if (writeBuffer().offer(task)) { 
          	//每次写成功之后，都是尽可能快的去调度Drain buffer任务执行
            scheduleAfterWrite();
            return;
          }
          scheduleDrainBuffers();
        }
    
        // 如果插入task到writeBuffer失败了，就由线程自己同步执行task。
        try { 
          performCleanUp(task);
        } catch (RuntimeException e) { 
          logger.log(Level.SEVERE, "Exception thrown when performing the maintenance task", e);
        }
      } else { 
        scheduleAfterWrite();
      }
    }

总而言之就是，写入WriteBuffer不允许丢失。写入失败了就同步执行。

### 异步PerformCleanupTask ###

PerformCleanupTask其实就是异步的maintenance任务。

static final class PerformCleanupTask extends ForkJoinTask<Void> implements Runnable { 
      @Override
      public boolean exec() { 
        try { 
          run();
        } catch (Throwable t) { 
          logger.log(Level.SEVERE, "Exception thrown when performing the maintenance task", t);
        }
        **// Indicates that the task has not completed to allow subsequent submissions to execute
        return false;
      }
      @Override
      public void run() { 
        BoundedLocalCache<?, ?> cache = reference.get();
        if (cache != null) { 
          cache.performCleanUp(/* ignored */ null);
        }
      }
    }
    
    void performCleanUp(@Nullable Runnable task) { 
      evictionLock.lock();
      try { 
        maintenance(task);
      } finally { 
        evictionLock.unlock();
      }
      if ((drainStatus() == REQUIRED) && (executor == ForkJoinPool.commonPool())) { 
        scheduleDrainBuffers();
      }
    }

PerformCleanupTasks实际上就是 `performCleanUp` 函数的调用。performCleanUp函数里面通过调用`maintenance(task);` 做一次cache的维护工作。

## TimeWheel设计 ##

Caffeine的过期策略，除了支持`expireAfterAccess`和`expireAfterWrite`之外，还支持`expireAfter`自定义过期时间。因为expireAfterAccess和expireAfterWrite都只能是固定的过期时间，这可能满足不了某些场景，譬如记录的过期时间是需要根据某些条件而不一样的，这就需要用户自定义过期时间。

先看看expireAfter的用法:

LoadingCache<String, String> cache = Caffeine.newBuilder()
     //最大个数限制
     .maximumSize(256L)
     //初始化容量
     .initialCapacity(1)
     //访问后过期（包括读和写）
     //.expireAfterAccess(2, TimeUnit.DAYS)
     //写后过期
     //.expireAfterWrite(2, 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;
              }
         }
     )
     //写后自动异步刷新
     .refreshAfterWrite(1, TimeUnit.HOURS)
     //记录下缓存的一些统计数据，例如命中率等
     .recordStats()
     //使用CacheLoader创建一个LoadingCache
     .build();

通过自定义过期时间，使得不同的key可以动态的得到不同的过期时间。

注意，我把expireAfterAccess和expireAfterWrite注释了，因为这两个特性不能跟expireAfter一起使用。

而当使用了expireAfter特性后，Caffeine会启用一种叫“时间轮”的算法来实现这个功能。

为什么要用时间轮？

对expireAfterAccess和expireAfterWrite的实现是用一个AccessOrderDeque双端队列，它是FIFO的，因为它们的过期时间是固定的，所以在队列头的数据肯定是最早过期的，要处理过期数据时，只需要首先看看头部是否过期，然后再挨个检查就可以了。但是，如果过期时间不一样的话，这需要对accessOrderQueue进行排序&插入，这个代价太大了。于是，Caffeine用了一种更加高效、优雅的算法-时间轮。

Caffeine对时间轮的实现在TimerWheel，它是一种多层时间轮（hierarchical timing wheels ）。

具体的细节这里不描述了。

## 整体设计架构 ##

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA4NTMyNjE_size_16_color_FFFFFF_t_70]: /images/20230521/cace527499fd432997fd4c14153e36d2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA4NTMyNjE_size_16_color_FFFFFF_t_70 1]: /images/20230521/4e23032b4cf74e4ea773ea36da306ad2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA4NTMyNjE_size_16_color_FFFFFF_t_70 2]: /images/20230521/882f3d2661f7471ba441b2305fb96ce1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA4NTMyNjE_size_16_color_FFFFFF_t_70 3]: /images/20230521/2aca0d5d8e4d41daa625fc9a160cc464.png