Disruptor多個消費者不重復處理生產者發送過來的消息

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

 

3、LongEvent事件消息者

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

}

 

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

 

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