Disruptor是一个用于生产者-消费者模型的高并发框架(CAS)，由于其实现原理(RingBuffer环形队列、CAS),性能极高

    下面是一个由生产者放入整数，消费者得到后进行平方运算的例子

    code

    1. package concurrent;
    2. import com.lmax.disruptor.*;
    3. import com.lmax.disruptor.dsl.Disruptor;
    4. import com.lmax.disruptor.dsl.ProducerType;
    5. import java.nio.ByteBuffer;
    6. import java.util.concurrent.Executor;
    7. import java.util.concurrent.ExecutorService;
    8. import java.util.concurrent.Executors;
    10. public class C11_Disruptor {
    11.     //用于存放数据的类
    12.     static class LongEvent {
    13.         private long data;
    14.         public void setData(long data){ this.data = data; }
    15.         public long getData(){ return data; }
    16.     }
    18.     /*
    19.      工厂类，它会在Disruptor系统初始化时，构造所有的缓冲区中的对象实例（Disruptor会预先分配系统）
    20.     */
    21.     static class LongEventFactory implements EventFactory<LongEvent>{
    22.         @Override
    23.         public LongEvent newInstance() {
    24.             return new LongEvent();
    25.         }
    26.     }
    28.     /*
    29.      消费者类，需要实现WorkHandler接口，我们只需要在onEvent中进行数据处理即可，线程同步等事情Disruptor会做
    30.     */
    31.     static class Consumer implements WorkHandler<LongEvent>{
    32.         @Override
    33.         public void onEvent(LongEvent event) throws Exception {
    34.             System.out.println(Thread.currentThread().getId() + ":" + event.getData() * event.getData());
    35.         }
    36.     }
    39.     /*
    40.      生产者
    41.     */
    42.     static class Producer{
    43.         private final RingBuffer<LongEvent> ringBuffer; //环形队列
    44.         public Producer(RingBuffer<LongEvent> ringBuffer){
    45.             this.ringBuffer = ringBuffer;
    46.         }
    48.         public void pushData(ByteBuffer bb){
    49.             long sequence = ringBuffer.next();  //获取下一个可用序列号
    50.             try{
    51.                 //通过序列号获取在环形队列中的入口(空的LongEvent对象),然后设置其数据
    52.                 LongEvent event = ringBuffer.get(sequence);
    53.                 event.setData(bb.getLong(0));
    54.             }finally {
    55.                 //数据发布，只有发布后才能被消费者发现
    56.                 ringBuffer.publish(sequence);
    57.             }
    58.         }
    59.     }
    61.     public static void main(String[] args) throws InterruptedException {
    62.         ExecutorService executor = Executors.newCachedThreadPool();
    63.         LongEventFactory factory = new LongEventFactory();
    64.         //RingBuffer环形队列的大小必须是2的整数幂
    65.         int bufferSize = 1024;
    66.         Disruptor<LongEvent> disruptor = new Disruptor<LongEvent>(
    67.                 factory,    //定义事件工厂 -- 返回事件实例
    68.                 bufferSize, //环形队列的大小 -- 2的整数幂
    69.                 executor,   //用于事件处理的线程池
    70.                 ProducerType.SINGLE,    //生产者是单例还是多例
    71.                 new YieldingWaitStrategy()  //指定等待策略
    72.                 /*  等待策略
    73.                  *   BlockingWaitStrategy:默认策略，使用了锁和条件，最节省CPU，但性能最差。
    74.                  *   SleepingWaitStrategy:节省CPU，但是对于数据处理可能产生比较高的平均延时，对生产者线程影响最小
    75.                  *   YieldingWaitStrategy:数据处理延时低，，但需要逻辑CPU数量>消费者线程数(这里的逻辑CPU，指的是“双核四线程”中的四线程)
    76.                  *   BusySpinWaitStrategy:数据处理延时苛刻，消费者线程采取轮询方式监视缓冲区的的变化，这意味着会消耗所有的CPU资源。
    77.                  *                       你的物理CPU数量>消费者线程数
    78.                  * */
    79.         );
    80.         //设置处理数据的消费者
    81.         disruptor.handleEventsWithWorkerPool(
    82.                 new Consumer(),
    83.                 new Consumer(),
    84.                 new Consumer(),
    85.                 new Consumer()
    86.         );
    87.         //初始化Disruptor系统
    88.         disruptor.start();
    89.         //模拟一个生产者不断向RingBuffer缓冲区添加数据
    90.         RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();
    91.         Producer producer = new Producer(ringBuffer);
    92.         //分配一个8字节缓冲区 ---- long为8字节
    93.         ByteBuffer bb = ByteBuffer.allocate(8);
    94.         for(long l=0 ; l<9 ; l++){
    95.             bb.putLong(0, l);
    96.             producer.pushData(bb);
    97.             Thread.sleep(100);
    98.             System.out.println("add data " + l);
    99.         }
    100.         //关闭资源
    101.         disruptor.shutdown();
    102.         executor.shutdown();
    103.     }
    105. }

    console

    1. add data 0
    2. 11:0
    3. add data 1
    4. 14:1
    5. add data 2
    6. 13:4
    7. add data 3
    8. 12:9
    9. add data 4
    10. 11:16
    11. add data 5
    12. 14:25
    13. add data 6
    14. 13:36
    15. add data 7
    16. 12:49
    17. add data 8
    18. 11:64