private final RingBuffer<T> ringBuffer;
/**消费者事件处理线程池**/
private final Executor executor;
/**消费者集合**/
private final ConsumerRepository<T> consumerRepository = new ConsumerRepository<T>();
/**Disruptor是否启动标示，只能启动一次**/
private final AtomicBoolean started = new AtomicBoolean(false);
/**消费者事件异常处理方法**/
private ExceptionHandler<? super T> exceptionHandler = new ExceptionHandlerWrapper<T>();

EventHandlerGroup<T> createEventProcessors(final Sequence[] barrierSequences,?                                           final EventHandler<T>[] eventHandlers)?{
    checkNotStarted();
    //每个消费者有自己的事件序号Sequence
    final Sequence[] processorSequences = new Sequence[eventHandlers.length];   
    //消费者通过SequenceBarrier等待可消费事件
    final SequenceBarrier barrier = ringBuffer.newBarrier(barrierSequences);    for (int i = 0, eventHandlersLength = eventHandlers.length; i < eventHandlersLength; i++)
    {
        final EventHandler<T> eventHandler = eventHandlers[i];
        //每个消费者都以BatchEventProcessor被调度
        final BatchEventProcessor<T> batchEventProcessor = new BatchEventProcessor<T>(ringBuffer, barrier, eventHandler);  
        if (exceptionHandler != null)
        {
            batchEventProcessor.setExceptionHandler(exceptionHandler);
        }
        consumerRepository.add(batchEventProcessor, eventHandler, barrier);
        processorSequences[i] = batchEventProcessor.getSequence();
    }
    
    if (processorSequences.length > 0)
    {
        consumerRepository.unMarkEventProcessorsAsEndOfChain(barrierSequences);
    }
    
    return new EventHandlerGroup<T>(this, consumerRepository, processorSequences);
}

@Override
public void run()
{
    if (!running.compareAndSet(false, true))
    {
        throw new IllegalStateException("Thread is already running");
    }
    sequenceBarrier.clearAlert();
    
    notifyStart();
    
    T event = null;
    // sequence.get()标示当前已经处理的序号
    long nextSequence = sequence.get() + 1L;
    try
    {
        while (true)
        {
            try
            {
                // sequenceBarrier最重要的作用，就是让消费者等待下一个可用的序号
                // 可用序号可能会大于nextSequence，从而消费者可以一次处理多个事件
                // 如果该消费者同时也依赖了其他消费者，则会返回最小的那个
                final long availableSequence = sequenceBarrier.waitFor(nextSequence);
                if (nextSequence > availableSequence)
                {
                    Thread.yield();
                }
                
                while (nextSequence <= availableSequence)
                {
                    event = dataProvider.get(nextSequence);
                    // eventHandler是用户定义的事件消费逻辑
                    eventHandler.onEvent(event, nextSequence, nextSequence == availableSequence);
                    nextSequence++;
                }
                
                // 跟踪自己处理的事件
                sequence.set(availableSequence);
            }
            catch (final TimeoutException e)
            {
                notifyTimeout(sequence.get());
            }
            catch (final AlertException ex)
            {
                if (!running.get())
                {
                    break;
                }
            }
            catch (final Throwable ex)
            {
                exceptionHandler.handleEventException(ex, nextSequence, event);
                sequence.set(nextSequence);
                nextSequence++;
            }
        }
    }
    finally
    {
        notifyShutdown();
        running.set(false);
    }
}

public ProcessingSequenceBarrier(final Sequencer sequencer,final WaitStrategy waitStrategy,final Sequence cursorSequence,final Sequence[] dependentSequences)
{
    // 生产者的ringBuffer控制器sequencer
    this.sequencer = sequencer;
    // 消费者等待可消费事件的策略
    this.waitStrategy = waitStrategy;
    // ringBuffer的cursor
    this.cursorSequence = cursorSequence;
    if (0 == dependentSequences.length)
    {
        dependentSequence = cursorSequence;
    }
    else
    {
    // 当依赖其他消费者时，dependentSequence就是其他消费者的序号
        dependentSequence = new FixedSequenceGroup(dependentSequences);
    }
}

@Override
public long waitFor(long sequence, Sequence cursorSequence, Sequence dependentSequence, SequenceBarrier barrier)
    throws AlertException, InterruptedException
{
    long availableSequence;
    if ((availableSequence = cursorSequence.get()) < sequence)
    {
        lock.lock();
        try
        {
            // 如果ringBuffer的cursor小于需要的序号，也就是生产者没有新的事件发出，则阻塞消费者线程，直到生产者通过Sequencer的publish方法唤醒消费者。
            while ((availableSequence = cursorSequence.get()) < sequence)
            {
                barrier.checkAlert();
                processorNotifyCondition.await();
            }
        }
        finally
        {
            lock.unlock();
        }
    }
    
    // 如果生产者新发布了事件，但是依赖的其他消费者还没处理完，则等待所依赖的消费者先处理。在本文的例子中，就是等B与C先处理完，D才能处理事件。
    while ((availableSequence = dependentSequence.get()) < sequence)
    {
        barrier.checkAlert();
    }
    
    return availableSequence;
}

@Override
public boolean hasAvailableCapacity(final int requiredCapacity)
{
    long nextValue = http://www.mamicode.com/pad.nextValue;>