import java.nio.ByteBuffer;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

import com.lmax.disruptor.RingBuffer;

import com.lmax.disruptor.YieldingWaitStrategy;

import com.lmax.disruptor.dsl.Disruptor;

import com.lmax.disruptor.dsl.ProducerType;

public class LongEventMain {

    public static void main(String[] args) throws Exception {

        //创建缓冲池

        ExecutorService  executor = Executors.newCachedThreadPool();

        //创建工厂

        LongEventFactory factory = new LongEventFactory();

        //创建bufferSize ,也就是RingBuffer大小，必须是2的N次方

        int ringBufferSize = 1024 * 1024; // 

        /**

        //BlockingWaitStrategy 是最低效的策略，但其对CPU的消耗最小并且在各种不同部署环境中能提供更加一致的性能表现

        WaitStrategy BLOCKING_WAIT = new BlockingWaitStrategy();

        //SleepingWaitStrategy 的性能表现跟BlockingWaitStrategy差不多，对CPU的消耗也类似，但其对生产者线程的影响最小，适合用于异步日志类似的场景

        WaitStrategy SLEEPING_WAIT = new SleepingWaitStrategy();

        //YieldingWaitStrategy 的性能是最好的，适合用于低延迟的系统。在要求极高性能且事件处理线数小于CPU逻辑核心数的场景中，推荐使用此策略；例如，CPU开启超线程的特性

        WaitStrategy YIELDING_WAIT = new YieldingWaitStrategy();

        */

        //创建disruptor

        Disruptor<LongEvent> disruptor =

                new Disruptor<LongEvent>(factory, ringBufferSize, executor, ProducerType.SINGLE, new YieldingWaitStrategy());

        // 连接消费事件方法

        disruptor.handleEventsWith(new LongEventHandler());

        // 启动

        disruptor.start();

        //Disruptor 的事件发布过程是一个两阶段提交的过程：

        //发布事件

        RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();

        LongEventProducer producer = new LongEventProducer(ringBuffer);

        //LongEventProducerWithTranslator producer = new LongEventProducerWithTranslator(ringBuffer);

        ByteBuffer byteBuffer = ByteBuffer.allocate(8);

        for(long l = 0; l<100; l++){

            byteBuffer.putLong(0, l);

            producer.onData(byteBuffer);

            //Thread.sleep(1000);

        }

        disruptor.shutdown();//关闭 disruptor，方法会堵塞，直至所有的事件都得到处理；

        executor.shutdown();//关闭 disruptor 使用的线程池；如果需要的话，必须手动关闭， disruptor 在 shutdown 时不会自动关闭；        

    }

}

public class LongEvent {

    private long value;

    public long getValue() {

        return value;

    } 

    public void setValue(long value) {

        this.value = value;

    }

} 

import com.lmax.disruptor.EventFactory;

// 需要让disruptor为我们创建事件，我们同时还声明了一个EventFactory来实例化Event对象。

public class LongEventFactory implements EventFactory { 

    @Override

    public Object newInstance() {

        return new LongEvent();

    }

} 

import java.nio.ByteBuffer;

import com.lmax.disruptor.RingBuffer;

/**

 * 很明显的是：当用一个简单队列来发布事件的时候会牵涉更多的细节，这是因为事件对象还需要预先创建。

 * 发布事件最少需要两步：获取下一个事件槽并发布事件（发布事件的时候要使用try/finnally保证事件一定会被发布）。

 * 如果我们使用RingBuffer.next()获取一个事件槽，那么一定要发布对应的事件。

 * 如果不能发布事件，那么就会引起Disruptor状态的混乱。

 * 尤其是在多个事件生产者的情况下会导致事件消费者失速，从而不得不重启应用才能会恢复。

 */

public class LongEventProducer {

    private final RingBuffer<LongEvent> ringBuffer;

    public LongEventProducer(RingBuffer<LongEvent> ringBuffer){

        this.ringBuffer = ringBuffer;

    }

    /**

     * onData用来发布事件，每调用一次就发布一次事件

     * 它的参数会用过事件传递给消费者

     */

    public void onData(ByteBuffer bb){

        //1.可以把ringBuffer看做一个事件队列，那么next就是得到下面一个事件槽

        long sequence = ringBuffer.next();

        try {

            //2.用上面的索引取出一个空的事件用于填充（获取该序号对应的事件对象）

            LongEvent event = ringBuffer.get(sequence);

            //3.获取要通过事件传递的业务数据

            event.setValue(bb.getLong(0));

        } finally {

            //4.发布事件

            //注意，最后的 ringBuffer.publish 方法必须包含在 finally 中以确保必须得到调用；如果某个请求的 sequence 未被提交，将会堵塞后续的发布操作或者其它的 producer。

            ringBuffer.publish(sequence);

        }

    }

}

import com.lmax.disruptor.EventHandler;

//我们还需要一个事件消费者，也就是一个事件处理器。这个事件处理器简单地把事件中存储的数据打印到终端：

public class LongEventHandler implements EventHandler<LongEvent>  {

    @Override

    public void onEvent(LongEvent longEvent, long l, boolean b) throws Exception {

        System.out.println(longEvent.getValue());

    }

}