1、定义事件

事件(Event)就是通过 Disruptor 进行交换的数据类型。

package com.ljq.disruptor;import java.io.Serializable;/** * 定义事件数据，本质是个普通JavaBean *  * @author jqlin */@SuppressWarnings("serial")public class LongEvent implements Serializable {    private long value;    public LongEvent() {        super();    }    public LongEvent(long value) {        super();        this.value = value;    }    public long getValue() {        return value;    }    public void setValue(long value) {        this.value = value;    }    @Override    public String toString() {        return "LongEvent [value=" + value + "]";    }        }

 

2、LongEvent事件生产者

package com.ljq.disruptor;import com.lmax.disruptor.RingBuffer;/** * LongEvent事件生产者，生产LongEvent事件  *  * @author jqlin */public class LongEventProducer {    private final RingBuffer
     
       ringBuffer;    public LongEventProducer(RingBuffer
      
        ringBuffer) {        this.ringBuffer = ringBuffer;    }    public void produceData(long value) {        long sequence = ringBuffer.next(); // 获得下一个Event槽的下标        try {            // 给Event填充数据            LongEvent event = ringBuffer.get(sequence);            event.setValue(value);                    } finally {            // 发布Event，激活观察者去消费， 将sequence传递给该消费者            // 注意，最后的 ringBuffer.publish() 方法必须包含在 finally 中以确保必须得到调用；如果某个请求的 sequence 未被提交，将会堵塞后续的发布操作或者其它的 producer。            ringBuffer.publish(sequence);        }    }}
      
     

 

3、LongEvent事件消息者

package com.ljq.disruptor;import com.lmax.disruptor.WorkHandler;/** * LongEvent事件消息者，消息LongEvent事件 *  * @author Administrator * */public class LongEventConsumer  implements WorkHandler
     
       {        @Override    public void onEvent(LongEvent event) throws Exception {        System.out.println("consumer:" + Thread.currentThread().getName() + " Event: value=" + event.getValue() );    }}
     

 

4、ProducerConsumerMain 

消费者-生产者启动类，其依靠构造Disruptor对象，调用start()方法完成启动线程。

package com.ljq.disruptor;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;import com.lmax.disruptor.EventFactory;import com.lmax.disruptor.RingBuffer;import com.lmax.disruptor.SequenceBarrier;import com.lmax.disruptor.WorkerPool;import com.lmax.disruptor.YieldingWaitStrategy;import com.lmax.disruptor.dsl.ProducerType;/** *  Disruptor多个消费者不重复处理生产者发送过来的消息  *  * @author Administrator * */public class ProducerConsumerMain {    public static void main(String[] args) throws InterruptedException {        Long time = System.currentTimeMillis();                // 指定 ring buffer字节大小，必需为2的N次方(能将求模运算转为位运算提高效率 )，否则影响性能        int bufferSize = 1024 * 1024;;        //固定线程数        int nThreads = 10;                ExecutorService executor = Executors.newFixedThreadPool(nThreads);             EventFactory
     
       factory = new EventFactory
      
       () {            @Override            public LongEvent newInstance() {                return new LongEvent();            }        };                // 创建ringBuffer        RingBuffer
       
         ringBuffer = RingBuffer.create(ProducerType.MULTI, factory, bufferSize,  new YieldingWaitStrategy());        SequenceBarrier barriers = ringBuffer.newBarrier();        // 创建10个消费者来处理同一个生产者发送过来的消息(这10个消费者不重复消费消息)        LongEventConsumer[] consumers = new LongEventConsumer[50];        for (int i = 0; i < consumers.length; i++) {            consumers[i] = new LongEventConsumer();        }        WorkerPool
        
          workerPool = new WorkerPool
         
          (ringBuffer, barriers,                 new EventExceptionHandler(), consumers);        ringBuffer.addGatingSequences(workerPool.getWorkerSequences());        workerPool.start(executor);        LongEventProducer producer = new LongEventProducer(ringBuffer);        for (int i = 0; i < 20000; i++) {            producer.produceData(i);        }                Thread.sleep(1000); //等上1秒，等消费都处理完成        workerPool.halt(); //通知事件(或者说消息)处理器 可以结束了（并不是马上结束!!!）         executor.shutdown();         System.out.println("总共耗时(单位毫秒) :" + (System.currentTimeMillis() - time));    }}
         
        
       
      
     

 

5、EventExceptionHandler

package com.ljq.disruptor;import com.lmax.disruptor.ExceptionHandler;public class EventExceptionHandler implements ExceptionHandler {    @Override    public void handleEventException(Throwable ex, long sequence, Object event) {        System.out.println("handleEventException：" + ex);    }    @Override    public void handleOnShutdownException(Throwable ex) {        System.out.println("handleEventException：" + ex);    }    @Override    public void handleOnStartException(Throwable ex) {        System.out.println("handleOnStartException：" + ex);    }     }