Disruptor簡單應用

前段時間閑得蛋疼就嘗試翻譯了一下有關Disruptor的一些文章，第一次做這事，爛得自己也不忍回頭去看了。。。

今天寫日志看到那幾篇文章，於是想寫一個簡單的例子，好讓一些對Disruptor有興趣但不是很明白的人更快的了解Disruptor的基本用法。

Disruptor使用起來非常簡單，初始化－>啟動消費者線程，然后每當生產者產生資源就往disruptor里放。

下面是一個接收短信狀態報告的例子。

首先自定義一個Event類:

public class DeliveryReportEvent {
    private DeliveryReport deliveryReport;

    public DeliveryReport getDeliveryReport() {
        return deliveryReport;
    }

    public void setDeliveryReport(DeliveryReport deliveryReport) {
        this.deliveryReport = deliveryReport;
    }


    public final static EventFactory<DeliveryReportEvent> EVENT_FACTORY = new EventFactory<DeliveryReportEvent>() {
        public DeliveryReportEvent newInstance() {
            return new DeliveryReportEvent();
        }
    };
}

代碼很簡單，就是用來存放短信狀態報告DeliveryReport類的，外加一個EventFactory，用來生成Event，也很簡單。

然后寫一個EventHandler類，用來處理Event，

public class DeliveryReportEventHandler implements EventHandler<DeliveryReportEvent> {
//    private static DeliveryReportRepository repository = new DeliveryReportRepository();
    public void onEvent(DeliveryReportEvent event, long sequence,
            boolean endOfBatch) throws Exception {
        System.out.println(event.getDeliveryReport().getMessageId());
//        repository.updateDeliveryReport(event.getDeliveryReport());
    }
}

三個參數，event就是生產者向Disruptor發布資源生產的事件，sequence是這個事件在ringbuffer中的序列號，endOfBatch指明該事件是不是ringbuffer中的最后一個事件。

在onEvent方法里處理消費者要做的事。本例直接打印狀態報告ID或是更新數據庫中的狀態報告。

在這里我寫了一個輔助類：

public class DisruptorHelper {
    /**
     * ringbuffer容量，最好是2的N次方
     */
    private static final int BUFFER_SIZE = 1024 * 8;
    private RingBuffer<DeliveryReportEvent> ringBuffer;
    private SequenceBarrier sequenceBarrier;
    private DeliveryReportEventHandler handler;
    private BatchEventProcessor<DeliveryReportEvent> batchEventProcessor;
    private  static DisruptorHelper instance;
    private static boolean inited = false;
    private DisruptorHelper(){
        ringBuffer = new RingBuffer<DeliveryReportEvent>(
                DeliveryReportEvent.EVENT_FACTORY, new SingleThreadedClaimStrategy(
                        BUFFER_SIZE), new YieldingWaitStrategy());
        sequenceBarrier = ringBuffer.newBarrier();
        handler = new DeliveryReportEventHandler();
        batchEventProcessor = new BatchEventProcessor<DeliveryReportEvent>(
                    ringBuffer, sequenceBarrier, handler);
        ringBuffer.setGatingSequences(batchEventProcessor
                .getSequence());
    }

    public static void initAndStart(){
        instance = new DisruptorHelper();
        new Thread(instance.batchEventProcessor).start();
        inited = true;
    }

    public static void shutdown(){
        if(!inited){
            throw new RuntimeException("Disruptor還沒有初始化！");
        }
        instance.shutdown0();
    }

    private void shutdown0(){
        batchEventProcessor.halt();
    }
    private void produce0(DeliveryReport deliveryReport) {
        //獲取下一個序列號
        long sequence = ringBuffer.next();
        //將狀態報告存入ringBuffer的該序列號中
        ringBuffer.get(sequence).setDeliveryReport(deliveryReport);
        //通知消費者該資源可以消費
        ringBuffer.publish(sequence);
    }

    /**
     * 將狀態報告放入資源隊列，等待處理
     * @param deliveryReport
     */
    public static void produce(DeliveryReport deliveryReport) {
        if(!inited){
            throw new RuntimeException("Disruptor還沒有初始化！");
        }
        instance.produce0(deliveryReport);
    }
}

哈哈，因為本例的initAndStart/shutdown方法是在一個loadonstartup的servlet中的init/destory方法里調用，所以就不加synchronized了，執行前也不判斷inited是否為true了。

最后是生產者調用的方法，在接收短信狀態報告的webservice里：

public void NotifySmsDeliveryReport(DeliveryReport deliveryReport) {
...
...
//向disruptor發布資源
DisruptorHelper.produce(deliveryReport);
}

 

關於disruptor的性能，可以去官網查看測試數據。也可以從官方svn檢出代碼，自己運行一下src/perf里的性能測試方法，

 

本例用到的包只有一個disruptor-2.7.1.jar，

<dependency>
  <groupId>com.googlecode.disruptor</groupId>
  <artifactId>disruptor</artifactId>

  <version>2.7.1</version>
</dependency>