public void put(E e) throws InterruptedException {
        checkNotNull(e);
        final ReentrantLock lock = this.lock;
        //获取当前对象锁
        lock.lockInterruptibly();
        try {
            while (count == items.length)
                //阻塞并释放锁，等待notFull.signal()通知
                notFull.await();
            //将数据放入数组
            enqueue(e);
        } finally {
            lock.unlock();
        }
    }
    private void enqueue(E x) {
        final Object[] items = this.items;
        //putIndex 就是入队的下标
        items[putIndex] = x;
        if (++putIndex == items.length)
            putIndex = 0;
        count++;
        notEmpty.signal();
    }
    public E take() throws InterruptedException {
        final ReentrantLock lock = this.lock;
        //加锁
        lock.lockInterruptibly();
        try {
            while (count == 0)
                //阻塞并释放对象锁，并等待notEmpty.signal()通知
                notEmpty.await();
            //在数据不为空的情况下
            return dequeue();
        } finally {
            lock.unlock();
        }
    }
    private E dequeue() {
        final Object[] items = this.items;
        //takeIndex 是出队的下标
        E x = (E) items[takeIndex];
        items[takeIndex] = null;
        if (++takeIndex == items.length)
            takeIndex = 0;
        count--;
        if (itrs != null)
            itrs.elementDequeued();
        notFull.signal();
        return x;
    }

class LhsPadding{
    //缓存行补齐， 提升cache缓存命中率
    protected long p1, p2, p3, p4, p5, p6, p7;
}
class Value extends LhsPadding{
    protected volatile long value;
}
class RhsPadding extends Value{
    //缓存行补齐， 提升cache缓存命中率
    protected long p9, p10, p11, p12, p13, p14, p15;
}
public class Sequence extends RhsPadding{
    ...
}

public Disruptor(
        final EventFactory<T> eventFactory,    // 数据实体构造工厂
        final int ringBufferSize,              // 队列大小，必须是2的次方
        final ThreadFactory threadFactory,     // 线程工厂
        final ProducerType producerType,       // 生产者类型，单个生产者还是多个
        final WaitStrategy waitStrategy){      // 消费者等待策略
    ...
}
public Disruptor(
    final EventFactory<T> eventFactory,     
    final int ringBufferSize, 
    final ThreadFactory threadFactory){
    ...
}

public void send(String data){
    RingBuffer<MsgEvent> ringBuffer = this.disruptor.getRingBuffer();
    //获取下一个放置数据的位置
    long next = ringBuffer.next();
    try{
        MsgEvent event = ringBuffer.get(next);
        event.setValue(data);
    }finally {
        //发布事件
        ringBuffer.publish(next);
    }
}

public EventHandlerGroup<T> handleEventsWith(final EventHandler<? super T>... handlers){
    ...
}
public EventHandlerGroup<T> handleEventsWith(final EventProcessor... processors){
    ...
}
public EventHandlerGroup<T> handleEventsWith(final EventProcessorFactory<T>... eventProcessorFactories){
    ...
}
public EventHandlerGroup<T> handleEventsWithWorkerPool(final WorkHandler<T>... workHandlers){
    ...
}

//消费者
public class MsgConsumer implements EventHandler<MsgEvent>{
    private String name;
    public MsgConsumer(String name){
        this.name = name;
    }
    @Override
    public void onEvent(MsgEvent msgEvent, long l, boolean b) throws Exception {
        System.out.println(this.name+" -> 接收到信息： "+msgEvent.getValue());
    }
}
//生产者处理
public class MsgProducer {
    private Disruptor disruptor;
    public MsgProducer(Disruptor disruptor){
        this.disruptor = disruptor;
    }
    public void send(String data){
        RingBuffer<MsgEvent> ringBuffer = this.disruptor.getRingBuffer();
        long next = ringBuffer.next();
        try{
            MsgEvent event = ringBuffer.get(next);
            event.setValue(data);
        }finally {
            ringBuffer.publish(next);
        }
    }
    public void send(List<String> dataList){
        dataList.stream().forEach(data -> this.send(data));
    }
}
//触发测试
public class DisruptorDemo {
    @Test
    public void test(){
        Disruptor<MsgEvent> disruptor = new Disruptor<>(MsgEvent::new, 1024, Executors.defaultThreadFactory());
        //定义消费者
        MsgConsumer msg1 = new MsgConsumer("1");
        MsgConsumer msg2 = new MsgConsumer("2");
        MsgConsumer msg3 = new MsgConsumer("3");
        //绑定配置关系
        disruptor.handleEventsWith(msg1, msg2, msg3);
        disruptor.start();
        // 定义要发送的数据
        MsgProducer msgProducer = new MsgProducer(disruptor);
        msgProducer.send(Arrays.asList("nihao","hah"));
    }
}

/**
 * 分组处理 handleEventWithWorkerPool
 */
@Test
public void test1(){
    Disruptor<MsgEvent> disruptor = new Disruptor(MsgEvent::new, 1024, Executors.defaultThreadFactory());
    disruptor.handleEventsWithWorkerPool(new MyWorkHandler("work1"), new MyWorkHandler("work2"));
    disruptor.start();
    MsgProducer msgProducer = new MsgProducer(disruptor);
    msgProducer.send(Arrays.asList("aaa","bbb"));
}

/**
 * 测试顺序消费
 * 每一条消息的消费者1和3消费完毕后，消费者2再进行消费
 */
@Test
public void test2(){
    MsgConsumer msg1 = new MsgConsumer("1");
    MsgConsumer msg2 = new MsgConsumer("2");
    MsgConsumer msg3 = new MsgConsumer("3");
    Disruptor<MsgEvent> disruptor = new Disruptor(MsgEvent::new, 1024, Executors.defaultThreadFactory());
    disruptor.handleEventsWith(msg1, msg3).then(msg2);
    disruptor.start();
    MsgProducer msgProducer = new MsgProducer(disruptor);
    msgProducer.send(Arrays.asList("aaa", "bbb", "ccc", "ddd"));
}

/**
 * 测试多支线消费
 * 消费者1和消费者3一个支线，消费者2和消费者4一个支线，消费者3和消费者4消费完毕后，消费者5再进行消费
 */
@Test
public void test3(){
    MsgConsumer msg1 = new MsgConsumer("1");
    MsgConsumer msg2 = new MsgConsumer("2");
    MsgConsumer msg3 = new MsgConsumer("3");
    MsgConsumer msg4 = new MsgConsumer("4");
    MsgConsumer msg5 = new MsgConsumer("5");
    //支线：消费者1和消费者3
    disruptor.handleEventsWith(msg1, msg3);
    //支线：消费者2和消费者4
    disruptor.handleEventsWith(msg2, msg4);
    //消费者3和消费者4执行完之后，指向消费者5
    disruptor.after(msg3, msg4).handleEventsWith(msg5);
    disruptor.start();
    MsgProducer msgProducer = new MsgProducer(disruptor);
    msgProducer.send(Arrays.asList("aaa", "bbb", "ccc", "ddd"));
}