@Data
public class EventData {
    private Long value;
}

public class EventConsumer implements WorkHandler<EventData> {
    /**
     * 消費(fèi)回調(diào)
     * @param eventData
     * @throws Exception
     */
    @Override
    public void onEvent(EventData eventData) throws Exception {
        Thread.sleep(5000);
        System.out.println(Thread.currentThread() + ", eventData:" + eventData.getValue());
    }
}

public class EventProducer {
    private final RingBuffer<EventData> ringBuffer;
    public EventProducer(RingBuffer<EventData> ringBuffer) {
        this.ringBuffer = ringBuffer;
    }
    public void sendData(Long v){
        // cas展位
        long next = ringBuffer.next();
        try {
            EventData eventData = ringBuffer.get(next);
            eventData.setValue(v);
        } finally {
            // 通知等待的消費(fèi)者
            System.out.println("EventProducer send success, sequence:"+next);
            ringBuffer.publish(next);
        }
    }
}

public class DisruptorTest {
    public static void main(String[] args) {
        // 2的n次方
        int bufferSize = 8;
        Disruptor<EventData> disruptor = new Disruptor<EventData>(
                () -> new EventData(), // 事件工廠
                bufferSize,            // 環(huán)形數(shù)組大小
                Executors.defaultThreadFactory(),       // 線程池工廠
                ProducerType.MULTI,    // 支持多事件發(fā)布者
                new BlockingWaitStrategy());    // 等待策略
        // 設(shè)置消費(fèi)者
        disruptor.handleEventsWithWorkerPool(
                new EventConsumer(),
                new EventConsumer(),
                new EventConsumer(),
                new EventConsumer());
        disruptor.start();
        RingBuffer<EventData> ringBuffer = disruptor.getRingBuffer();
        EventProducer eventProducer = new EventProducer(ringBuffer);
        long i  = 0;
        for(;;){
            i++;
            eventProducer.sendData(i);
            try {
                Thread.sleep(1500);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

// 消費(fèi)者上一次消費(fèi)的最小序號(hào) // 后續(xù)第二點(diǎn)會(huì)講到
private final Sequence gatingSequenceCache = new Sequence(Sequencer.INITIAL_CURSOR_VALUE);
// 當(dāng)前進(jìn)度的序號(hào)
protected final Sequence cursor = new Sequence(Sequencer.INITIAL_CURSOR_VALUE);
// 所有消費(fèi)者的序號(hào) //后續(xù)第二點(diǎn)會(huì)講到
protected volatile Sequence[] gatingSequences = new Sequence[0];
 public long next(int n)
    {
        if (n < 1)
        {
            throw new IllegalArgumentException("n must be > 0");
        }
        long current;
        long next;
        do
        {
            // 當(dāng)前進(jìn)度的序號(hào)，Sequence的value具有可見性，保證多線程間線程之間能感知到可申請(qǐng)的最新值
            current = cursor.get();
            // 要申請(qǐng)的序號(hào)空間:最大序列號(hào)
            next = current + n;
            long wrapPoint = next - bufferSize;
            // 消費(fèi)者最小序列號(hào)
            long cachedGatingSequence = gatingSequenceCache.get();
            // 大于一圈 || 最小消費(fèi)序列號(hào)>當(dāng)前進(jìn)度
            if (wrapPoint > cachedGatingSequence || cachedGatingSequence > current)
            {
                long gatingSequence = Util.getMinimumSequence(gatingSequences, current);
                // 說明大于1圈，并沒有多余空間可以申請(qǐng)
                if (wrapPoint > gatingSequence)
                {
                    LockSupport.parkNanos(1); // TODO, should we spin based on the wait strategy?
                    continue;
                }
                // 更新最小值到Sequence的value中
                gatingSequenceCache.set(gatingSequence);
            }
            // CAS成功后更新當(dāng)前Sequence的value
            else if (cursor.compareAndSet(current, next))
            {
                break;
            }
        }
        while (true);
        return next;
    }

public final EventHandlerGroup<T> handleEventsWithWorkerPool(final WorkHandler<T>... workHandlers)
{
    return createWorkerPool(new Sequence[0], workHandlers);
}
EventHandlerGroup<T> createWorkerPool(
    final Sequence[] barrierSequences, final WorkHandler<? super T>[] workHandlers)
{
    final SequenceBarrier sequenceBarrier = ringBuffer.newBarrier(barrierSequences);
    final WorkerPool<T> workerPool = new WorkerPool<>(ringBuffer, sequenceBarrier, exceptionHandler, workHandlers);
    consumerRepository.add(workerPool, sequenceBarrier);
    final Sequence[] workerSequences = workerPool.getWorkerSequences();
    updateGatingSequencesForNextInChain(barrierSequences, workerSequences);
    return new EventHandlerGroup<>(this, consumerRepository, workerSequences);
}
    private void updateGatingSequencesForNextInChain(final Sequence[] barrierSequences, final Sequence[] processorSequences)
{
    if (processorSequences.length > 0)
    {
        // 消費(fèi)者啟動(dòng)后就會(huì)將所有消費(fèi)者存放入AbstractSequencer中g(shù)atingSequences
        ringBuffer.addGatingSequences(processorSequences);
        for (final Sequence barrierSequence : barrierSequences)
        {
            ringBuffer.removeGatingSequence(barrierSequence);
        }
        consumerRepository.unMarkEventProcessorsAsEndOfChain(barrierSequences);
    }
}

public MultiProducerSequencer(int bufferSize, final WaitStrategy waitStrategy)
{
    super(bufferSize, waitStrategy);
    // 初始化可用數(shù)組
    availableBuffer = new int[bufferSize];
    indexMask = bufferSize - 1;
    indexShift = Util.log2(bufferSize);
    initialiseAvailableBuffer();
}
// 初始化默認(rèn)availableBuffer為-1
private void initialiseAvailableBuffer()
{
    for (int i = availableBuffer.length - 1; i != 0; i--)
    {
        setAvailableBufferValue(i, -1);
    }
    setAvailableBufferValue(0, -1);
}
// 生產(chǎn)者成功生產(chǎn)事件將可用區(qū)數(shù)組置為1
public void publish(final long sequence)
{
    setAvailable(sequence);
    waitStrategy.signalAllWhenBlocking();
}
private void setAvailableBufferValue(int index, int flag)
{
    long bufferAddress = (index * SCALE) + BASE;
    UNSAFE.putOrderedInt(availableBuffer, bufferAddress, flag);
}

WorkProcessor類中消費(fèi)run方法
public void run()
    {
        boolean processedSequence = true;
        long cachedAvailableSequence = Long.MIN_VALUE;
        long nextSequence = sequence.get();
        T event = null;
        while (true)
        {
            try
            {
                // 先通過cas獲取消費(fèi)事件的占有權(quán)
                if (processedSequence)
                {
                    processedSequence = false;
                    do
                    {
                        nextSequence = workSequence.get() + 1L;
                        sequence.set(nextSequence - 1L);
                    }
                    while (!workSequence.compareAndSet(nextSequence - 1L, nextSequence));
                }
                // 數(shù)據(jù)就緒，可以消費(fèi)
                if (cachedAvailableSequence >= nextSequence)
                {
                    event = ringBuffer.get(nextSequence);
                    // 觸發(fā)回調(diào)函數(shù)
                    workHandler.onEvent(event);
                    processedSequence = true;
                }
                else
                {
                    // 獲取可以被讀取的下標(biāo)
                    cachedAvailableSequence = sequenceBarrier.waitFor(nextSequence);
                }
            }
        // ....省略
        }
        notifyShutdown();
        running.set(false);
    }
    public long waitFor(final long sequence)
        throws AlertException, InterruptedException, TimeoutException
    {
        checkAlert();
        // 這個(gè)值獲取的current write 下標(biāo)，可以認(rèn)為全局消費(fèi)下標(biāo)。此處與每一段的write1和write2下標(biāo)區(qū)分開
        long availableSequence = waitStrategy.waitFor(sequence, cursorSequence, dependentSequence, this);
        if (availableSequence < sequence)
        {
            return availableSequence;
        }
        // 通過availableBuffer篩選出第一個(gè)不可用序號(hào) -1
        return sequencer.getHighestPublishedSequence(sequence, availableSequence);
    }
    public long getHighestPublishedSequence(long lowerBound, long availableSequence)
    {
        // 從current read下標(biāo)開始， 循環(huán)至 current write,如果碰到availableBuffer 為-1 直接返回
        for (long sequence = lowerBound; sequence <= availableSequence; sequence++)
        {
            if (!isAvailable(sequence))
            {
                return sequence - 1;
            }
        }
        return availableSequence;
    }