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 + "]";

    }

}

package com.ljq.disruptor;

import com.lmax.disruptor.RingBuffer;

/**

 * LongEvent事件生产者，生产LongEvent事件

 *

 * @author jqlin

 */

public class LongEventProducer {

    private final RingBuffer<LongEvent> ringBuffer;

    public LongEventProducer(RingBuffer<LongEvent> 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);

        }

    }

}

package com.ljq.disruptor;

import com.lmax.disruptor.WorkHandler;

/**

 * LongEvent事件消息者，消息LongEvent事件

 *

 * @author Administrator

 *

 */

public class LongEventConsumer  implements WorkHandler<LongEvent> {

    @Override

    public void onEvent(LongEvent event) throws Exception {

        System.out.println("consumer:" + Thread.currentThread().getName() + " Event: value=" + event.getValue() );

    }

}

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<LongEvent> factory = new EventFactory<LongEvent>() {

            @Override

            public LongEvent newInstance() {

                return new LongEvent();

            }

        };

        // 创建ringBuffer

        RingBuffer<LongEvent> 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<LongEvent> workerPool = new WorkerPool<LongEvent>(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));

    }

}

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);

    }

}