办网站需流程,wordpress 文章音频,微信公众平台官网首页,刚刚好痛上篇文章#xff1a;缓存之美#xff1a;万文详解 Caffeine 实现原理#xff08;上#xff09; getIfPresent
现在我们对 put 方法有了基本了解#xff0c;现在我们继续深入 getIfPresent 方法#xff1a;
public class TestReadSourceCode {Testpublic void doRead() …上篇文章缓存之美万文详解 Caffeine 实现原理上 getIfPresent
现在我们对 put 方法有了基本了解现在我们继续深入 getIfPresent 方法
public class TestReadSourceCode {Testpublic void doRead() {// read constructorCacheString, String cache Caffeine.newBuilder().maximumSize(10_000).build();// read putcache.put(key, value);// read getcache.getIfPresent(key);}}对应源码如下关注注释信息
abstract class BoundedLocalCacheK, V extends BLCHeader.DrainStatusRef implements LocalCacheK, V {final ConcurrentHashMapObject, NodeK, V data;final BufferNodeK, V readBuffer;Overridepublic Nullable V getIfPresent(Object key, boolean recordStats) {// 直接由 ConcurrentHashMap 获取元素NodeK, V node data.get(nodeFactory.newLookupKey(key));if (node null) {// 更新统计未命中if (recordStats) {statsCounter().recordMisses(1);}// 当前 drainStatus 为 REQUIRED 表示有任务需要处理则调度处理if (drainStatusOpaque() REQUIRED) {// 这个方法在上文中介绍过它会提交 PerformCleanupTask 执行维护方法 maintenancescheduleDrainBuffers();}return null;}V value node.getValue();long now expirationTicker().read();// 判断是否过期或者需要被回收且value对应的值为nullif (hasExpired(node, now) || (collectValues() (value null))) {// 更新统计未命中if (recordStats) {statsCounter().recordMisses(1);}scheduleDrainBuffers();return null;}// 检查节点没有在进行异步计算if (!isComputingAsync(node)) {SuppressWarnings(unchecked)K castedKey (K) key;// 更新访问时间setAccessTime(node, now);// 更新读后过期时间tryExpireAfterRead(node, castedKey, value, expiry(), now);}// 处理读取后操作主要关注V refreshed afterRead(node, now, recordStats);return (refreshed null) ? value : refreshed;}
}getIfPresent 方法中部分内容我们已经在上文中介绍过比如 scheduleDrainBuffers 方法。最后一步 afterRead 方法是我们本次关注的重点从命名来看它表示“读后操作”接下来看看它的具体流程
abstract class BoundedLocalCacheK, V extends BLCHeader.DrainStatusRef implements LocalCacheK, V {final BufferNodeK, V readBuffer;NullableV afterRead(NodeK, V node, long now, boolean recordHit) {// 更新统计命中if (recordHit) {statsCounter().recordHits(1);}// 注意这里如果 skipReadBuffer 为 false那么它会执行 readBuffer.offer(node) 逻辑向 ReadBuffer 中添加待处理元素boolean delayable skipReadBuffer() || (readBuffer.offer(node) ! Buffer.FULL);// 判断是否需要延迟处理维护任务if (shouldDrainBuffers(delayable)) {scheduleDrainBuffers();}// 处理必要的刷新操作return refreshIfNeeded(node, now);}boolean skipReadBuffer() {// fastpath 方法访问元素是否可以跳过“通知”驱逐策略true 表示跳过// 第二个判断条件判断频率草图是否初始化如果“未初始化”则返回 truereturn fastpath() frequencySketch().isNotInitialized();}// 状态流转没有满 delayable 为 true 表示延迟执行维护任务boolean shouldDrainBuffers(boolean delayable) {switch (drainStatusOpaque()) {case IDLE:return !delayable;// 当前有任务需要处理则调度维护任务执行否则均延迟执行 case REQUIRED:return true;case PROCESSING_TO_IDLE:case PROCESSING_TO_REQUIRED:return false;default:throw new IllegalStateException(Invalid drain status: drainStatus);}}
}该方法非常简单都是熟悉的内容只有数据结构 ReadBuffer 还没深入了解过它也是在 Caffeine 的构造方法中完成初始化的。
ReadBuffer
以下为 ReadBuffer 在 Caffeine 缓存中完成初始化的逻辑
abstract class BoundedLocalCacheK, V extends BLCHeader.DrainStatusRefimplements LocalCacheK, V {final BufferNodeK, V readBuffer;protected BoundedLocalCache(CaffeineK, V builder,Nullable AsyncCacheLoaderK, V cacheLoader, boolean isAsync) {// ...// 如果指定了过期策略或 key 定义了 week refenence value 定义了 week or soft reference 或定义了访问后过期策略 则 创建 BoundBufferreadBuffer evicts() || collectKeys() || collectValues() || expiresAfterAccess()? new BoundedBuffer(): Buffer.disabled();}
}Buffer.disabled() 会创建如下枚举来表示 DisabledBuffer:
enum DisabledBuffer implements BufferObject {INSTANCE;Overridepublic int offer(Object e) {return Buffer.SUCCESS;}Overridepublic void drainTo(ConsumerObject consumer) {}Overridepublic long size() {return 0;}Overridepublic long reads() {return 0;}Overridepublic long writes() {return 0;}
}满足其中条件判断时ReadBuffer 的实际类型为 BoundedBuffer它的类关系图如下 在 Buffer 接口的注释声明中能获取很多有效信息它同样也是 多生产者单消费者MPSC 缓冲区上文我们在介绍 WriteBuffer 时它的单消费者实现方式是加同步锁ReadBuffer 的实现单消费者的方式一样因为它们都是在维护方法 maintenance 中加同步锁对元素进行消费。不同的是如果 ReadBuffer 缓冲区满了或者发生争抢则会拒绝添加新元素而且它不像队列或栈不保证 FIFO 或 LIFO。 A multiple-producer / single-consumer buffer that rejects new elements if it is full or fails spuriously due to contention. Unlike a queue and stack, a buffer does not guarantee an ordering of elements in either FIFO or LIFO order. Beware that it is the responsibility of the caller to ensure that a consumer has exclusive read access to the buffer. This implementation does not include fail-fast behavior to guard against incorrect consumer usage. 在类关系图中抽象类 StripedBuffer 的实现最值得学习它采用了分段设计Striped和CAS操作实现高效并发写入。分段是将缓冲区分成多个“段”根据线程的探针值将它们哈希到不同的“段”减少竞争接下来我们看一下它具体的实现逻辑首先是 StripedBuffer#offer 方法
abstract class StripedBufferE implements BufferE {volatile BufferE Nullable[] table;Overridepublic int offer(E e) {// 扰动函数计算 64位 线程探针值long z mix64(Thread.currentThread().getId());// 取高 32 位值位或 1 保证它为奇数int increment ((int) (z 32)) | 1;// 转换为 int 32 位int h (int) z;// 掩码值为已分段的缓冲区数量-1int mask;int result;// 线程哈希到的具体缓冲区BufferE buffer;// 未竞争标志位boolean uncontended true;BufferE[] buffers table;if ((buffers null)|| ((mask buffers.length - 1) 0)// 位与运算获取缓冲区|| ((buffer buffers[h mask]) null)// 向缓冲区中添加元素|| !(uncontended ((result buffer.offer(e)) ! Buffer.FAILED))) {// 扩容或重试操作return expandOrRetry(e, h, increment, uncontended);}return result;}
}在 StripedBuffer 中我们能发现定义了 volatile BufferE Nullable[] table 是数组的形式这便对应了它“分段”的思想将元素保存在多个缓冲区中。通过线程探针值哈希获取对应的缓冲区逻辑并不复杂。expandOrRetry 方法我们稍后再介绍我们先假设线程哈希到的具体缓冲区 BufferE buffer 对象已经被创建那么它会执行 buffer.offer(e) 方法。BufferE buffer 对应的实现类为定义在 BoundedBuffer 的静态内部类 RingBuffer它也实现了 Buffer 接口源码如下
final class BoundedBufferE extends StripedBufferE {static final int BUFFER_SIZE 16;static final int MASK BUFFER_SIZE - 1;static final class RingBufferE extends BBHeader.ReadAndWriteCounterRef implements BufferE {static final VarHandle BUFFER MethodHandles.arrayElementVarHandle(Object[].class);final Object[] buffer;// 有参构造这里表示缓冲区是被延迟创建的创建时第一个元素便为 epublic RingBuffer(E e) {buffer new Object[BUFFER_SIZE];BUFFER.set(buffer, 0, e);WRITE.set(this, 1);}Overridepublic int offer(E e) {// ReadCounterRef#readCounterlong head readCounter;// ReadAndWriteCounterRef#writeCounterlong tail writeCounterOpaque();// 计算可操作容量 sizelong size (tail - head);// 超过缓存大小则证明它已经满了if (size BUFFER_SIZE) {return Buffer.FULL;}// CAS 更新 writeCounter 为 writeCounter1if (casWriteCounter(tail, tail 1)) {// 位与掩码值获取缓冲区中的索引int index (int) (tail MASK);// 将元素 e 更新在指定索引处BUFFER.setRelease(buffer, index, e);return Buffer.SUCCESS;}return Buffer.FAILED;}Overridepublic void drainTo(ConsumerE consumer) {// ReadCounterRef#readCounterlong head readCounter;// ReadAndWriteCounterRef#writeCounterlong tail writeCounterOpaque();// 计算可操作容量 sizelong size (tail - head);// size 为 0 表示无元素可操作if (size 0) {return;}// 循环遍历消费缓冲区中所有元素do {// 计算具体的索引int index (int) (head MASK);SuppressWarnings(unchecked)E e (E) BUFFER.getAcquire(buffer, index);// 索引处元素为空表示无元素可消费if (e null) {break;}// 获取到具体元素后将缓冲区该元素位置更新成 nullBUFFER.setRelease(buffer, index, null);// 执行消费逻辑consumer.accept(e);// head累加head;} while (head ! tail);// 更新读索引的值setReadCounterOpaque(head);}}
}final class BBHeader {SuppressWarnings(PMD.AbstractClassWithoutAbstractMethod)abstract static class PadReadCounter {byte p000, /*省略118字节占位符...*/ p119;}abstract static class ReadCounterRef extends PadReadCounter {volatile long readCounter;}abstract static class PadWriteCounter extends ReadCounterRef {byte p120, /*省略118字节占位符...*/ p239;}abstract static class ReadAndWriteCounterRef extends PadWriteCounter {static final VarHandle READ, WRITE;volatile long writeCounter;// ...}
}在 BBHeader 类中又看到了熟悉的 120 字节内存占位在上文中我们详细介绍过这样能够保证 readCounter 和 writeCounter 分布在不同内存行避免了内存伪共享问题保证不同线程读取这两个字段时互不影响。在添加元素的 offer 方法和消费元素的 drainTo 方法中都能看见它使用了“读索引readCounter”和“写索引writeCounter”这也对应了它命名中的 Ring。Ring 表示环形读、写索引在操作过程中会不断累加但是它会执行位与运算保证索引值一直落在缓冲区长度的有效范围内也就是说这两个索引值会不断在有效索引范围内“转圈”则形成一个“环形”缓冲区。
RingBuffer 通过 CAS 操作来确保并发添加元素操作的安全如果 CAS 操作失败则返回 Buffer.FAILED这时便会执行 StripedBuffer#expandOrRetry 方法我们先来看一下它的方法注释内容它说这个方法用于处理写过程中发生的初始化、扩容、创建新缓存或竞争写情况。 Handles cases of updates involving initialization, resizing, creating new Buffers, and/ or contention. 具体源码如下
abstract class StripedBufferE implements BufferE {// 最大尝试 3 次static final int ATTEMPTS 3;// table 的最大大小static final int MAXIMUM_TABLE_SIZE 4 * ceilingPowerOfTwo(NCPU);// 1 表示忙碌扩容或正在创建0 表示缓冲区无操作通过 CAS 操作进行更新volatile int tableBusy;volatile BufferE Nullable[] table;/*** 扩展或重试* * param e 元素* param h 调用该方法时为线程探针值高 32 位但在方法中会变更* param increment 线程探针值高 32 位* param wasUncontended true 未发生竞争 false 发生竞争*/final int expandOrRetry(E e, int h, int increment, boolean wasUncontended) {int result Buffer.FAILED;// true 标志缓冲区中最后一个槽位非空 false 表示为空boolean collide false;for (int attempt 0; attempt ATTEMPTS; attempt) {BufferE[] buffers;BufferE buffer;int n;// 如果缓冲区数组已经被创建if (((buffers table) ! null) ((n buffers.length) 0)) {// 检查具体的缓冲区是否为空if ((buffer buffers[(n - 1) h]) null) {// 准备创建缓冲区并更新 tableBusy 标志为 1if ((tableBusy 0) casTableBusy()) {boolean created false;try { BufferE[] rs;int mask, j;if (((rs table) ! null) ((mask rs.length) 0) (rs[j (mask - 1) h] null)) {// 创建缓冲区 return new RingBuffer(e);rs[j] create(e);created true;}} finally {tableBusy 0;}// 如果创建成功if (created) {result Buffer.SUCCESS;break;}// 缓冲区已经被其他线程创建了重新循环重试continue;}collide false;}// 如果发生竞争表示向缓冲区中CAS添加元素失败else if (!wasUncontended) {wasUncontended true;} // 如果重试添加元素成功结束循环else if ((result buffer.offer(e)) ! Buffer.FAILED) {break;}// table 超过最大大小或已完成扩容但未变更引用staleelse if ((n MAXIMUM_TABLE_SIZE) || (table ! buffers)) {collide false;} else if (!collide) {collide true;}// 扩容操作将缓冲区数组扩容为原来的两倍大小// 扩容条件未超过最大 table 限制且重试添加元素依然失败else if ((tableBusy 0) casTableBusy()) {try {if (table buffers) {table Arrays.copyOf(buffers, n 1);}} finally {tableBusy 0;}collide false;continue;}// 变更探针哈希值尝试下一个索引位置h increment;}// 缓冲区数组的初始化逻辑else if ((tableBusy 0) (table buffers) casTableBusy()) {boolean init false;try {if (table buffers) {// 初始大小为 1会随着扩容不断将容量扩大两倍SuppressWarnings({rawtypes, unchecked})BufferE[] rs new Buffer[1];rs[0] create(e);table rs;init true;}} finally {tableBusy 0;}// 完成初始化元素添加成功if (init) {result Buffer.SUCCESS;break;}}}return result;}
}根据注释信息了解该方法的逻辑并不难接下来我们再看一下它的消费方法 drainTo非常简单
abstract class StripedBufferE implements BufferE {volatile BufferE Nullable[] table;Overridepublic void drainTo(ConsumerE consumer) {BufferE[] buffers table;if (buffers null) {return;}// 循环遍历消费所有缓冲区for (BufferE buffer : buffers) {if (buffer ! null) {buffer.drainTo(consumer);}}}
}总结一下ReadBuffer 是一个 MPSC 的缓冲区采用了分段的设计将缓冲区划分为多份根据线程的探针值哈希到不同的缓冲区减少竞争的发生并使用CAS操作来保证多线程下写入操作高效执行。因为它没有记录元素的写入顺序所以它并不会像栈或队列一样保证 FIFO 或 LIFO。随着写入竞争发生会不断对缓冲区数组扩容每次扩容为原来大小的两倍每个缓冲区为环形缓冲区通过位与运算计算元素实际的索引将被消费的元素标记为 null 实现缓冲区中槽位的重用。
现在读写方法已经了解差不多了需要我们再次回到维护方法 maintenance 中看一看消费读缓冲区和其他逻辑。
maintenance
维护方法 maintenance 如下所示第 2 步中处理写缓冲区任务的逻辑已在上文中介绍过接下来我们会关注第 1 步的处理读缓冲区任务第 4 步驱逐策略和第 5 步的 “增值climb”操作。
abstract class BoundedLocalCacheK, V extends BLCHeader.DrainStatusRef implements LocalCacheK, V {GuardedBy(evictionLock)void maintenance(Nullable Runnable task) {// 更新状态为执行中setDrainStatusRelease(PROCESSING_TO_IDLE);try {// 1. 处理读缓冲区中的任务drainReadBuffer();// 2. 处理写缓冲区中的任务drainWriteBuffer();if (task ! null) {task.run();}// 3. 处理 key 和 value 的引用drainKeyReferences();drainValueReferences();// 4. 过期和驱逐策略expireEntries();evictEntries();// 5. “增值” 操作climb();} finally {// 状态不是 PROCESSING_TO_IDLE 或者无法 CAS 更新为 IDLE 状态的话需要更新状态为 REQUIRED该状态会再次执行维护任务if ((drainStatusOpaque() ! PROCESSING_TO_IDLE) || !casDrainStatus(PROCESSING_TO_IDLE, IDLE)) {setDrainStatusOpaque(REQUIRED);}}}
}drainReadBuffer
首先我们来看处理读缓冲区的逻辑源码如下
abstract class BoundedLocalCacheK, V extends BLCHeader.DrainStatusRef implements LocalCacheK, V {final BufferNodeK, V readBuffer;final ConsumerNodeK, V accessPolicy;GuardedBy(evictionLock)void drainReadBuffer() {if (!skipReadBuffer()) {readBuffer.drainTo(accessPolicy);}}}它会执行到 StripedBuffer#drainTo 方法并且入参了 ConsumerNodeK, V accessPolicy 消费者。前者会遍历所有缓冲区中对象进行消费后者在 caffeine 构造方法中完成初始化
abstract class BoundedLocalCacheK, V extends BLCHeader.DrainStatusRef implements LocalCacheK, V {final BufferNodeK, V readBuffer;final ConsumerNodeK, V accessPolicy;protected BoundedLocalCache(CaffeineK, V builder,Nullable AsyncCacheLoaderK, V cacheLoader, boolean isAsync) {accessPolicy (evicts() || expiresAfterAccess()) ? this::onAccess : e - {};}}onAccess 方法在上文中也提到过具体逻辑我们在这里赘述下
abstract class BoundedLocalCacheK, V extends BLCHeader.DrainStatusRef implements LocalCacheK, V {GuardedBy(evictionLock)void onAccess(NodeK, V node) {if (evicts()) {K key node.getKey();if (key null) {return;}// 更新访问频率frequencySketch().increment(key);// 如果节点在窗口区则将其移动到尾节点if (node.inWindow()) {reorder(accessOrderWindowDeque(), node);}// 在试用区的节点执行 reorderProbation 方法可能会将该节点从试用区晋升到保护区else if (node.inMainProbation()) {reorderProbation(node);}// 否则移动到保护区的尾结点else {reorder(accessOrderProtectedDeque(), node);}// 更新命中量setHitsInSample(hitsInSample() 1);}// 配置了访问过期策略else if (expiresAfterAccess()) {reorder(accessOrderWindowDeque(), node);}// 配置了自定义时间过期策略if (expiresVariable()) {timerWheel().reschedule(node);}}
}简单概括来说ReadBuffer 中所有的元素都会被执行 onAccess 的逻辑频率草图会被更新窗口区元素会被移动到该区的尾结点试用区元素在满足条件的情况下会被晋升到保护区。在原理图中补充 ReadBuffer 相关逻辑相比于原有 put 方法的逻辑ReadBuffer 的消费并没有引入特别“新颖”的内容 在 reorderProbation 方法中有一段注释比较有意思它说如果保护区空间超过它的最大值它会将其中的元素降级到试用区。但是这个操作被推迟到 maintenance 方法的最后执行也就是后续我们会介绍的 climb 方法相当于是对缓存元素的移动做了剧透。 If the protected space exceeds its maximum, the LRU items are demoted to the probation space. This is deferred to the adaption phase at the end of the maintenance cycle. evictEntries
evictEntries 方法注释这么描述如果缓存超过最大值则将元素驱逐。 Evicts entries if the cache exceeds the maximum 它的主方法逻辑非常简单
abstract class BoundedLocalCacheK, V extends BLCHeader.DrainStatusRef implements LocalCacheK, V {GuardedBy(evictionLock)void evictEntries() {if (!evicts()) {return;}// 从窗口区“驱逐”var candidate evictFromWindow();// 从候选区或保护区进行驱逐evictFromMain(candidate);}
}首先先来看从窗口区“驱逐”的方法 evictFromWindow:
abstract class BoundedLocalCacheK, V extends BLCHeader.DrainStatusRef implements LocalCacheK, V {GuardedBy(evictionLock)NullableNodeK, V evictFromWindow() {NodeK, V first null;// 获取队首元素NodeK, V node accessOrderWindowDeque().peekFirst();// 循环操作直到窗口区权重小于窗口区权重最大限制while (windowWeightedSize() windowMaximum()) {if (node null) {break;}// 获取队首节点的下一个节点NodeK, V next node.getNextInAccessOrder();// 如果队首节点权重不为 0if (node.getPolicyWeight() ! 0) {// 标记为试用区节点并移动到试用区尾节点node.makeMainProbation();accessOrderWindowDeque().remove(node);accessOrderProbationDeque().offerLast(node);// 记录队首节点引用if (first null) {first node;}// 更新窗口区权重setWindowWeightedSize(windowWeightedSize() - node.getPolicyWeight());}// node 记录操作完成后的下一个头节点node next;}// 返回此时的头节点return first;}
}该方法会根据窗口区最大权重限制 将节点由窗口区移动到试用区直到窗口区内元素小于最大值限制并不是直接调用 evictEntry 方法真正地将元素驱逐。如果已经在窗口区中将元素移动到试用区那么接下来会以窗口区头节点会作为入参执行 evictFromMain 方法它有非常详细的注释内容
如果缓存超过最大容量限制则将元素从主空间中移除。主空间通过频率草图决定从窗口区来的元素是被驱逐还是被保留以便将使用频率最低的元素移除。
窗口区的元素被提升到试用区尾节点MRU 位置驱逐策略驱逐的元素从试用区头节点LRU 位置开始。在需要执行驱逐策略时元素会按照由头节点到尾节点的顺序进行评估如果评估完试用区和保护区仍然需要驱逐元素那么则会从窗口区驱逐。相似地如果试用区驱逐完元素后仍然不够则需要从保护区检查元素进行驱逐。队列按照从头节点到尾节点的顺序消费使用频率相对较低的元素先被驱逐在相同频率的情况下优先保留主空间中的元素而不是窗口区元素。 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 decision is made using a frequency filter so that the least frequently used entry is removed. The window space’s candidates were previously promoted to the probation space at its MRU position and the eviction policy’s victim starts at the LRU position. The candidates are evaluated in promotion order while an eviction is required, and if exhausted then additional entries are retrieved from the window space. Likewise, if the victim selection exhausts the probation space then additional entries are retrieved the protected space. The queues are consumed in LRU order and the evicted entry is the one with a lower relative frequency, where the preference is to retain the main space’s victims versus the window space’s candidates on a tie. 接下来我们看下源码的具体实现
abstract class BoundedLocalCacheK, V extends BLCHeader.DrainStatusRef implements LocalCacheK, V {public static final int WINDOW 0;public static final int PROBATION 1;public static final int PROTECTED 2;static final int ADMIT_HASHDOS_THRESHOLD 6;// 为了方便理解定义 victim 为驱逐区candidate 为候选驱逐区实际上它们不对应区域而是对应某个区域中的节点元素GuardedBy(evictionLock)void evictFromMain(Nullable NodeK, V candidate) {int victimQueue PROBATION;int candidateQueue PROBATION;// 首先获取试用区头节点作为首先要被驱逐的区域NodeK, V victim accessOrderProbationDeque().peekFirst();// 如果权重大小超过最大值不断地执行驱逐策略直到满足条件while (weightedSize() maximum()) {// 如果候选驱逐区为空且候选驱逐区为试用区则指定候选驱逐区为窗口区if ((candidate null) (candidateQueue PROBATION)) {// 指定候选驱逐区为窗口区candidate accessOrderWindowDeque().peekFirst();candidateQueue WINDOW;}// 候选驱逐区和驱逐区都为空if ((candidate null) (victim null)) {// 当前驱逐区为试用区指定保护区为驱逐区if (victimQueue PROBATION) {victim accessOrderProtectedDeque().peekFirst();victimQueue PROTECTED;continue;}// 当前驱逐区为保护区指定驱逐区为窗口区else if (victimQueue PROTECTED) {victim accessOrderWindowDeque().peekFirst();victimQueue WINDOW;continue;}// 没有更多元素供驱逐则退出循环break;}// 跳过权重为 0 的元素权重为 0 表示无需驱逐if ((victim ! null) (victim.getPolicyWeight() 0)) {victim victim.getNextInAccessOrder();continue;} else if ((candidate ! null) (candidate.getPolicyWeight() 0)) {candidate candidate.getNextInAccessOrder();continue;}// 如果要驱逐区为空则从候选驱逐区中进行驱逐if (victim null) {// 驱逐当前节点并将指针指向下一个节点 NodeK, V previous candidate.getNextInAccessOrder();NodeK, V evict candidate;candidate previous;evictEntry(evict, RemovalCause.SIZE, 0L);continue;}// 候选驱逐区为空在驱逐区中驱逐元素else if (candidate null) {NodeK, V evict victim;victim victim.getNextInAccessOrder();evictEntry(evict, RemovalCause.SIZE, 0L);continue;}// 驱逐区和候选驱逐区是同一个区的元素if (candidate victim) {victim victim.getNextInAccessOrder();evictEntry(candidate, RemovalCause.SIZE, 0L);candidate null;continue;}// 如果元素已经被垃圾回收则驱逐K victimKey victim.getKey();K candidateKey candidate.getKey();if (victimKey null) {NodeK, V evict victim;victim victim.getNextInAccessOrder();evictEntry(evict, RemovalCause.COLLECTED, 0L);continue;} else if (candidateKey null) {NodeK, V evict candidate;candidate candidate.getNextInAccessOrder();evictEntry(evict, RemovalCause.COLLECTED, 0L);continue;}// 如果元素已经被标记为删除驱逐它们if (!victim.isAlive()) {NodeK, V evict victim;victim victim.getNextInAccessOrder();evictEntry(evict, RemovalCause.SIZE, 0L);continue;} else if (!candidate.isAlive()) {NodeK, V evict candidate;candidate candidate.getNextInAccessOrder();evictEntry(evict, RemovalCause.SIZE, 0L);continue;}// 如果候选区节点元素超过最大权重直接驱逐if (candidate.getPolicyWeight() maximum()) {NodeK, V evict candidate;candidate candidate.getNextInAccessOrder();evictEntry(evict, RemovalCause.SIZE, 0L);continue;}// 驱逐频率较低的元素if (admit(candidateKey, victimKey)) {NodeK, V evict victim;victim victim.getNextInAccessOrder();evictEntry(evict, RemovalCause.SIZE, 0L);// 变更候选区元素引用candidate candidate.getNextInAccessOrder();} else {NodeK, V evict candidate;candidate candidate.getNextInAccessOrder();evictEntry(evict, RemovalCause.SIZE, 0L);}}}GuardedBy(evictionLock)boolean admit(K candidateKey, K victimKey) {// 获取候选驱逐区中元素频率int victimFreq frequencySketch().frequency(victimKey);int candidateFreq frequencySketch().frequency(candidateKey);// 候选区元素频率大于驱逐区中元素返回 trueif (candidateFreq victimFreq) {return true;}// 如果候选区元素频率大于 6else if (candidateFreq ADMIT_HASHDOS_THRESHOLD) {// 计算随机值来决定两元素之间的去留int random ThreadLocalRandom.current().nextInt();return ((random 127) 0);// 使用计算随机值的方法来防止 HASH DOS 攻击攻击者可能人为地将某些不被常用的缓存访问频率提高如果不计算随机性那么会将真正有价值的元素驱逐添加这种随机性计算可能减少这种攻击带来的影响保证缓存的有效命中率}// 候选驱逐区元素小于驱逐区元素频率return false;}
}方法虽然很长但是逻辑清晰明了元素的驱逐流程根据注释可以很明确的了解。窗口区中元素会优先被晋升到试用区在试用区和保护区中不断的驱逐节点直到满足条件如果驱逐完成之后还不满足条件则会从窗口区中驱逐元素此外在逻辑中使用随机驱逐的方式来减少 HASH DOS 攻击带来的影响也很值得学习更新原理图如下 climb
现在我们来到了维护方法的最后一个步骤 climb 方法看看它是如何为缓存“增值climb”的源码如下
abstract class BoundedLocalCacheK, V extends BLCHeader.DrainStatusRef implements LocalCacheK, V {static final double HILL_CLIMBER_RESTART_THRESHOLD 0.05d;static final double HILL_CLIMBER_STEP_PERCENT 0.0625d;// 步长值衰减比率static final double HILL_CLIMBER_STEP_DECAY_RATE 0.98d;static final int QUEUE_TRANSFER_THRESHOLD 1_000;GuardedBy(evictionLock)void climb() {if (!evicts()) {return;}// 确定要调整的量determineAdjustment();// 将保护区中的元素降级到试用区demoteFromMainProtected();// 获取第一步计算完毕的调整大小long amount adjustment();// 不调整则结束否则根据正负增大或减小窗口大小if (amount 0) {return;} else if (amount 0) {increaseWindow();} else {decreaseWindow();}}GuardedBy(evictionLock)void determineAdjustment() {// 检查频率草图是否被初始化if (frequencySketch().isNotInitialized()) {// 没有被初始化则重置命中率、命中和未命中样本数setPreviousSampleHitRate(0.0);setMissesInSample(0);setHitsInSample(0);return;}// 请求总数 命中样本数 未命中样本数int requestCount hitsInSample() missesInSample();if (requestCount frequencySketch().sampleSize) {return;}// 计算命中率、命中率变化double hitRate (double) hitsInSample() / requestCount;double hitRateChange hitRate - previousSampleHitRate();// 计算调整量如果命中率增加获取正的步长值否则获取负的步长值double amount (hitRateChange 0) ? stepSize() : -stepSize();// 计算下一个步长值如果变化量超过阈值那么重新计算步长否则按照固定衰减率计算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);}GuardedBy(evictionLock)void demoteFromMainProtected() {// 获取保护区的最大值和当前值long mainProtectedMaximum mainProtectedMaximum();long mainProtectedWeightedSize mainProtectedWeightedSize();// 当前值没有超过最大值则不处理if (mainProtectedWeightedSize mainProtectedMaximum) {return;}// 每次从保护区转换到试用区有 1000 个最大限制for (int i 0; i QUEUE_TRANSFER_THRESHOLD; i) {// 一旦不超过最大阈值则停止if (mainProtectedWeightedSize mainProtectedMaximum) {break;}// 在保护区取出头节点NodeK, V demoted accessOrderProtectedDeque().poll();if (demoted null) {break;}// 标记为试用区demoted.makeMainProbation();// 加入到试用区尾节点accessOrderProbationDeque().offerLast(demoted);// 计算变更后保护区权重大小mainProtectedWeightedSize - demoted.getPolicyWeight();}// 更新保护区权重setMainProtectedWeightedSize(mainProtectedWeightedSize);}GuardedBy(evictionLock)void increaseWindow() {// 保护区最大容量为 0 则没有可调整的空间if (mainProtectedMaximum() 0) {return;}// 窗口调整的变化量由保护区贡献取能够变化额度 quota 为 调整量adjustment 和 保护区最大值 中的小值long quota Math.min(adjustment(), mainProtectedMaximum());// 减小保护区大小增加窗口区大小setMainProtectedMaximum(mainProtectedMaximum() - quota);setWindowMaximum(windowMaximum() quota);// 保护区大小变动后需要操作元素由保护区降级到试用区demoteFromMainProtected();// 窗口区增加容量之后需要优先从试用区获取元素将增加的容量填满如果试用区元素不够则从保护区获取元素来填for (int i 0; i QUEUE_TRANSFER_THRESHOLD; i) {// 获取试用区头节点为“候选节点”NodeK, V candidate accessOrderProbationDeque().peekFirst();boolean probation true;// 如果试用区元素为空或者窗口调整的变化量要比该节点所占的权重小那么尝试从保护区获取节点if ((candidate null) || (quota candidate.getPolicyWeight())) {candidate accessOrderProtectedDeque().peekFirst();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().offerLast(candidate);// 标记为窗口区节点candidate.makeWindow();}// 可能存在 quota 小于 节点权重 的情况那么这些量无法再调整需要重新累加到保护区并在窗口区中减掉setMainProtectedMaximum(mainProtectedMaximum() quota);setWindowMaximum(windowMaximum() - quota);// 将未完成调整的 quota 记录在调整值中setAdjustment(quota);}GuardedBy(evictionLock)void decreaseWindow() {// 如果窗口区大小小于等于 1 则无法再减少了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) {// 从窗口区获取“候选节点”NodeK, V candidate accessOrderWindowDeque().peekFirst();// 未获取到说明窗口区已经没有元素了不能再减小了结束循环操作if (candidate null) {break;}// 获取候选节点的权重int weight candidate.getPolicyWeight();// 可变化的额度小于权重则不支持变化结束循环if (quota weight) {break;}// 随着节点的移动变更可变化额度quota - weight;// 更新窗口区大小并将元素从窗口区移除setWindowWeightedSize(windowWeightedSize() - weight);accessOrderWindowDeque().remove(candidate);// 将从窗口区中移除的元素添加到试用区accessOrderProbationDeque().offerLast(candidate);// 将节点标记为试用区元素candidate.makeMainProbation();}// 此时 quote 为剩余无法变更的额度需要在保护区中减去在窗口区中加上setMainProtectedMaximum(mainProtectedMaximum() - quota);setWindowMaximum(windowMaximum() quota);// 记录未变更完的额度在调整值中setAdjustment(-quota);}}现在我们了解了 climb 方法的逻辑正如它的注释所述 Adapts the eviction policy to towards the optimal recency / frequency configuration.它会根据访问情况动态调整最佳的分区配置以适应驱逐策略。元素被添加时会优先被放在窗口区窗口区越大则意味着短期内有大量缓存被添加或元素添加后被再次访问缓存命中率提高需要更大的窗口区来承接这部分新晋的元素。根据 climb 中的逻辑窗口区增大也会有试用区/保护区的元素不断被移动到窗口区如果保护区越大意味着缓存中维护的元素都是访问频率较高的元素命中率降低并趋于某稳定值附近试用区元素由窗口区元素晋升得来再被访问时会被晋升到保护区它更像是 JVM 分区的 survivor 区。缓冲区不同分区的动态调整可以适应不同的访问模式优化缓存的性能。接下来我们在原理图中补充上各个分区间元素的变换路径元素也可由保护区直接降级到窗口区但在图中未标出并根据图示对 Caffeine 的实现原理进行概括 在图示1中put 方法会直接将元素添加到 ConcurrentHashMap 中并在 WriteBuffer 中添加任务由单线程异步调用维护方法对任务进行消费元素访问频率会被更新试用区元素可能会被晋升到保护区在图示2调用 getIfPresent 方法会直接从 ConcurrentHashMap 中获取元素并添加任务到 ReadBuffer 中由单线程异步消费它相比于1并没有什么额外操作两个缓冲区均采用 MPSC 的设计模式这种设计参考了 WALWrite-Ahead Logging思想图示3和图示4均发生在维护方法逻辑中图示3驱逐元素时窗口区元素会被“驱逐”到试用区而试用区和保护区元素可能被直接驱逐图示4“增值climb”操作会根据命中率调整窗口区和保护区的大小合理分配分区间的元素。
在文中提到过每个分区的双端队列使用了 LRU 算法被访问过的元素会被放在尾节点但对元素进行驱逐时并不以 LRU 的顺序为准而是会参考频率草图中记录的元素频率保证使用频率高的被保留低的被驱逐。这和 LFU 算法很像区别于 LFU 算法的是它采用了 Count-Min Sketch 数据结构来记录频率能够在较小的内存开销下实现对频率较为精准93.75%的估计这种算法实际被称为 TinyLFU 算法它结合了两者的有点在内存和计算开销上达到更好的平衡。
技术选型
现在我们已经对 Caffeine 缓存有了一定的了解那么究竟什么时候适合选择使用它呢那就要根据它的特点来了首先它是线程安全的适合在多线程环境下使用其次它的性能很好使用了 TinyLFU 算法并采用了高性能缓存的设计再就是它提供了多种缓存管理机制除了基于最大容量的驱逐策略还支持基于时间、软/虚引用等驱逐策略。所以 它适合在高并发环境并且需要高性能、支持多种缓存管理策略的场景下使用。
如果要在多种缓存中选取可以以如下表格为参考
缓存是否线程安全性能缓存管理机制HashMap否高无ConcurrentHashMap是高无Guava是中提供基于最大容量、时间、软/虚引用等驱逐策略Caffeine是高提供基于最大容量、时间、软/虚引用等驱逐策略
巨人的肩膀
Github - caffeine并发编程网 - 现代化的缓存设计方案博客园 - CPU Cache与缓存行 文章转载自： http://www.morning.jxscp.cn.gov.cn.jxscp.cn http://www.morning.frxsl.cn.gov.cn.frxsl.cn http://www.morning.tkgjl.cn.gov.cn.tkgjl.cn http://www.morning.zmyhn.cn.gov.cn.zmyhn.cn http://www.morning.rhnn.cn.gov.cn.rhnn.cn http://www.morning.klpwl.cn.gov.cn.klpwl.cn http://www.morning.rlrxh.cn.gov.cn.rlrxh.cn http://www.morning.ckhyj.cn.gov.cn.ckhyj.cn http://www.morning.qtltg.cn.gov.cn.qtltg.cn http://www.morning.hbkkc.cn.gov.cn.hbkkc.cn http://www.morning.mqwdh.cn.gov.cn.mqwdh.cn http://www.morning.bqmsm.cn.gov.cn.bqmsm.cn http://www.morning.ypdmr.cn.gov.cn.ypdmr.cn http://www.morning.kwqt.cn.gov.cn.kwqt.cn http://www.morning.rxnr.cn.gov.cn.rxnr.cn http://www.morning.xjnw.cn.gov.cn.xjnw.cn http://www.morning.fykqh.cn.gov.cn.fykqh.cn http://www.morning.mzydm.cn.gov.cn.mzydm.cn http://www.morning.ie-comm.com.gov.cn.ie-comm.com http://www.morning.tqbqb.cn.gov.cn.tqbqb.cn http://www.morning.lgtcg.cn.gov.cn.lgtcg.cn http://www.morning.ngkgy.cn.gov.cn.ngkgy.cn http://www.morning.jcxzq.cn.gov.cn.jcxzq.cn http://www.morning.srbl.cn.gov.cn.srbl.cn http://www.morning.hdrrk.cn.gov.cn.hdrrk.cn http://www.morning.brps.cn.gov.cn.brps.cn http://www.morning.nnrqg.cn.gov.cn.nnrqg.cn http://www.morning.rdpps.cn.gov.cn.rdpps.cn http://www.morning.mwqbp.cn.gov.cn.mwqbp.cn http://www.morning.zfzgp.cn.gov.cn.zfzgp.cn http://www.morning.mcjrf.cn.gov.cn.mcjrf.cn http://www.morning.tnktt.cn.gov.cn.tnktt.cn http://www.morning.fwqgy.cn.gov.cn.fwqgy.cn http://www.morning.lcmhq.cn.gov.cn.lcmhq.cn http://www.morning.glnfn.cn.gov.cn.glnfn.cn http://www.morning.rpdmj.cn.gov.cn.rpdmj.cn http://www.morning.rmpfh.cn.gov.cn.rmpfh.cn http://www.morning.gybnk.cn.gov.cn.gybnk.cn http://www.morning.yxmcx.cn.gov.cn.yxmcx.cn http://www.morning.nkrmh.cn.gov.cn.nkrmh.cn http://www.morning.ssfq.cn.gov.cn.ssfq.cn http://www.morning.nxdqz.cn.gov.cn.nxdqz.cn http://www.morning.wprxm.cn.gov.cn.wprxm.cn http://www.morning.fxygn.cn.gov.cn.fxygn.cn http://www.morning.xxwfq.cn.gov.cn.xxwfq.cn http://www.morning.jwbnm.cn.gov.cn.jwbnm.cn http://www.morning.kpgft.cn.gov.cn.kpgft.cn http://www.morning.snzgg.cn.gov.cn.snzgg.cn http://www.morning.qxdrw.cn.gov.cn.qxdrw.cn http://www.morning.gllgf.cn.gov.cn.gllgf.cn http://www.morning.pzbjy.cn.gov.cn.pzbjy.cn http://www.morning.kjjbz.cn.gov.cn.kjjbz.cn http://www.morning.xfmzk.cn.gov.cn.xfmzk.cn http://www.morning.pzwfw.cn.gov.cn.pzwfw.cn http://www.morning.lwgrf.cn.gov.cn.lwgrf.cn http://www.morning.qwnqt.cn.gov.cn.qwnqt.cn http://www.morning.prprj.cn.gov.cn.prprj.cn http://www.morning.hwcgg.cn.gov.cn.hwcgg.cn http://www.morning.mrcpy.cn.gov.cn.mrcpy.cn http://www.morning.khpgd.cn.gov.cn.khpgd.cn http://www.morning.rqlzz.cn.gov.cn.rqlzz.cn http://www.morning.lbssg.cn.gov.cn.lbssg.cn http://www.morning.tpbhf.cn.gov.cn.tpbhf.cn http://www.morning.dtzxf.cn.gov.cn.dtzxf.cn http://www.morning.ccsdx.cn.gov.cn.ccsdx.cn http://www.morning.kdrly.cn.gov.cn.kdrly.cn http://www.morning.jlboyuan.cn.gov.cn.jlboyuan.cn http://www.morning.nynlf.cn.gov.cn.nynlf.cn http://www.morning.xsrnr.cn.gov.cn.xsrnr.cn http://www.morning.ywpcs.cn.gov.cn.ywpcs.cn http://www.morning.yxnkr.cn.gov.cn.yxnkr.cn http://www.morning.wqfj.cn.gov.cn.wqfj.cn http://www.morning.dgxrz.cn.gov.cn.dgxrz.cn http://www.morning.xysdy.cn.gov.cn.xysdy.cn http://www.morning.kggxj.cn.gov.cn.kggxj.cn http://www.morning.ybgpk.cn.gov.cn.ybgpk.cn http://www.morning.hkpn.cn.gov.cn.hkpn.cn http://www.morning.wqgr.cn.gov.cn.wqgr.cn http://www.morning.rxyz.cn.gov.cn.rxyz.cn http://www.morning.ydmml.cn.gov.cn.ydmml.cn
