什么是 Disruptor？

Disruptor是一個高性能的異步處理框架，或者可以認為是最快的消息框架（輕量的JMS），也可以認為是一個觀察者模式的實現(xiàn)，或者事件監(jiān)聽模式的實現(xiàn)

創(chuàng)新互聯(lián)專注于獲嘉企業(yè)網(wǎng)站建設,響應式網(wǎng)站,購物商城網(wǎng)站建設。獲嘉網(wǎng)站建設公司,為獲嘉等地區(qū)提供建站服務。全流程按需定制網(wǎng)站，專業(yè)設計，全程項目跟蹤，創(chuàng)新互聯(lián)專業(yè)和態(tài)度為您提供的服務

性能遠遠高于傳統(tǒng)的BlockingQueue容器

Disruptor使用觀察者模式，主動將消息發(fā)送給消費者，而不是等消費者從隊列中取，在無鎖的情況下, 實現(xiàn)queue（環(huán)形, RingBuffer）的并發(fā)操作, 性能遠高于BlockingQueue

Disruptor 的設計思想

環(huán)形數(shù)組結構

為了避免垃圾回收，使用數(shù)組，數(shù)組對處理器的緩存機制更加友好

數(shù)組長度為 2^n，通過位運算，加快定位速度，下標采用遞增的方式，不用擔心索引溢出

無鎖設計

每個生產(chǎn)者或者消費者線程，會先申請可以操作的元素在數(shù)組中的位置，申請到之后，直接在該位置寫入或者讀取數(shù)據(jù)

Disruptor 實現(xiàn)生產(chǎn)消費模型

pom

    com.lmax
    disruptor
    3.2.1

LongEvent

// 聲明一個Event來包含需要傳遞的數(shù)據(jù)
public class LongEvent {
    private Long value;

    public Long getValue() {
        return value;
    }

    public void setValue(Long value) {
        this.value = value;
    }
}

LongEventFactory

// Event工廠
public class LongEventFactory implements EventFactory {

    public LongEvent newInstance() {
        return new LongEvent();
    }
}

LongEventHandler

// 事件消費者
public class LongEventHandler implements EventHandler {

    public void onEvent(LongEvent event, long sequence, boolean endOfBatch) throws Exception {
        System.out.println("消費者:"+event.getValue());
    }
}

LongEventProducer

public class LongEventProducer {
    private RingBuffer ringBuffer;

    public LongEventProducer(RingBuffer ringBuffer) {
        this.ringBuffer = ringBuffer;
    }

    public void onData(ByteBuffer byteBuffer) {
        // 獲取事件隊列下標位置
        long sequence = ringBuffer.next();
        try {
            // 取出空隊列
            LongEvent longEvent = ringBuffer.get(sequence);
            // 賦值
            longEvent.setValue(byteBuffer.getLong(0));
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            System.out.println("生產(chǎn)者發(fā)送數(shù)據(jù)。。。");
            // 發(fā)送數(shù)據(jù)
            ringBuffer.publish(sequence);
        }
    }
}

Main

public class Main {
    public static void main(String[] args) {
        // 創(chuàng)建可緩存線程池
        ExecutorService executorService = Executors.newCachedThreadPool();
        // 創(chuàng)建工廠
        EventFactory eventFactory = new LongEventFactory();
        // 創(chuàng)建ringBufferSize
        int ringBufferSize = 1024 * 1024;
        // 創(chuàng)建disruptor
        // MULTI表示可以多個生產(chǎn)者
        Disruptor longEventDisruptor = new Disruptor(eventFactory, ringBufferSize, executorService, ProducerType.MULTI, new YieldingWaitStrategy());
        // 注冊消費者
        longEventDisruptor.handleEventsWith(new LongEventHandler());
        // 啟動
        longEventDisruptor.start();
        // 創(chuàng)建RingBuffer容器
        RingBuffer ringBuffer = longEventDisruptor.getRingBuffer();
        // 創(chuàng)建生產(chǎn)者
        LongEventProducer longEventProducer = new LongEventProducer(ringBuffer);
        // 指定緩沖區(qū)大小
        ByteBuffer byteBuffer = ByteBuffer.allocate(8);
        for (int i = 0; i < 100; i++) {
            byteBuffer.putLong(0, i);
            longEventProducer.onData(byteBuffer);
        }
        executorService.shutdown();
        longEventDisruptor.shutdown();
    }
}

什么是 RingBuffer

它是一個環(huán)（首尾相接的環(huán)），作用是存儲數(shù)據(jù)，實現(xiàn)不同線程之間的數(shù)據(jù)傳輸

如果圓環(huán)滿了，它會將金數(shù)據(jù)覆蓋，如上圖：現(xiàn)在12的區(qū)域的下個區(qū)域目前是3，如果有新的數(shù)據(jù)到來，那么指針往下移的時候就會把區(qū)域3的數(shù)據(jù)給覆蓋變成13，框架提供了一系列幫助我們平行消費的監(jiān)控，會很好的控制生產(chǎn)者和消費者之間的速度，從而達到生產(chǎn)和消費之間的平衡

RingBuffer 為什么效率高？

采用數(shù)組，數(shù)組支持索引訪問

數(shù)組的內存分配是預先加載的，一但指定大小創(chuàng)建后，就一直存在，這也意味著不需要花大量的時間做垃圾回收，而阻塞隊列采用鏈表實現(xiàn)，需要不斷的刪除、創(chuàng)建節(jié)點

Disruptor的核心概念

• RingBuffer：Disruptor 底層數(shù)據(jù)結構實現(xiàn)，核心類，是線程間交換數(shù)據(jù)的中轉地

• Sequence：序號，聲明一個序號，用于跟蹤 ringbuffer 中任務的變化和消費者的消費情況

• Sequencer：生產(chǎn)者與緩存 RingBuffer 之間的橋梁，單生產(chǎn)者與多生產(chǎn)者分別對應于兩個實現(xiàn)SingleProducerSequencer 與 MultiProducerSequencer，Sequencer 用于向 RingBuffer 申請空間，使用 publish 方法通過 waitStrategy 通知所有在等待可消費事件的 SequenceBarrie
• SequenceBarrier：序號柵欄，管理和協(xié)調生產(chǎn)者的游標序號和各個消費者的序號，確保生產(chǎn)者不會覆蓋消費者未來得及處理的消息，確保存在依賴的消費者之間能夠按照正確的順序處理

• WaitStrategy：有多種實現(xiàn)，用以表示當無可消費事件時，消費者的等待策略

• Event：消費事件

• EventProcessor：事件處理器，監(jiān)聽 RingBuffer 的事件，并消費可用事件，從 RingBuffer 讀取的事件會交由實際的生產(chǎn)者實現(xiàn)類來消費，它會一直偵聽下一個可用的序號，直到該序號對應的事件已經(jīng)準備好

• EventHandler：業(yè)務處理器，是實際消費者的接口，完成具體的業(yè)務邏輯實現(xiàn)，第三方實現(xiàn)該接口，代表著消費者

• Producer：生產(chǎn)者接口，第三方線程充當該角色，producer 向 RingBuffer 寫入事件