SpringBoot disruptor高效能佇列使用

2023-09-12 18:03:46

1、Disruptor簡介

Disruptor 是英國外匯交易公司LMAX開發的一個高效能佇列，研發的初衷是解決記憶體佇列的延遲問題（在效能測試中發現竟然與I/O操作處於同樣的數量級）。基於 Disruptor 開發的系統單執行緒能支撐每秒 600 萬訂單，2010 年在 QCon 演講後，獲得了業界關注。

其特點簡單總結如下：

開源的java框架，用於生產者-消費者場景；
高吞吐量和低延遲；
有界佇列；

disruptor在github網址為：https://github.com/LMAX-Exchange/disruptor

2、Disruptor概念

Ring Buffer：環形的緩衝區，環形陣列中的元素採用覆蓋方式，避免了jvm的GC；
Sequence Disruptor：通過順序遞增的序號來編號管理通過其進行交換的資料（事件），對資料(事件)的處理過程總是沿著序號逐個遞增處理；
Sequencer：Sequencer 是 Disruptor 的真正核心。此介面有兩個實現類 SingleProducerSequencer、MultiProducerSequencer ，它們定義在生產者和消費者之間快速、正確地傳遞資料的並行演演算法；
Sequence Barrier：用於保持對RingBuffer的 main published Sequence 和Consumer依賴的其它Consumer的 Sequence 的參照；
Wait Strategy：定義 Consumer 如何進行等待下一個事件的策略；
Event：在 Disruptor 的語意中，生產者和消費者之間進行交換的資料被稱為事件(Event)。它不是一個被 Disruptor 定義的特定型別，而是由 Disruptor 的使用者定義並指定；
EventProcessor：EventProcessor 持有特定消費者(Consumer)的 Sequence，並提供用於呼叫事件處理實現的事件迴圈(Event Loop)；
EventHandler：定義的事件處理介面，由使用者實現，用於處理事件，是 Consumer 的真正實現；
Producer：生產者，只是泛指呼叫 Disruptor 釋出事件的使用者程式碼，Disruptor 沒有定義特定介面或型別；

3、springboot+disruptor範例

在pom.xml檔案中新增依賴

		<dependency>
            <groupId>com.lmax</groupId>
            <artifactId>disruptor</artifactId>
            <version>3.3.4</version>
        </dependency>

訊息體Model

@Data
public class MessageModel {
    private String message;
}

構造EventFactory

public class HelloEventFactory implements EventFactory<MessageModel> {
    @Override
    public MessageModel newInstance() {
        return new MessageModel();
    }
}

構造消費者

@Slf4j
public class HelloEventHandler implements EventHandler<MessageModel> {
    @Override
    public void onEvent(MessageModel event, long sequence, boolean endOfBatch) {
        try {
            //這裡停止1000ms是為了確定消費訊息是非同步的
            Thread.sleep(1000);
            log.info("消費者處理訊息開始");
            if (event != null) {
                log.info("消費者消費的資訊是：{}",event);
            }
        } catch (Exception e) {
            log.info("消費者處理訊息失敗");
        }
        log.info("消費者處理訊息結束");
    }
}

構造MQManager

@Configuration
public class MqManager {
    @Bean("messageModel")
    public RingBuffer<MessageModel> messageModelRingBuffer() {
        //定義用於事件處理的執行緒池， Disruptor通過java.util.concurrent.ExecutorSerivce提供的執行緒來觸發consumer的事件處理
        ExecutorService executor = Executors.newFixedThreadPool(2);
        //指定事件工廠
        HelloEventFactory factory = new HelloEventFactory();
        //指定ringbuffer位元組大小，必須為2的N次方（能將求模運算轉為位運算提高效率），否則將影響效率
        int bufferSize = 1024 * 256;
        //單執行緒模式，獲取額外的效能
        Disruptor<MessageModel> disruptor = new Disruptor<>(factory, bufferSize, executor, ProducerType.SINGLE, new BlockingWaitStrategy());
        //設定事件業務處理器---消費者
        disruptor.handleEventsWith(new HelloEventHandler());
        //啟動disruptor執行緒
        disruptor.start();
        //獲取ringbuffer環，用於接取生產者生產的事件
        RingBuffer<MessageModel> ringBuffer = disruptor.getRingBuffer();
        return ringBuffer;
    }
}

構造生產者

@Configuration
public class MqManager {
    @Bean("messageModel")
    public RingBuffer<MessageModel> messageModelRingBuffer() {
        //定義用於事件處理的執行緒池， Disruptor通過java.util.concurrent.ExecutorSerivce提供的執行緒來觸發consumer的事件處理
        ExecutorService executor = Executors.newFixedThreadPool(2);
        //指定事件工廠
        HelloEventFactory factory = new HelloEventFactory();
        //指定ringbuffer位元組大小，必須為2的N次方（能將求模運算轉為位運算提高效率），否則將影響效率
        int bufferSize = 1024 * 256;
        //單執行緒模式，獲取額外的效能
        Disruptor<MessageModel> disruptor = new Disruptor<>(factory, bufferSize, executor, ProducerType.SINGLE, new BlockingWaitStrategy());
        //設定事件業務處理器---消費者
        disruptor.handleEventsWith(new HelloEventHandler());
        //啟動disruptor執行緒
        disruptor.start();
        //獲取ringbuffer環，用於接取生產者生產的事件
        RingBuffer<MessageModel> ringBuffer = disruptor.getRingBuffer();
        return ringBuffer;
    }
}

測試

	/**
     * 專案內部使用Disruptor做訊息佇列
     * @throws Exception
     */
    @Test
    public void sayHelloMqTest() throws Exception{
        helloEventProducer.sayHelloMq("Hello world!");
        log.info("訊息佇列已傳送完畢");
        //這裡停止2000ms是為了確定是處理訊息是非同步的
        Thread.sleep(2000);
    }

執行結果如下

4、小結

參照disruptor作為內部的高效能佇列，應用於生產者-消費者模式中還是非常nice的，後面若有工程需求可以嘗試一下。

到此這篇關於SpringBoot disruptor高效能佇列使用的文章就介紹到這了,更多相關SpringBoot disruptor內容請搜尋it145.com以前的文章或繼續瀏覽下面的相關文章希望大家以後多多支援it145.com！

SpringBoot disruptor高效能佇列使用

目錄

1、Disruptor簡介

2、Disruptor概念

3、springboot+disruptor範例

4、小結

熱門文章