淺析Disruptor高性能線程消息傳遞并發(fā)框架

更新時間：2022年03月01日 11:33:53 作者：kl

這篇文章主要為大家介紹了Disruptor高性能線程消息傳遞并發(fā)框架的簡單分析，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步

前言碎語

Disruptor是英國LMAX公司開源的高性能的線程間傳遞消息的并發(fā)框架，和jdk中的BlockingQueue非常類似，但是性能卻是BlockingQueue不能比擬的，下面是官方給出的一分測試報告，可以直觀的看出兩者的性能區(qū)別：

Disruptor 項目地址：https://github.com/LMAX-Exchange/disruptor

核心概念？

這么性能炸裂的框架肯定要把玩一番，試用前，我們先了解下disruptor的主要的概念，然后結合樓主的weblog項目（之前使用的BlockingQueue），來實踐下

RingBuffer：環(huán)形的緩沖區(qū)，消息事件信息的載體。曾經(jīng) RingBuffer 是 Disruptor 中的最主要的對象，但從3.0版本開始，其職責被簡化為僅僅負責對通過 Disruptor 進行交換的數(shù)據(jù)（事件）進行存儲和更新。在一些更高級的應用場景中，Ring Buffer 可以由用戶的自定義實現(xiàn)來完全替代。

Event：定義生產(chǎn)者和消費者之間進行交換的數(shù)據(jù)類型。

EventFactory：創(chuàng)建事件的工廠類接口，由用戶實現(xiàn)，提供具體的事件

EventHandler：事件處理接口，由用戶實現(xiàn)，用于處理事件。

目前為止，我們了解以上核心內容即可，更多的詳情，可以移步wiki文檔：https://github.com/LMAX-Exchange/disruptor

核心架構圖：

實踐Disruptor

改造boot-websocket-log項目，這是一個典型的生產(chǎn)者消費者模式的實例。然后將BlockingQueue替換成Disruptor，完成功能，有興趣的可以對比下。

第一步，定義事件類型

/**
 * Created by kl on 2018/8/24.
 * Content :進程日志事件內容載體
 */
public class LoggerEvent {
    private LoggerMessage log;
    public LoggerMessage getLog() {
        return log;
    }
    public void setLog(LoggerMessage log) {
        this.log = log;
    }
}

第二步，定義事件工廠

/**
 * Created by kl on 2018/8/24.
 * Content :進程日志事件工廠類
 */
public class LoggerEventFactory implements EventFactory{
    @Override
    public LoggerEvent newInstance() {
        return new LoggerEvent();
    }
}

第三步，定義數(shù)據(jù)處理器

/**
 * Created by kl on 2018/8/24.
 * Content :進程日志事件處理器
 */
@Component
public class LoggerEventHandler implements EventHandler{
    @Autowired
    private SimpMessagingTemplate messagingTemplate;
    @Override
    public void onEvent(LoggerEvent stringEvent, long l, boolean b) {
        messagingTemplate.convertAndSend("/topic/pullLogger",stringEvent.getLog());
    }
}

第四步，創(chuàng)建Disruptor實操類，定義事件發(fā)布方法，發(fā)布事件

/**
 * Created by kl on 2018/8/24.
 * Content :Disruptor 環(huán)形隊列
 */
@Component
public class LoggerDisruptorQueue {
    private Executor executor = Executors.newCachedThreadPool();
    // The factory for the event
    private LoggerEventFactory factory = new LoggerEventFactory();
    private FileLoggerEventFactory fileLoggerEventFactory = new FileLoggerEventFactory();
    // Specify the size of the ring buffer, must be power of 2.
    private int bufferSize = 2 * 1024;
    // Construct the Disruptor
    private Disruptordisruptor = new Disruptor<>(factory, bufferSize, executor);;
    private DisruptorfileLoggerEventDisruptor = new Disruptor<>(fileLoggerEventFactory, bufferSize, executor);;
    private static  RingBufferringBuffer;
    private static  RingBufferfileLoggerEventRingBuffer;
    @Autowired
    LoggerDisruptorQueue(LoggerEventHandler eventHandler,FileLoggerEventHandler fileLoggerEventHandler) {
        disruptor.handleEventsWith(eventHandler);
        fileLoggerEventDisruptor.handleEventsWith(fileLoggerEventHandler);
        this.ringBuffer = disruptor.getRingBuffer();
        this.fileLoggerEventRingBuffer = fileLoggerEventDisruptor.getRingBuffer();
        disruptor.start();
        fileLoggerEventDisruptor.start();
    }
    public static void publishEvent(LoggerMessage log) {
        long sequence = ringBuffer.next();  // Grab the next sequence
        try {
            LoggerEvent event = ringBuffer.get(sequence); // Get the entry in the Disruptor
            // for the sequence
            event.setLog(log);  // Fill with data
        } finally {
            ringBuffer.publish(sequence);
        }
    }
    public static void publishEvent(String log) {
        if(fileLoggerEventRingBuffer == null) return;
        long sequence = fileLoggerEventRingBuffer.next();  // Grab the next sequence
        try {
            FileLoggerEvent event = fileLoggerEventRingBuffer.get(sequence); // Get the entry in the Disruptor
            // for the sequence
            event.setLog(log);  // Fill with data
        } finally {
            fileLoggerEventRingBuffer.publish(sequence);
        }
    }
}