接下來我們看看其他抽象類的作用。BaseMpscLinkedArrayQueueProducerFields 定義生產(chǎn)者相關字段：

abstract class BaseMpscLinkedArrayQueueProducerFields extends BaseMpscLinkedArrayQueuePad1 {
    // 生產(chǎn)者操作索引（并不對應緩沖區(qū) producerBuffer 中索引位置）
    protected long producerIndex;
}

BaseMpscLinkedArrayQueueConsumerFields 負責定義消費者相關字段：

abstract class BaseMpscLinkedArrayQueueConsumerFields extends BaseMpscLinkedArrayQueuePad2 {
    // 掩碼值，用于計算消費者實際的索引位置
    protected long consumerMask;
    // 消費者訪問這個緩沖區(qū)來獲取元素消費
    protected E[] consumerBuffer;
    // 消費者操作索引（并不對應緩沖區(qū) consumerBuffer 中索引位置）
    protected long consumerIndex;
}

BaseMpscLinkedArrayQueueColdProducerFields 中定義字段如下，該類的命名包含 Cold，表示其中字段被修改的次數(shù)會比較少：

abstract class BaseMpscLinkedArrayQueueColdProducerFields extends BaseMpscLinkedArrayQueuePad3 {
    // 生產(chǎn)者可以操作的最大索引上限
    protected volatile long producerLimit;
    // 掩碼值，用于計算生產(chǎn)者在數(shù)組中實際的索引
    protected long producerMask;
    // 存儲生產(chǎn)者生產(chǎn)的元素
    protected E[] producerBuffer;
}

現(xiàn)在關鍵字段我們已經(jīng)介紹完了，接下來看一下創(chuàng)建 MpscGrowableArrayQueue 的邏輯，執(zhí)行它的構(gòu)造方法時會為我們剛剛提到的字段進行賦值：

class MpscGrowableArrayQueue extends MpscChunkedArrayQueue {

    MpscGrowableArrayQueue(int initialCapacity, int maxCapacity) {
        // 調(diào)用父類的構(gòu)造方法
        super(initialCapacity, maxCapacity);
    }
}

abstract class MpscChunkedArrayQueue extends MpscChunkedArrayQueueColdProducerFields {
    // 省略字節(jié)占位字段...
    byte p119;

    MpscChunkedArrayQueue(int initialCapacity, int maxCapacity) {
        // 調(diào)用父類的構(gòu)造方法
        super(initialCapacity, maxCapacity);
    }

}

abstract class MpscChunkedArrayQueueColdProducerFields extends BaseMpscLinkedArrayQueue {
    protected final long maxQueueCapacity;

    MpscChunkedArrayQueueColdProducerFields(int initialCapacity, int maxCapacity) {
        // 調(diào)用父類的構(gòu)造方法
        super(initialCapacity);
        if (maxCapacity < 4) {
            throw new IllegalArgumentException("Max capacity must be 4 or more");
        }
        // 保證了最大值最少比初始值大 2 倍
        if (ceilingPowerOfTwo(initialCapacity) >= ceilingPowerOfTwo(maxCapacity)) {
            throw new IllegalArgumentException(
                    "Initial capacity cannot exceed maximum capacity(both rounded up to a power of 2)");
        }
        // 最大容量也為 2的n次冪
        maxQueueCapacity = ((long) ceilingPowerOfTwo(maxCapacity))  extends BaseMpscLinkedArrayQueueColdProducerFields {

    BaseMpscLinkedArrayQueue(final int initialCapacity) {
        if (initialCapacity < 2) {
            throw new IllegalArgumentException("Initial capacity must be 2 or more");
        }

        // 初始化緩沖區(qū)大小為數(shù)值最接近的 2 的 n 次冪
        int p2capacity = ceilingPowerOfTwo(initialCapacity);
        // 掩碼值，-1L 使其低位均為 1，左移 1 位則最低位為 0，eg: 00000110，注意該值會被生產(chǎn)者和消費者掩碼值共同賦值
        long mask = (p2capacity - 1L) 

	

	現(xiàn)在 MpscGrowableArrayQueue 的構(gòu)建已經(jīng)看完了，了解了其中關鍵字段的賦值，現(xiàn)在我們就需要看它是如何實現(xiàn) MPSC 的?！岸嗌a(chǎn)者”也就意味著會有多個線程向其中添加元素，既然是多線程就需要重點關注它是如何在多線程間完成協(xié)同的。添加操作對應了 BaseMpscLinkedArrayQueue#offer 方法，它的實現(xiàn)如下：

	
abstract class BaseMpscLinkedArrayQueue extends BaseMpscLinkedArrayQueueColdProducerFields {

    private static final Object JUMP = new Object();

    @Override
    @SuppressWarnings("MissingDefault")
    public boolean offer(final E e) {
        if (e == null) {
            throw new NullPointerException();
        }

        long mask;
        E[] buffer;
        long pIndex;

        while (true) {
            // 生產(chǎn)者最大索引（生產(chǎn)者掩碼值），獲取 BaseMpscLinkedArrayQueueColdProducerFields 中定義的該字段
            long producerLimit = lvProducerLimit();
            // 生產(chǎn)者當前索引，初始值為 0，BaseMpscLinkedArrayQueueProducerFields 中字段 
            pIndex = lvProducerIndex(this);
            // producerIndex 最低位用來表示擴容（索引生產(chǎn)者索引 producerIndex 并不對應緩沖區(qū)中實際的索引）
            // 低位為 1 表示正在擴容，自旋等待直到擴容完成（表示只有一個線程操作擴容）
            if ((pIndex & 1) == 1) {
                continue;
            }

            // 掩碼值和buffer可能在擴容中被改變，每次循環(huán)使用最新值
            mask = this.producerMask;
            buffer = this.producerBuffer;

            // 檢查是否需要擴容
            if (producerLimit <= pIndex) {
                int result = offerSlowPath(mask, pIndex, producerLimit);
                switch (result) {
                    case 0:
                        break;
                    case 1:
                        continue;
                    case 2:
                        return false;
                    case 3:
                        resize(mask, buffer, pIndex, e);
                        return true;
                }
            }

            // CAS 操作更新生產(chǎn)者索引，注意這里是 +2，更新成功結(jié)束循環(huán)
            if (casProducerIndex(this, pIndex, pIndex + 2)) {
                break;
            }
        }
        // 計算該元素在 buffer 中的實際偏移量，并將其添加到緩沖區(qū)中
        final long offset = modifiedCalcElementOffset(pIndex, mask);
        soElement(buffer, offset, e);
        return true;
    }

    // 沒有將 resize 邏輯封裝在該方法中，而是由該方法判斷是否需要擴容
    private int offerSlowPath(long mask, long pIndex, long producerLimit) {
        int result;
        // 獲取消費者索引 BaseMpscLinkedArrayQueueConsumerFields 類中
        final long cIndex = lvConsumerIndex(this);
        // 通過掩碼值計算當前緩沖區(qū)容量
        long bufferCapacity = getCurrentBufferCapacity(mask);
        result = 0;
        // 如果隊列還有空間
        if (cIndex + bufferCapacity > pIndex) {
            // 嘗試更新生產(chǎn)者最大限制，更新失敗則返回 1 重試
            if (!casProducerLimit(this, producerLimit, cIndex + bufferCapacity)) {
                result = 1;
            }
        }
        // 如果隊列已滿且無法擴展
        else if (availableInQueue(pIndex, cIndex) <= 0) {
            result = 2;
        }
        // 更新 producerIndex 最低位為 1，成功則進行擴容，否則重試
        else if (casProducerIndex(this, pIndex, pIndex + 1)) {
            result = 3;
        } else {
            result = 1;
        }
        return result;
    }

    private void resize(long oldMask, E[] oldBuffer, long pIndex, final E e) {
        // 計算新緩沖區(qū)大小并創(chuàng)建，2 * (buffer.length - 1) + 1
        int newBufferLength = getNextBufferSize(oldBuffer);
        final E[] newBuffer = allocate(newBufferLength);

        // 更新緩沖區(qū)引用為新的緩沖區(qū)
        producerBuffer = newBuffer;
        // 更新新的掩碼
        final int newMask = (newBufferLength - 2) > 1;
    }
}


	

	可見，在這個過程中它并沒有限制操作線程數(shù)量，也沒有使用加鎖的同步機制。它通過保證 可見性，并使用 自旋鎖結(jié)合 CAS 操作 更新生產(chǎn)者索引值，因為該操作是原子的，同時只有一個線程能更新獲取索引值成功，更新失敗的線程會自旋重試，這樣便允許多線程同時添加元素，可見性保證和CAS操作源碼如下：

	
abstract class BaseMpscLinkedArrayQueue extends BaseMpscLinkedArrayQueueColdProducerFields {

    static final VarHandle P_INDEX = pIndexLookup.findVarHandle(
            BaseMpscLinkedArrayQueueProducerFields.class, "producerIndex", long.class);
    
    // volatile 可見性保證
    static long lvProducerIndex(BaseMpscLinkedArrayQueue self) {
        return (long) P_INDEX.getVolatile(self);
    }
    
    // CAS 操作
    static boolean casProducerIndex(BaseMpscLinkedArrayQueue self, long expect, long newValue) {
        return P_INDEX.compareAndSet(self, expect, newValue);
    }
}


	

	保證可見性（內(nèi)存操作對其他線程可見）的原理是 內(nèi)存屏障，除了保證可見性以外，內(nèi)存屏障還能夠 防止重排序（確保在內(nèi)存屏障前后的內(nèi)存操作不會被重排序，從而保證程序的正確性）。到這里，生產(chǎn)者添加元素的邏輯我們已經(jīng)分析完了，接下來我們需要繼續(xù)看一下消費者獲取元素的邏輯，它對應了 BaseMpscLinkedArrayQueue#poll 方法，同樣地，在這過程中需要關注“在這個方法中有沒有限制單一線程執(zhí)行”，以此實現(xiàn)單消費者呢：

	
abstract class BaseMpscLinkedArrayQueue extends BaseMpscLinkedArrayQueueColdProducerFields {
    
    private static final Object JUMP = new Object();
    
    public E poll() {
        // 讀取消費者相關字段 BaseMpscLinkedArrayQueueConsumerFields 類
        final E[] buffer = consumerBuffer;
        final long index = consumerIndex;
        final long mask = consumerMask;

        // 根據(jù)消費索引，計算出元素在消費者緩沖區(qū)中實際的位置
        final long offset = modifiedCalcElementOffset(index, mask);
        // 讀取該元素（volatile 可見性讀?。?        Object e = lvElement(buffer, offset);
        
        // 如果為空
        if (e == null) {
            // 比較生產(chǎn)者索引，如果兩個索引不相等，那么證明兩索引間存在距離表示還有元素能夠被消費
            if (index != lvProducerIndex(this)) {
                // 自旋讀取元素，直到讀到元素
                do {
                    e = lvElement(buffer, offset);
                } while (e == null);
            } else {
                // 索引相等證明確實是空隊列
                return null;
            }
        }
        if (e == JUMP) {
            // 獲取到新緩沖區(qū)
            final E[] nextBuffer = getNextBuffer(buffer, mask);
            // 在新緩沖區(qū)中獲取到對應元素
            return newBufferPoll(nextBuffer, index);
        }
        // 清除當前索引的元素，表示該元素已經(jīng)被消費
        soElement(buffer, offset, null);
        // 更新消費者索引，這里也是 +2，它并不表示實際的在緩沖區(qū)的索引
        soConsumerIndex(this, index + 2);
        return (E) e;
    }

    private E[] getNextBuffer(final E[] buffer, final long mask) {
        // 如果已經(jīng)發(fā)生擴容，此時 consumerMask 仍然對應的是擴容前的 mask
        // 此處與生產(chǎn)者操作擴容時拼接新舊緩沖區(qū)調(diào)用的是一樣的方法，這樣便能夠獲取到新緩沖區(qū)的偏移量
        final long nextArrayOffset = nextArrayOffset(mask);
        // 獲取到新緩沖區(qū)，因為在擴容操作時已經(jīng)將新緩沖區(qū)鏈接到舊緩沖區(qū)上了
        final E[] nextBuffer = (E[]) lvElement(buffer, nextArrayOffset);
        // 將舊緩沖區(qū)中新緩沖區(qū)位置設置為 null 表示舊緩沖區(qū)中已經(jīng)沒有任何元素需要被消費了，也不再需要被引用了（能被垃圾回收了）
        soElement(buffer, nextArrayOffset, null);
        return nextBuffer;
    }

    private long nextArrayOffset(final long mask) {
        return modifiedCalcElementOffset(mask + 2, Long.MAX_VALUE);
    }

    private E newBufferPoll(E[] nextBuffer, final long index) {
        // 計算出消費者操作索引在新緩沖區(qū)中對應的實際位置
        final long offsetInNew = newBufferAndOffset(nextBuffer, index);
        // 在新緩沖區(qū)中獲取到對應元素
        final E n = lvElement(nextBuffer, offsetInNew);
        if (n == null) {
            throw new IllegalStateException("new buffer must have at least one element");
        }
        // 清除當前索引的元素，表示該元素已經(jīng)被消費
        soElement(nextBuffer, offsetInNew, null);
        // 更新消費者索引
        soConsumerIndex(this, index + 2);
        return n;
    }

    private long newBufferAndOffset(E[] nextBuffer, final long index) {
        // 將消費者緩沖區(qū)引用和掩碼值更新
        consumerBuffer = nextBuffer;
        consumerMask = (nextBuffer.length - 2L) > 1;
    }
    
    static  E lvElement(E[] buffer, long offset) {
        return (E) REF_ARRAY.getVolatile(buffer, (int) offset);
    }
}


	

	可以發(fā)現(xiàn)在該方法中并沒有限制單一線程執(zhí)行，所以理論上這個方法可能被多個線程調(diào)用，那么它又為什么被稱為 MPSC 呢？在這個方法中的一段注釋值得細心體會：

	This implementation is correct for single consumer thread use only.

	此實現(xiàn)僅適用于單消費者線程使用

	所以調(diào)用該方法時開發(fā)者本身需要保證單線程調(diào)用而并不是在實現(xiàn)中控制。

	到這里 MpscGrowableArrayQueue 中核心的邏輯已經(jīng)講解完了，現(xiàn)在我們回過頭來再看一下隊列擴容前后生產(chǎn)者和消費者是如何協(xié)同的？在擴容前，consumerBuffer 和 producerBuffer 引用的是同一個緩沖區(qū)對象。如果發(fā)生擴容，那么生產(chǎn)者會創(chuàng)建一個新的緩沖區(qū)，并將 producerBuffer 引用指向它，此時它做了一個 非常巧妙 的操作，將 新緩沖區(qū)依然鏈接到舊緩沖區(qū) 上，并將觸發(fā)擴容的元素對應的舊緩沖區(qū)的索引處標記為 JUMP，表示這及之后的元素已經(jīng)都在新緩沖區(qū)中。此時，消費者依然會在舊緩沖區(qū)中慢慢地消費，直到遇到 JUMP 標志位，消費者就知道需要到新緩沖區(qū)中取獲取元素了。因為之前生產(chǎn)者在擴容時對新舊緩沖區(qū)進行鏈接，所以消費者能夠通過舊緩沖區(qū)獲取到新緩沖區(qū)的引用，并變更 consumerBuffer 的引用和 consumerMask 掩碼值，接下來的消費過程便和擴容前沒有差別了。

	scheduleAfterWrite

	現(xiàn)在我們再回到 put 方法的邏輯中，如果向 WriterBuffer 中添加元素成功，則會調(diào)用 scheduleAfterWrite 方法，調(diào)度任務的執(zhí)行：

	
abstract class BoundedLocalCache extends BLCHeader.DrainStatusRef implements LocalCache {

    final ReentrantLock evictionLock = new ReentrantLock();
    // 默認為 ForkJoinPool.commonPool()
    final Executor executor;
    // 該任務在創(chuàng)建緩存時已經(jīng)完成初始化
    final PerformCleanupTask drainBuffersTask;
    
    // 根據(jù)狀態(tài)的變化來調(diào)度執(zhí)行任務
    void scheduleAfterWrite() {
        // 獲取當前 drainStatus，drain 譯為排空，耗盡
        int drainStatus = drainStatusOpaque();
        for (; ; ) {
            // 這里的狀態(tài)機變更需要關注下
            switch (drainStatus) {
                // IDLE 表示當前無任務可做
                case IDLE:
                    // CAS 更新狀態(tài)為 REQUIRED
                    casDrainStatus(IDLE, REQUIRED);
                    // 調(diào)度任務執(zhí)行
                    scheduleDrainBuffers();
                    return;
                // REQUIRED 表示當前有任務需要執(zhí)行
                case REQUIRED:
                    // 調(diào)度任務執(zhí)行
                    scheduleDrainBuffers();
                    return;
                // PROCESSING_TO_IDLE 表示當前任務處理完成后會變成 IDLE 狀態(tài)
                case PROCESSING_TO_IDLE:
                    // 又來了新的任務，則 CAS 操作將它更新為 PROCESSING_TO_REQUIRED 狀態(tài)
                    if (casDrainStatus(PROCESSING_TO_IDLE, PROCESSING_TO_REQUIRED)) {
                        return;
                    }
                    drainStatus = drainStatusAcquire();
                    continue;
                    // PROCESSING_TO_REQUIRED 表示正在處理任務，處理完任務后還有任務需要處理
                case PROCESSING_TO_REQUIRED:
                    return;
                default:
                    throw new IllegalStateException("Invalid drain status: " + drainStatus);
            }
        }
    }

    // 調(diào)度執(zhí)行緩沖區(qū)中的任務
    void scheduleDrainBuffers() {
        // 如果狀態(tài)表示正在有任務處理則返回
        if (drainStatusOpaque() >= PROCESSING_TO_IDLE) {
            return;
        }
        // 注意這里要獲取同步鎖 evictionLock
        if (evictionLock.tryLock()) {
            try {
                // 獲取鎖后再次校驗當前處理狀態(tài)
                int drainStatus = drainStatusOpaque();
                if (drainStatus >= PROCESSING_TO_IDLE) {
                    return;
                }
                // 更新狀態(tài)為 PROCESSING_TO_IDLE
                setDrainStatusRelease(PROCESSING_TO_IDLE);
                // 同步機制保證任何時刻只能有一個線程能夠提交任務
                executor.execute(drainBuffersTask);
            } catch (Throwable t) {
                logger.log(Level.WARNING, "Exception thrown when submitting maintenance task", t);
                maintenance(/* ignored */ null);
            } finally {
                evictionLock.unlock();
            }
        }
    }

}


	

	寫后調(diào)度處理任務（scheduleAfterWrite）會根據(jù)狀態(tài)選擇性執(zhí)行 scheduleDrainBuffers 方法，執(zhí)行該方法時通過同步鎖 evictionLock 保證同時只有一個線程能提交 PerformCleanupTask 任務。這個任務在創(chuàng)建緩存時已經(jīng)被初始化完成了，每次提交任務都會被復用，接下來我們看一下這個任務的具體實現(xiàn)：

	
abstract class BoundedLocalCache extends BLCHeader.DrainStatusRef implements LocalCache {

    // 可重用的任務，用于執(zhí)行 maintenance 方法，避免了使用 ForkJoinPool 來包裝
    static final class PerformCleanupTask extends ForkJoinTask implements Runnable {
        private static final long serialVersionUID = 1L;

        final WeakReference> reference;

        PerformCleanupTask(BoundedLocalCache cache) {
            reference = new WeakReference>(cache);
        }

        @Override
        public boolean exec() {
            try {
                run();
            } catch (Throwable t) {
                logger.log(Level.ERROR, "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(null);
            }
        }
        // ...
    }
}


	

	它的實現(xiàn)非常簡單，其中 reference 字段在調(diào)用構(gòu)造方法時被賦值，引用的是緩存對象本身。當任務被執(zhí)行時，調(diào)用的是 BoundedLocalCache#performCleanUp 方法：

	
abstract class BoundedLocalCache extends BLCHeader.DrainStatusRef implements LocalCache {

    final ReentrantLock evictionLock = new ReentrantLock();
    
    // 執(zhí)行該任務時，也要獲取同步鎖，表示任務只能由一個線程來執(zhí)行
    void performCleanUp(@Nullable Runnable task) {
        evictionLock.lock();
        try {
            // 執(zhí)行維護任務
            maintenance(task);
        } finally {
            evictionLock.unlock();
        }
        rescheduleCleanUpIfIncomplete();
    }

    @GuardedBy("evictionLock")
    void maintenance(@Nullable Runnable task) {
        // 更新狀態(tài)為執(zhí)行中
        setDrainStatusRelease(PROCESSING_TO_IDLE);

        try {
            // 處理讀緩沖區(qū)中的任務
            drainReadBuffer();

            // 處理寫緩沖區(qū)中的任務
            drainWriteBuffer();
            if (task != null) {
                task.run();
            }

            // 處理 key 和 value 的引用
            drainKeyReferences();
            drainValueReferences();

            // 過期和驅(qū)逐策略
            expireEntries();
            evictEntries();

            // “增值” 操作，后續(xù)重點講
            climb();
        } finally {
            // 狀態(tài)不是 PROCESSING_TO_IDLE 或者無法 CAS 更新為 IDLE 狀態(tài)的話，需要更新狀態(tài)為 REQUIRED，該狀態(tài)會再次執(zhí)行維護任務
            if ((drainStatusOpaque() != PROCESSING_TO_IDLE) || !casDrainStatus(PROCESSING_TO_IDLE, IDLE)) {
                setDrainStatusOpaque(REQUIRED);
            }
        }
    }
}


	

	注意在執(zhí)行 performCleanUp 方法時，也需要獲取到同步鎖 evictionLock，那么任務的提交和任務的執(zhí)行也是互斥的。這個執(zhí)行的核心邏輯在 maintenance “維護”方法中，注意這個方法被標記了注解 @GuardedBy("evictionLock")，源碼中還有多個方法也標記了該注解，執(zhí)行這些方法時都要獲取同步鎖，這也是在提醒我們這些方法同時只有由一條線程被執(zhí)行。因為目前關注的是 put 方法，所以重點先看維護方法中 drainWriteBuffer 方法處理寫緩沖區(qū)中任務的邏輯：

	
abstract class BoundedLocalCache extends BLCHeader.DrainStatusRef implements LocalCache {

    static final int NCPU = Runtime.getRuntime().availableProcessors();

    static final int WRITE_BUFFER_MAX = 128 * ceilingPowerOfTwo(NCPU);

    final MpscGrowableArrayQueue writeBuffer;

    @GuardedBy("evictionLock")
    void drainWriteBuffer() {
        // 最大循環(huán)次數(shù)為 writeBuffer 最大容量，直至彈出元素為 null
        for (int i = 0; i <= WRITE_BUFFER_MAX; i++) {
            Runnable task = writeBuffer.poll();
            if (task == null) {
                return;
            }
            task.run();
        }
        // 更新狀態(tài)為 PROCESSING_TO_REQUIRED
        setDrainStatusOpaque(PROCESSING_TO_REQUIRED);
    }
}


	

	執(zhí)行邏輯非常簡單，在獲取到同步鎖之后，在 WriteBuffer 中獲取要被執(zhí)行的任務并執(zhí)行。在這里我們能發(fā)現(xiàn)“SC 單消費者”的實現(xiàn)使用 同步鎖的機制保證同時只能有一個消費者消費緩沖區(qū)中的任務。在上文中我們已經(jīng)知道，調(diào)用 put 方法時向緩沖區(qū) WriteBuffer 中添加的任務為 AddTask，下面我們看一下該任務的實現(xiàn)：

	
abstract class BoundedLocalCache extends BLCHeader.DrainStatusRef implements LocalCache {

    static final long MAXIMUM_CAPACITY = Long.MAX_VALUE - Integer.MAX_VALUE;

    final class AddTask implements Runnable {
        final Node node;
        // 節(jié)點權重
        final int weight;

        AddTask(Node node, int weight) {
            this.weight = weight;
            this.node = node;
        }

        @Override
        @GuardedBy("evictionLock")
        @SuppressWarnings("FutureReturnValueIgnored")
        public void run() {
            // 是否指定了驅(qū)逐策略
            if (evicts()) {
                // 更新緩存權重和窗口區(qū)權重
                setWeightedSize(weightedSize() + weight);
                setWindowWeightedSize(windowWeightedSize() + weight);
                // 更新節(jié)點的 policyWeight，該字段只有在自定了權重計算規(guī)則時才有效
                // 否則像只定義了固定容量的驅(qū)逐策略，使用默認元素權重為 1 是不需要關注該字段的
                node.setPolicyWeight(node.getPolicyWeight() + weight);

                // 檢測當前總權重是否超過一半的最大容量
                long maximum = maximum();
                if (weightedSize() >= (maximum >>> 1)) {
                    // 如果超過最大容量
                    if (weightedSize() > MAXIMUM_CAPACITY) {
                        // 執(zhí)行驅(qū)逐操作
                        evictEntries();
                    } else {
                        // 延遲加載頻率草圖 frequencySketch 數(shù)據(jù)結(jié)構(gòu)，用于統(tǒng)計元素訪問頻率
                        long capacity = isWeighted() ? data.mappingCount() : maximum;
                        frequencySketch().ensureCapacity(capacity);
                    }
                }

                // 更新頻率統(tǒng)計信息
                K key = node.getKey();
                if (key != null) {
                    // 因為頻率草圖數(shù)據(jù)結(jié)構(gòu)具有延遲加載機制（權重超過半數(shù)）
                    // 所以實際上在元素權重還未過半未完成初始化時，調(diào)用 increment 是沒有作用的
                    frequencySketch().increment(key);
                }

                // 增加未命中樣本數(shù)
                setMissesInSample(missesInSample() + 1);
            }

            // 同步檢測節(jié)點是否還有效
            boolean isAlive;
            synchronized (node) {
                isAlive = node.isAlive();
            }
            if (isAlive) {
                // 寫后過期策略
                if (expiresAfterWrite()) {
                    writeOrderDeque().offerLast(node);
                }
                // 過期策略
                if (expiresVariable()) {
                    timerWheel().schedule(node);
                }
                // 驅(qū)逐策略
                if (evicts()) {
                    // 如果權重比配置的最大權重大
                    if (weight > maximum()) {
                        // 執(zhí)行固定權重（RemovalCause.SIZE）的驅(qū)逐策略
                        evictEntry(node, RemovalCause.SIZE, expirationTicker().read());
                    }
                    // 如果權重超過窗口區(qū)最大權重，則將其放在窗口區(qū)頭節(jié)點
                    else if (weight > windowMaximum()) {
                        accessOrderWindowDeque().offerFirst(node);
                    }
                    // 否則放在窗口區(qū)尾節(jié)點
                    else {
                        accessOrderWindowDeque().offerLast(node);
                    }
                }
                // 訪問后過期策略
                else if (expiresAfterAccess()) {
                    accessOrderWindowDeque().offerLast(node);
                }
            }

            // 處理異步計算
            if (isComputingAsync(node)) {
                synchronized (node) {
                    if (!Async.isReady((CompletableFuture) node.getValue())) {
                        long expirationTime = expirationTicker().read() + ASYNC_EXPIRY;
                        setVariableTime(node, expirationTime);
                        setAccessTime(node, expirationTime);
                        setWriteTime(node, expirationTime);
                    }
                }
            }
        }
    }
}


	

	根據(jù)注釋很容易理解該方法的作用，因為我們目前對緩存只定義了固定容量的驅(qū)逐策略，所以我們需要在看一下 evictEntry 方法：

	
abstract class BoundedLocalCache extends BLCHeader.DrainStatusRef implements LocalCache {

    final ConcurrentHashMap> data;
    
    @GuardedBy("evictionLock")
    @SuppressWarnings({"GuardedByChecker", "NullAway", "PMD.CollapsibleIfStatements"})
    boolean evictEntry(Node node, RemovalCause cause, long now) {
        K key = node.getKey();
        @SuppressWarnings("unchecked")
        V[] value = (V[]) new Object[1];
        boolean[] removed = new boolean[1];
        boolean[] resurrect = new boolean[1];
        Object keyReference = node.getKeyReference();
        RemovalCause[] actualCause = new RemovalCause[1];

        data.computeIfPresent(keyReference, (k, n) -> {
            if (n != node) {
                return n;
            }
            synchronized (n) {
                value[0] = n.getValue();

                // key 或 value 為 null，這種情況下可能使用了 Caffeine.weakKeys, Caffeine.weakValues, or Caffeine.softValues
                // 導致被垃圾回收了
                if ((key == null) || (value[0] == null)) {
                    // 標記實際失效原因為垃圾回收 
                    actualCause[0] = RemovalCause.COLLECTED;
                }
                // 如果原因為垃圾回收，記錄 resurrect 復活標記為 true
                else if (cause == RemovalCause.COLLECTED) {
                    resurrect[0] = true;
                    return n;
                }
                // 否則記錄入?yún)⒅械脑?                else {
                    actualCause[0] = cause;
                }

                // 過期驅(qū)逐策略判斷
                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;
                    }
                }
                // 固定容量驅(qū)逐策略
                else if (actualCause[0] == RemovalCause.SIZE) {
                    int weight = node.getWeight();
                    if (weight == 0) {
                        resurrect[0] = true;
                        return n;
                    }
                }

                // 通知驅(qū)逐策略監(jiān)聽器，調(diào)用它的方法
                notifyEviction(key, value[0], actualCause[0]);
                // 將該 key 對應的刷新策略失效
                discardRefresh(keyReference);
                // 標記該節(jié)點被驅(qū)逐
                removed[0] = true;
                // 退休準備被垃圾回收
                node.retire();
            }
            return null;
        });

        // 如果復活標記為 true 那么不被移除
        if (resurrect[0]) {
            return false;
        }

        // 節(jié)點已經(jīng)要被驅(qū)逐
        // 如果在窗口區(qū)，那么直接從窗口區(qū)移除
        if (node.inWindow() && (evicts() || expiresAfterAccess())) {
            accessOrderWindowDeque().remove(node);
        }
        // 如果沒在窗口區(qū)
        else if (evicts()) {
            // 在試用區(qū)直接在試用區(qū)移除
            if (node.inMainProbation()) {
                accessOrderProbationDeque().remove(node);
            }
            // 在保護區(qū)則直接從保護區(qū)移除
            else {
                accessOrderProtectedDeque().remove(node);
            }
        }
        // 將寫后失效和時間輪中關于該節(jié)點的元素移除
        if (expiresAfterWrite()) {
            writeOrderDeque().remove(node);
        } else if (expiresVariable()) {
            timerWheel().deschedule(node);
        }

        // 同步機制將 node 置為 dead
        synchronized (node) {
            logIfAlive(node);
            makeDead(node);
        }

        if (removed[0]) {
            // 節(jié)點被移除監(jiān)控計數(shù)和節(jié)點移除通知回調(diào)
            statsCounter().recordEviction(node.getWeight(), actualCause[0]);
            notifyRemoval(key, value[0], actualCause[0]);
        }

        return true;
    }
}


	

	該方法比較簡單，是將節(jié)點進行驅(qū)逐的邏輯，在后文中它會被多次復用，需要留一個印象。回到 AddTask 任務的邏輯中，當被添加的元素權重超過最大權重限制時會被直接移除。這種特殊情況試用于指定了權重計算策略的緩存，如果只指定了固定容量，元素權重默認為 1，所以不會直接超過最大緩存數(shù)量限制。

	現(xiàn)在我們已經(jīng)將 put 方法中向緩存中添加元素的邏輯介紹完了，接下來需要關注 put 方法中對已存在的相同 key 值元素的處理邏輯：

	
abstract class BoundedLocalCache extends BLCHeader.DrainStatusRef implements LocalCache {

    static final int MAX_PUT_SPIN_WAIT_ATTEMPTS = 1024 - 1;

    static final long EXPIRE_WRITE_TOLERANCE = TimeUnit.SECONDS.toNanos(1);
    
    final ConcurrentHashMap> data;
    
    @Nullable
    V put(K key, V value, Expiry expiry, boolean onlyIfAbsent) {
        requireNonNull(key);
        requireNonNull(value);

        Node node = null;
        long now = expirationTicker().read();
        int newWeight = weigher.weigh(key, value);
        Object lookupKey = nodeFactory.newLookupKey(key);
        for (int attempts = 1; ; attempts++) {
            Node prior = data.get(lookupKey);
            if (prior == null) {
                // ... 
            }

            // 元素被讀到之后可能已經(jīng)被驅(qū)逐了
            if (!prior.isAlive()) {
                // 自旋嘗試重新從 ConcurrentHashMap 中獲取，再獲取時如果為 null 則執(zhí)行新增邏輯
                if ((attempts & MAX_PUT_SPIN_WAIT_ATTEMPTS) != 0) {
                    Thread.onSpinWait();
                    continue;
                }
                // 如果自旋嘗試后元素仍未被刪除，校驗元素是否處于存活狀態(tài)
                // 如果處于非存活狀態(tài)，那么可能這個元素已經(jīng)被破壞，無法被移除，拋出異常
                data.computeIfPresent(lookupKey, (k, n) -> {
                    requireIsAlive(key, n);
                    return n;
                });
                continue;
            }

            V oldValue;
            // 新的過期時間
            long varTime;
            int oldWeight;
            boolean expired = false;
            boolean mayUpdate = true;
            boolean exceedsTolerance = false;
            // 為元素加同步鎖
            synchronized (prior) {
                // 如果此時元素已經(jīng)失效了，那么需要重新循環(huán)
                if (!prior.isAlive()) {
                    continue;
                }
                oldValue = prior.getValue();
                oldWeight = prior.getWeight();
                // oldValue 為 null 證明它被垃圾回收器回收了
                if (oldValue == null) {
                    // 記錄元素創(chuàng)建后的過期時間
                    varTime = expireAfterCreate(key, value, expiry, now);
                    // 驅(qū)逐監(jiān)聽器回調(diào)
                    notifyEviction(key, null, RemovalCause.COLLECTED);
                }
                // 如果元素已經(jīng)過期了
                else if (hasExpired(prior, now)) {
                    // 標記過期標志為 true
                    expired = true;
                    // 記錄元素創(chuàng)建后的過期時間并回調(diào)驅(qū)逐監(jiān)聽器
                    varTime = expireAftexpireAfterCreateerCreate(key, value, expiry, now);
                    notifyEviction(key, oldValue, RemovalCause.EXPIRED);
                }
                // onlyInAbsent 為 true 時不會對已存在 key 的值進行修改
                else if (onlyIfAbsent) {
                    mayUpdate = false;
                    // 記錄元素讀后過期時間
                    varTime = expireAfterRead(prior, key, value, expiry, now);
                } else {
                    // 記錄元素修改后過期時間
                    varTime = expireAfterUpdate(prior, key, value, expiry, now);
                }

                // 需要修改原有 key 的 value 值
                if (mayUpdate) {
                    exceedsTolerance =
                            // 配置了寫后過期策略且已經(jīng)超過寫后時間的容忍范圍
                            (expiresAfterWrite() && (now - prior.getWriteTime()) > EXPIRE_WRITE_TOLERANCE)
                                    // 或者配置了可變時間過期策略同樣判斷是否超過時間的容忍范圍
                                    || (expiresVariable() && Math.abs(varTime - prior.getVariableTime()) > EXPIRE_WRITE_TOLERANCE);

                    // 更新值，更新權重，更新寫時間
                    prior.setValue(value, valueReferenceQueue());
                    prior.setWeight(newWeight);
                    setWriteTime(prior, now);

                    // 寫后刷新策略失效
                    discardRefresh(prior.getKeyReference());
                }

                // 更新過期時間
                setVariableTime(prior, varTime);
                // 更新訪問時間
                setAccessTime(prior, now);
            }

            // 根據(jù)不同的情況回調(diào)不同的監(jiān)聽器
            if (expired) {
                notifyRemoval(key, oldValue, RemovalCause.EXPIRED);
            } else if (oldValue == null) {
                notifyRemoval(key, /* oldValue */ null, RemovalCause.COLLECTED);
            } else if (mayUpdate) {
                notifyOnReplace(key, oldValue, value);
            }

            // 計算寫后權重變化
            int weightedDifference = mayUpdate ? (newWeight - oldWeight) : 0;
            // 如果 oldValue 已經(jīng)被回收 或 權重修改前后發(fā)生變更 或 已經(jīng)過期，添加更新任務
            if ((oldValue == null) || (weightedDifference != 0) || expired) {
                afterWrite(new UpdateTask(prior, weightedDifference));
            }
            // 如果超過了時間容忍范圍，添加更新任務
            else if (!onlyIfAbsent && exceedsTolerance) {
                afterWrite(new UpdateTask(prior, weightedDifference));
            } else {
                // 沒有超過時間容忍范圍，更新寫時間
                if (mayUpdate) {
                    setWriteTime(prior, now);
                }
                // 處理讀后操作
                afterRead(prior, now, /* recordHit */ false);
            }

            return expired ? null : oldValue;
        }
    }
}


	

	對于已有元素的變更，會對節(jié)點添加同步鎖，更新它的權重等一系列變量，如果超過 1s 的時間容忍范圍，則會添加 UpdateTask 更新任務，至于處理讀后操作 afterRead 在讀方法中再去介紹。接下來我們需要重新再看一下 afterWrite 方法，其中有部分我們在上文中沒有介紹的邏輯：

	
abstract class BoundedLocalCache extends BLCHeader.DrainStatusRef implements LocalCache {

    final ReentrantLock evictionLock;
    
    void afterWrite(Runnable task) {
        // 這段邏輯我們在看 AddTask 的邏輯時已經(jīng)看過了，所以略過
        for (int i = 0; i < WRITE_BUFFER_RETRIES; i++) {
            if (writeBuffer.offer(task)) {
                scheduleAfterWrite();
                return;
            }
            scheduleDrainBuffers();
            Thread.onSpinWait();
        }

        // 以下邏輯用于解決在重試了 100 次后仍然寫入失敗的問題，它會嘗試獲取 evictionLock 同步鎖
        // 直接同步執(zhí)行“維護”方法并執(zhí)行當前任務，但是它并無法解決某個寫入操作執(zhí)行時間很長的問題
        // 發(fā)生這種情況的原因可能是由于執(zhí)行器的所有線程都很忙（可能是寫入此緩存），寫入速率大大超過了消耗速率，優(yōu)先級反轉(zhuǎn)，或者執(zhí)行器默默地丟棄了維護任務
        lock();
        try {
            maintenance(task);
        } catch (RuntimeException e) {
            logger.log(Level.ERROR, "Exception thrown when performing the maintenance task", e);
        } finally {
            evictionLock.unlock();
        }
        // 重新調(diào)度異步維護任務，確保維護操作能及時執(zhí)行
        rescheduleCleanUpIfIncomplete();
    }

    void lock() {
        long remainingNanos = WARN_AFTER_LOCK_WAIT_NANOS;
        long end = System.nanoTime() + remainingNanos;
        boolean interrupted = false;
        try {
            for (;;) {
                try {
                    if (evictionLock.tryLock(remainingNanos, TimeUnit.NANOSECONDS)) {
                        return;
                    }
                    logger.log(Level.WARNING, "The cache is experiencing excessive wait times for acquiring "
                            + "the eviction lock. This may indicate that a long-running computation has halted "
                            + "eviction when trying to remove the victim entry. Consider using AsyncCache to "
                            + "decouple the computation from the map operation.", new TimeoutException());
                    evictionLock.lock();
                    return;
                } catch (InterruptedException e) {
                    remainingNanos = end - System.nanoTime();
                    interrupted = true;
                }
            }
        } finally {
            if (interrupted) {
                Thread.currentThread().interrupt();
            }
        }
    }

    // 調(diào)用同步的維護方法時，可能發(fā)生獲取鎖超時，那么再重新開啟一個異步維護調(diào)度
    void rescheduleCleanUpIfIncomplete() {
        // 校驗是否有任務需要被執(zhí)行
        if (drainStatusOpaque() != REQUIRED) {
            return;
        }
        
        // 默認線程池調(diào)度任務執(zhí)行，這個方法我們在上文中已經(jīng)詳細介紹過
        if (executor == ForkJoinPool.commonPool()) {
            scheduleDrainBuffers();
            return;
        }
        
        // 如果自定義了線程池，那么會使用自定義的線程池進行處理
        var pacer = pacer();
        if ((pacer != null) && !pacer.isScheduled() && evictionLock.tryLock()) {
            try {
                if ((drainStatusOpaque() == REQUIRED) && !pacer.isScheduled()) {
                    pacer.schedule(executor, drainBuffersTask, expirationTicker().read(), Pacer.TOLERANCE);
                }
            } finally {
                evictionLock.unlock();
            }
        }
    }
}


	

	寫后操作除了在添加任務到緩沖區(qū)成功后會執(zhí)行維護方法，添加失?。ㄗC明寫入操作非常頻繁）依然會嘗試同步執(zhí)行維護方法和發(fā)起異步維護，用于保證緩存中的任務能夠被及時執(zhí)行，使緩存中元素都處于“預期”狀態(tài)中。接下來我們在看一下 UpdateTask 更新任務的邏輯：

	
abstract class BoundedLocalCache extends BLCHeader.DrainStatusRef implements LocalCache {

    final class UpdateTask implements Runnable {
        final int weightDifference;
        final Node node;

        public UpdateTask(Node node, int weightDifference) {
            this.weightDifference = weightDifference;
            this.node = node;
        }

        @Override
        @GuardedBy("evictionLock")
        public void run() {
            // 寫后過期和自定義過期邏輯
            if (expiresAfterWrite()) {
                reorder(writeOrderDeque(), node);
            } else if (expiresVariable()) {
                timerWheel().reschedule(node);
            }
            // 指定了驅(qū)逐策略
            if (evicts()) {
                // 變更節(jié)點權重
                int oldWeightedSize = node.getPolicyWeight();
                node.setPolicyWeight(oldWeightedSize + weightDifference);
                // 如果是窗口區(qū)節(jié)點
                if (node.inWindow()) {
                    // 更新窗口區(qū)權重
                    setWindowWeightedSize(windowWeightedSize() + weightDifference);
                    // 節(jié)點權重超過最大權重限制，直接驅(qū)逐
                    if (node.getPolicyWeight() > maximum()) {
                        evictEntry(node, RemovalCause.SIZE, expirationTicker().read());
                    }
                    // 節(jié)點權重比窗口區(qū)最大值小
                    else if (node.getPolicyWeight() <= windowMaximum()) {
                        onAccess(node);
                    }
                    // 窗口區(qū)包含該節(jié)點且該節(jié)點的權重大于窗口最大權重，則放到頭節(jié)點
                    else if (accessOrderWindowDeque().contains(node)) {
                        accessOrderWindowDeque().moveToFront(node);
                    }
                }
                // 如果是試用區(qū)節(jié)點
                else if (node.inMainProbation()) {
                    // 節(jié)點權重比最大權重限制小
                    if (node.getPolicyWeight() <= maximum()) {
                        onAccess(node);
                    }
                    // 否則將該節(jié)點驅(qū)逐
                    else {
                        evictEntry(node, RemovalCause.SIZE, expirationTicker().read());
                    }
                }
                // 如果是保護區(qū)節(jié)點
                else if (node.inMainProtected()) {
                    // 更新保護區(qū)權重
                    setMainProtectedWeightedSize(mainProtectedWeightedSize() + weightDifference);
                    // 同樣的邏輯
                    if (node.getPolicyWeight() <= maximum()) {
                        onAccess(node);
                    } else {
                        evictEntry(node, RemovalCause.SIZE, expirationTicker().read());
                    }
                }

                // 更新緩存權重大小
                setWeightedSize(weightedSize() + weightDifference);
                // 更新完成后超過最大權重限制執(zhí)行驅(qū)逐操作
                if (weightedSize() > MAXIMUM_CAPACITY) {
                    evictEntries();
                }
            }
            // 配置了訪問后過期
            else if (expiresAfterAccess()) {
                onAccess(node);
            }
        }
    }

    @GuardedBy("evictionLock")
    void onAccess(Node node) {
        if (evicts()) {
            K key = node.getKey();
            if (key == null) {
                return;
            }
            // 更新訪問頻率
            frequencySketch().increment(key);
            // 如果節(jié)點在窗口區(qū)，則將其移動到尾節(jié)點
            if (node.inWindow()) {
                reorder(accessOrderWindowDeque(), node);
            }
            // 在試用區(qū)的節(jié)點執(zhí)行 reorderProbation 方法，可能會將該節(jié)點從試用區(qū)晉升到保護區(qū)
            else if (node.inMainProbation()) {
                reorderProbation(node);
            }
            // 否則移動到保護區(qū)的尾結(jié)點
            else {
                reorder(accessOrderProtectedDeque(), node);
            }
            // 更新命中量
            setHitsInSample(hitsInSample() + 1);
        }
        // 配置了訪問過期策略
        else if (expiresAfterAccess()) {
            reorder(accessOrderWindowDeque(), node);
        }
        // 配置了自定義時間過期策略
        if (expiresVariable()) {
            timerWheel().reschedule(node);
        }
    }

    static  void reorder(LinkedDeque> deque, Node node) {
        // 如果節(jié)點存在，將其移動到尾結(jié)點
        if (deque.contains(node)) {
            deque.moveToBack(node);
        }
    }

    @GuardedBy("evictionLock")
    void reorderProbation(Node node) {
        // 檢查試用區(qū)是否包含該節(jié)點，不包含則證明已經(jīng)被移除，則不處理
        if (!accessOrderProbationDeque().contains(node)) {
            return;
        }
        // 檢查節(jié)點的權重是否超過保護區(qū)最大值
        else if (node.getPolicyWeight() > mainProtectedMaximum()) {
            // 如果超過，將該節(jié)點移動到 試用區(qū) 尾巴節(jié)點，保證超重的節(jié)點不會被移動到保護區(qū)
            reorder(accessOrderProbationDeque(), node);
            return;
        }

        // 更新保護區(qū)權重大小
        setMainProtectedWeightedSize(mainProtectedWeightedSize() + node.getPolicyWeight());
        // 在試用區(qū)中移除該節(jié)點
        accessOrderProbationDeque().remove(node);
        // 在保護區(qū)尾節(jié)點中添加
        accessOrderProtectedDeque().offerLast(node);
        // 將該節(jié)點標記為保護區(qū)節(jié)點
        node.makeMainProtected();
    }
}


	

	UpdateTask 修改任務負責變更權重值，并更新節(jié)點所在隊列的順序和訪問頻率，這里我們也能發(fā)現(xiàn)，這三個區(qū)域的隊列采用了 LRU 算法，一般情況下，最新被訪問的元素會被移動到尾節(jié)點。到現(xiàn)在，向有固定容量限制的緩存中調(diào)用 put 方法添加元素的邏輯基本已經(jīng)介紹完了，目前對 Caffeine 緩存的了解程度如下所示：

	

	put 添加元素時會先直接添加到 ConcurrentHashMap 中，并在 WriteBuffer 中添加 AddTask/UpdateTask 任務，WriteBuffer 是一個 MPSC 的緩沖區(qū)，添加成功后會有加鎖的同步機制在默認的 ForkJoinPool.commonPool() 線程池中提交 PerformCleanupTask 任務，PerformCleanupTask 任務的主要作用是執(zhí)行 maintenance 維護方法，該方法執(zhí)行前需要先獲取同步鎖，單線程消費 WriteBuffer 中的任務。執(zhí)行 AddTask 任務時會將元素先添加到窗口區(qū)，如果是 UpdateTask，它會修改三個不同區(qū)域的雙端隊列，這些隊列采用LRU算法，最新被訪問的元素會被放在尾節(jié)點處，并且試用區(qū)的元素被訪問后會被晉升到保護區(qū)尾節(jié)點，元素對應的訪問頻率也會在頻率草圖中更新，如果被添加的節(jié)點權重超過緩存最大權重會直接被驅(qū)逐。（目前維護方法中除了 drainWriteBuffer 方法外，其他步驟還未詳細解釋，之后會在后文中不斷完善）

	

	審核編輯 黃宇