SequenceBarrier

SequenceBarrier是消費者與Ringbuffer之間建立消費關係的橋樑，同時也是消費者與消費者之間消費依賴的抽象。

SequenceBarrier只有一個實現類，就是ProcessingSequenceBarrier。ProcessingSequenceBarrier由生產者Sequencer，消費定位cursorSequence，等待策略waitStrategy還有一組依賴sequence：dependentSequence組成：

public ProcessingSequenceBarrier(
        final Sequencer sequencer,
        final
 
 WaitStrategy waitStrategy,
        final Sequence cursorSequence,
        final Sequence[] dependentSequences)
{
   this.sequencer = sequencer;
   this.waitStrategy = waitStrategy;
   this.cursorSequence = cursorSequence;
   if (0 == dependentSequences.length)
   {
       dependentSequence = cursorSequence;
   }
   else
 

   {
       dependentSequence = new FixedSequenceGroup(dependentSequences);
   }
}

首先，為了實現消費依賴，SequenceBarrier肯定有一個獲取可以消費的sequence方法，就是

long waitFor(long sequence) throws AlertException, InterruptedException, TimeoutException;

實現為：

@Override
    public long waitFor(final long sequence)
        throws
 
 AlertException, InterruptedException, TimeoutException
    {
        //檢查是否alerted
        checkAlert();
        //通過等待策略獲取下一個可消費的sequence，這個sequence通過之前的講解可以知道，需要大於cursorSequence和dependentSequence，我們可以通過dependentSequence實現先後消費
        long availableSequence = waitStrategy.waitFor(sequence, cursorSequence, dependentSequence, this);
        //等待可能被中斷，所以檢查下availableSequence是否小於sequence
        if (availableSequence < sequence)
        {
            return availableSequence;
        }
        //如果不小於，返回所有sequence（可能多生產者）和availableSequence中最大的
        return sequencer.getHighestPublishedSequence(sequence, availableSequence);
    }

其他方法實現很簡單，功能上分別有：
1. 獲取當前cursorSequence（並沒有什麼用，就是為了監控）
2. 負責中斷和恢復的alert標記

    @Override
    public long getCursor()
    {
        return dependentSequence.get();
    }

    @Override
    public boolean isAlerted()
    {
        return alerted;
    }

    @Override
    public void alert()
    {
        alerted = true;
        waitStrategy.signalAllWhenBlocking();
    }

    @Override
    public void clearAlert()
    {
        alerted = false;
    }

    @Override
    public void checkAlert() throws AlertException
    {
        if (alerted)
        {
            throw AlertException.INSTANCE;
        }
    }

構造SequenceBarrier在框架中只有一個入口，就是AbstractSequencer的：

public SequenceBarrier newBarrier(Sequence... sequencesToTrack)
    {
        return new ProcessingSequenceBarrier(this, waitStrategy, cursor, sequencesToTrack);
    }

SequenceProcessor

通過SequenceBarrier，我們可以實現消費之間的依賴關係，但是，消費方式（比如廣播，群組消費等等），需要通過SequenceProcessor的實現類實現：

通過類依賴關係我們發現，EventProcessor都是拓展了Runnable介面，也就是我們可以把它們當做執行緒處理。

1. BatchEventProcessor：

它的構造方法：

/**
     * 構造一個消費者之間非互斥消費的消費者
     *
     * @param dataProvider    對應的RingBuffer
     * @param sequenceBarrier 依賴關係，通過構造不同的sequenceBarrier用互相的dependentsequence，我們可以構造出先後消費關係
     * @param eventHandler    使用者實現的處理消費的event的業務消費者.
     */
    public BatchEventProcessor(
        final DataProvider<T> dataProvider,
        final SequenceBarrier sequenceBarrier,
        final EventHandler<? super T> eventHandler)
    {
        this.dataProvider = dataProvider;
        this.sequenceBarrier = sequenceBarrier;
        this.eventHandler = eventHandler;

        if (eventHandler instanceof SequenceReportingEventHandler)
        {
            ((SequenceReportingEventHandler<?>) eventHandler).setSequenceCallback(sequence);
        }

        timeoutHandler = (eventHandler instanceof TimeoutHandler) ? (TimeoutHandler) eventHandler : null;
    }

執行緒為一個死迴圈：

 @Override
    public void run()
    {
        //檢查狀態
        if (!running.compareAndSet(false, true))
        {
            throw new IllegalStateException("Thread is already running");
        }
        //清理
        sequenceBarrier.clearAlert();
        //如果使用者實現的EventHandler繼承了LifecycleAware，則執行其onStart方法
        notifyStart();

        T event = null;
        //sequence初始值為-1，設計上當前值是已經消費過的
        long nextSequence = sequence.get() + 1L;
        try
        {
            while (true)
            {
                try
                {
                    //獲取當前可以消費的最大sequence
                    final long availableSequence = sequenceBarrier.waitFor(nextSequence);
                    while (nextSequence <= availableSequence)
                    {
                        //獲取並處理
                        event = dataProvider.get(nextSequence);
                        eventHandler.onEvent(event, nextSequence, nextSequence == availableSequence);
                        nextSequence++;
                    }
                    //設定當前sequence，注意，出現異常需要特殊處理，防止重複消費
                    sequence.set(availableSequence);
                }
                catch (final TimeoutException e)
                {
                    //wait超時異常
                    notifyTimeout(sequence.get());
                }
                catch (final AlertException ex)
                {
                    //中斷異常
                    if (!running.get())
                    {
                        break;
                    }
                }
                catch (final Throwable ex)
                {
                    exceptionHandler.handleEventException(ex, nextSequence, event);
                    //如果出現異常則設定為nextSequence
                    sequence.set(nextSequence);
                    nextSequence++;
                }
            }
        }
        finally
        {
            //如果使用者實現的EventHandler繼承了LifecycleAware，則執行其onShutdown方法
            notifyShutdown();
            running.set(false);
        }
    }

可以看出：
1. BatchEventProcessor可以處理超時，可以處理中斷，可以通過使用者實現的異常處理類處理異常，同時，發生異常之後再次啟動，不會漏消費，也不會重複消費。
2. 不同的BatchEventProcessor之間通過SequenceBarrier進行依賴消費。原理如下圖所示：

假設我們有三個消費者BatchEventProcessor1，BatchEventProcessor2，BatchEventProcessor3. 1需要先於2和3消費，那麼構建BatchEventProcessor和SequenceBarrier時，我們需要讓BatchEventProcessor2和BatchEventProcessor3的SequenceBarrier的dependentSequence中加入SequenceBarrier1的sequence。
其實這裡2和3共用一個SequenceBarrier就行。

2. WorkProcessor

另一種消費者是WorkProcessor。利用它，可以實現互斥消費，同樣的利用SequenceBarrier可以實現消費順序

public void run()
    {
        if (!running.compareAndSet(false, true))
        {
            throw new IllegalStateException("Thread is already running");
        }
        sequenceBarrier.clearAlert();

        notifyStart();

        boolean processedSequence = true;
        long cachedAvailableSequence = Long.MIN_VALUE;
        long nextSequence = sequence.get();
        T event = null;
        while (true)
        {
            try
            {
                if (processedSequence)
                {
                    processedSequence = false;
                    //獲取下一個可以消費的Sequence
                    do
                    {
                        nextSequence = workSequence.get() + 1L;
                        sequence.set(nextSequence - 1L);
                    }
                    //多個WorkProcessor之間，如果共享一個workSequence，那麼，可以實現互斥消費，因為只有一個執行緒可以CAS更新成功
                    while (!workSequence.compareAndSet(nextSequence - 1L, nextSequence));
                }

                if (cachedAvailableSequence >= nextSequence)
                {
                    event = ringBuffer.get(nextSequence);
                    workHandler.onEvent(event);
                    processedSequence = true;
                }
                else
                {
                    cachedAvailableSequence = sequenceBarrier.waitFor(nextSequence);
                }
            }
            catch (final TimeoutException e)
            {
                notifyTimeout(sequence.get());
            }
            catch (final AlertException ex)
            {
                if (!running.get())
                {
                    break;
                }
            }
            catch (final Throwable ex)
            {
                // handle, mark as processed, unless the exception handler threw an exception
                exceptionHandler.handleEventException(ex, nextSequence, event);
                processedSequence = true;
            }
        }

        notifyShutdown();

        running.set(false);
    }

3. WorkerPool

多個WorkerProcessor可以組成一個WorkerPool：

public WorkerPool(
        final RingBuffer<T> ringBuffer,
        final SequenceBarrier sequenceBarrier,
        final ExceptionHandler<? super T> exceptionHandler,
        final WorkHandler<? super T>... workHandlers)
    {
        this.ringBuffer = ringBuffer;
        final int numWorkers = workHandlers.length;
        workProcessors = new WorkProcessor[numWorkers];

        for (int i = 0; i < numWorkers; i++)
        {
            workProcessors[i] = new WorkProcessor<T>(
                ringBuffer,
                sequenceBarrier,
                workHandlers[i],
                exceptionHandler,
                workSequence);
        }
    }

裡面的 workHandlers[i]共享同一個workSequence，所以，同一個WorkerPool內，是互斥消費。