??在上一個文章中分析BufferAsyncEmitter時，說到BufferAsyncEmitter使用了SpscLinkedArrayQueue隊列來緩存數(shù)據(jù)。當時在文末時，只是簡單的提了一句，并沒有詳細介紹SpscLinkedArrayQueue隊列的原理，在本文，將詳細介紹SpscLinkedArrayQueue隊列的神奇之處。

1.數(shù)據(jù)結(jié)構(gòu)

??在分析SpscLinkedArrayQueue隊列之前，我們先來了解一下SpscLinkedArrayQueue的第一個神奇之處，它神奇的數(shù)據(jù)結(jié)構(gòu)。
??SpscLinkedArrayQueue的數(shù)據(jù)結(jié)構(gòu)主要神奇在它既不是傳統(tǒng)的數(shù)組，又不是傳統(tǒng)的鏈表，而是數(shù)組+鏈表。我說的好像過于玄乎了，還是具體來看看吧。

2.成員變量

??了解了SpscLinkedArrayQueue的數(shù)據(jù)結(jié)構(gòu)，我們開始正式來分析SpscLinkedArrayQueue，當然，我們還是從它的成員變量開始，來看看它成員變量有哪些，分別表示什么含義。

3.構(gòu)造方法

??我們先來看看SpscLinkedArrayQueue的構(gòu)造方法，看看它為我們做了哪些初始化。

    public SpscLinkedArrayQueue(final int bufferSize) {
        int p2capacity = Pow2.roundToPowerOfTwo(Math.max(8, bufferSize));
        int mask = p2capacity - 1;
        AtomicReferenceArray<Object> buffer = new AtomicReferenceArray<Object>(p2capacity + 1);
        producerBuffer = buffer;
        producerMask = mask;
        adjustLookAheadStep(p2capacity);
        consumerBuffer = buffer;
        consumerMask = mask;
        producerLookAhead = mask - 1; // we know it's all empty to start with
        soProducerIndex(0L);
    }

??初始化的東西還真的不少，但是我們這里挑比較重要的說。
??首先，是對傳遞進來的bufferSize進行了重新計算的操作，讓它始終為2^n。也就是如下的代碼。

 int p2capacity = Pow2.roundToPowerOfTwo(Math.max(8, bufferSize));

??其實這個調(diào)整為2^n的操作也不是什么騷操作，就是最普通的位運算。我們先來看看Pow2的roundToPowerOfTwo方法里面究竟是怎么計算的。

    public static int roundToPowerOfTwo(final int value) {
        return 1 << (32 - Integer.numberOfLeadingZeros(value - 1));
    }

??是不是一臉懵逼？其實我們這樣來考慮，一個int為32bit，其中32個bit中，有且只有1，那么這個數(shù)字肯定是2^n。如果我們這樣想的話，就非常的簡單。
??在這個構(gòu)造方法里面，還有幾個地方需要我們注意。

        AtomicReferenceArray<Object> buffer = new AtomicReferenceArray<Object>(p2capacity + 1);

??數(shù)組的容量為2 ^ n + 1，這個得需要我們特別注意，如果這里沒有注意，后面就會有理解上的誤差。
??還有就是需要注意producerLookAhead:

        producerLookAhead = mask - 1; // we know it's all empty to start with

?? producerLookAhead為mask - 1,也就是 2 ^ n - 2,至于為什么是，這里也有很大的學(xué)問咯。

4. offer方法

??在SpscLinkedArrayQueue中，offer方法和poll方法是占據(jù)非常重要的地位，所以分析這兩個方法是非常有必要的，當然我們也可以通過分析這兩個方法來了解SpscLinkedArrayQueue的本質(zhì)。我們首先來看看offer方法。

    public boolean offer(final T e) {
        if (null == e) {
            throw new NullPointerException("Null is not a valid element");
        }
        // local load of field to avoid repeated loads after volatile reads
        final AtomicReferenceArray<Object> buffer = producerBuffer;
        final long index = lpProducerIndex();
        final int mask = producerMask;
        final int offset = calcWrappedOffset(index, mask);
        if (index < producerLookAhead) {
            return writeToQueue(buffer, e, index, offset);
        } else {
            final int lookAheadStep = producerLookAheadStep;
            // go around the buffer or resize if full (unless we hit max capacity)
            int lookAheadElementOffset = calcWrappedOffset(index + lookAheadStep, mask);
            if (null == lvElement(buffer, lookAheadElementOffset)) { // LoadLoad
                producerLookAhead = index + lookAheadStep - 1; // joy, there's plenty of room
                return writeToQueue(buffer, e, index, offset);
            } else if (null == lvElement(buffer, calcWrappedOffset(index + 1, mask))) { // buffer is not full
                return writeToQueue(buffer, e, index, offset);
            } else {
                resize(buffer, index, offset, e, mask); // add a buffer and link old to new
                return true;
            }
        }
    }

??整個offer過程，我將它分為兩種情形。
??1.第一情形是index還未超過producerLookAhead，這種情形下，直接通過插入到相應(yīng)位置當中去就行了。
??2.第二情形就是index超過producerLookAhead，這種情形比較復(fù)雜，既要考慮到producerLookAheadStep的存在，又要考慮到是否達到必須擴容的條件。

(1).當index還未超過producerLookAhead

??這種情況比較簡單，在這里主要講解offset的計算。offset的計算主要通過calcWrappedOffset方法來計算,我們來看看這個方法到底為我們做了什么吧。

    private static int calcWrappedOffset(long index, int mask) {
        return calcDirectOffset((int)index & mask);
    }
    private static int calcDirectOffset(int index) {
        return index;
    }

??方法比較簡單，歸根結(jié)底就是index & mask。這個計算有什么特殊的含義嗎？當然有了，還記得mask的值為多少嗎？mask為2^n - 1,index & mask相當于index % 2 ^ n。
??在統(tǒng)一說明成員變量那里，我曾說過index是只增不減的，這里的計算就能體現(xiàn)出來。當index超出了這個數(shù)組的長度時，通過mask來取模又能定位一個位置。
??但是在哪種情況下可能會出現(xiàn)index超過數(shù)組的長度呢？我們從producerLookAhead這個變量里面尋找答案，之前也說過，producerLookAhead是index能達到的最大位置。當生產(chǎn)者產(chǎn)生數(shù)組已經(jīng)達到了數(shù)組末尾時，此時還不能立即進行擴容，而是得看看這個數(shù)組節(jié)點的前面部分是否已經(jīng)被消費者消費了，如果已經(jīng)消費了，我們此時可以往前部分產(chǎn)生數(shù)據(jù)，而沒必要去擴容。這個相當于是一個循環(huán)隊列的設(shè)計思想。

(2).當index超過producerLookAhead

??這種情形下，我們還可以分為三種小情形：
??1.判斷當前index + producerLookAheadStep的位置上是否為null，如果為null，那么表示producerLookAheadStep可以增大，同時index也可以繼續(xù)增大到新的producerLookAheadStep；如果不為null，就是第二種情形
??2.如果第一種情形的條件不能達到，那么看看是否存在有消費者被消費了的位置，但是還不足producerLookAheadStep。如果存在，就進行index + 1操作，同時producerLookAheadStep不能變；如果不存在，那么進行第三種情形的操作。
??3.如果上面兩種情形都不成立的話，那么進行此種情形的操作，那就是擴容。
??這里先只講解前面兩種情形，擴容操作比較特殊，單獨來講。
??先來看看第一種情形：

            final int lookAheadStep = producerLookAheadStep;
            // go around the buffer or resize if full (unless we hit max capacity)
            int lookAheadElementOffset = calcWrappedOffset(index + lookAheadStep, mask);
            if (null == lvElement(buffer, lookAheadElementOffset)) { // LoadLoad
                producerLookAhead = index + lookAheadStep - 1; // joy, there's plenty of room
                return writeToQueue(buffer, e, index, offset);
            } 

??這里的邏輯非常簡單，簡單來說就是我們上面所說的。先是通過取模計算了index + producerLookAheadStep的offset，然后判斷offset位置上是否為null，如果為null，表示達到producerLookAhead增大的條件，然后就是，producerLookAhead增大了lookAheadStep - 1,雖然這里是index + lookAheadStep - 1,但是我認為，在只要符合這種條件的，index等于producerLookAhead。因為如果index不等于producerLookAhead,肯定不是第一次進入這個判斷語句，而第一次進來的話，如果符合的話，就已經(jīng)擴容了，就變成了index還未超過producerLookAhead的情況。
??再來看看第二種情形：

 else if (null == lvElement(buffer, calcWrappedOffset(index + 1, mask))) { // buffer is not full
                return writeToQueue(buffer, e, index, offset);
            } 

??這種情況比較簡單，此時index只是簡單的做加1操作。這種情形就像是，前面步子邁大了，扯著X了，開始一步一步的邁????。
??第三種情形便是我們的擴容操作，這個也是我們SpscLinkedArrayQueue神奇之處之一。接下來，我們慢慢看這個擴容騷操作。

5.擴容騷操作

??說這個騷，不是一般的騷，不得不佩服大佬們寫的代碼。
??在這里，我們會知道數(shù)組的容量為什么為2 ^ n + 1，而不是2 ^ n。
??首先，我們還是先來看看整個擴容的過程。

    private void resize(final AtomicReferenceArray<Object> oldBuffer, final long currIndex, final int offset, final T e,
            final long mask) {
        final int capacity = oldBuffer.length();
        final AtomicReferenceArray<Object> newBuffer = new AtomicReferenceArray<Object>(capacity);
        producerBuffer = newBuffer;
        producerLookAhead = currIndex + mask - 1;
        soElement(newBuffer, offset, e);// StoreStore
        soNext(oldBuffer, newBuffer);
        soElement(oldBuffer, offset, HAS_NEXT); // new buffer is visible after element is
                                                                 // inserted
        soProducerIndex(currIndex + 1);// this ensures correctness on 32bit platforms
    }

??整個方法的比較簡單，就是創(chuàng)建了一個新的AtomicReferenceArray對象，跟原對象容量是一模一樣的，然后通過soNext方法將兩個節(jié)點連接起來，我們來看看放的位置：

    private void soNext(AtomicReferenceArray<Object> curr, AtomicReferenceArray<Object> next) {
        soElement(curr, calcDirectOffset(curr.length() - 1), next);
    }

??沒錯，將新的AtomicReferenceArray對象放在了原對象的最后一位，這樣相當于是將兩個節(jié)點起來，也應(yīng)證我們前面的總結(jié)。

        soElement(newBuffer, offset, e);// StoreStore
        soElement(oldBuffer, offset, HAS_NEXT); // new buffer is visible after element is

??然后接下來，你會看到極其騷的操作，將原數(shù)組的offset位置上設(shè)置為HAS_NEXT，同時在新數(shù)組的offset位置上放入需要加入的數(shù)據(jù)。這樣做有什么好處呢？
??這個有利于poll操作，當poll操作操作到這個位置上時，發(fā)現(xiàn)是HAS_NEXT，會到下一個節(jié)點的offset位置上去尋找。因為offer操作也是offset開始，所以必須保證poll操作從offer操作開始的地方進行。
??整個過程差不多就是這樣的，接下來我們分析上面的兩個問題。

(1).數(shù)組容量為什么為2 ^ n + 1

??我們知道index是自增不減的，同時offset是通過index & mask計算得到的。同時HAS_NEXT坐標也是offset,所以，我們可以知道，在擴容是，HAS_NEXT的坐標是不定的。那數(shù)組容量為129為例來說，HAS_NEXT可能出現(xiàn)在0 ~ 127任何一個位置，但是128位置上始終是為next指針準備。
??這是為什么？我們可以這樣來理解，將HAS_NEXT當成一個特殊的數(shù)據(jù)，它也屬于生產(chǎn)者生成的數(shù)據(jù)其中一個，但是next指針不可能當成其中一個，因為消費者不能正確找到next指針，除非將整個數(shù)組遍歷，顯然這是一個愚蠢的做法。所以next指針放在一個固定位置上，哪個位置不可能被占據(jù)呢？在0 ~ 127的范圍里面顯然是不可能的，所以得單獨找一個不在0 ~ 127范圍里面的位置，哪個位置呢？肯定是128，所以數(shù)組容量才為2 ^ n + 1。

6.poll方法

??看完了offer方法，現(xiàn)在我們再來看看poll方法。

    public T poll() {
        // local load of field to avoid repeated loads after volatile reads
        final AtomicReferenceArray<Object> buffer = consumerBuffer;
        final long index = lpConsumerIndex();
        final int mask = consumerMask;
        final int offset = calcWrappedOffset(index, mask);
        final Object e = lvElement(buffer, offset);// LoadLoad
        boolean isNextBuffer = e == HAS_NEXT;
        if (null != e && !isNextBuffer) {
            soElement(buffer, offset, null);// StoreStore
            soConsumerIndex(index + 1);// this ensures correctness on 32bit platforms
            return (T) e;
        } else if (isNextBuffer) {
            return newBufferPoll(lvNextBufferAndUnlink(buffer, mask + 1), index, mask);
        }

        return null;
    }

??整個poll方法比較簡單，通過獲取offset的數(shù)據(jù)，先判斷是否是HAS_NEXT,如果不是那么就可以取出；如果是的話，那么就到mask + 1的位置找到下一個節(jié)點，再到下一個節(jié)點的offset位置上去取數(shù)據(jù)。

7.總結(jié)

??總的來說，SpscLinkedArrayQueue涉及過于神奇。這里我來做一個簡單的總結(jié)。
??1.SpscLinkedArrayQueue的數(shù)據(jù)結(jié)構(gòu)為數(shù)組+鏈表，其中SpscLinkedArrayQueue不會遍歷數(shù)組，這個是SpscLinkedArrayQueue涉及的神奇之處。
??2.SpscLinkedArrayQueue擴容必須同時達到三個條件，一是index大于producerLookAhead，二是index + lookAheadStep位置上不為null，三是index + 1不為null，也就是說，當前0 ~ 2^n - 1范圍內(nèi)，只有index位置上為null。在這種情況下，才會擴容。
??3.擴容時，會將原數(shù)組的offset位置上設(shè)置為HAS_NEXT,同時將新數(shù)組的offset位置上設(shè)置為新添加的數(shù)據(jù)，然后就是，將新數(shù)組的指針設(shè)置在原數(shù)組的最后一位。
??4.poll時，當發(fā)現(xiàn)是HAS_NEXT，此時就去下一個數(shù)組相應(yīng)的offset位置上去找。