Disruptor高性能線程消息傳遞并發框架的示例分析

這篇文章主要介紹了Disruptor高性能線程消息傳遞并發框架的示例分析，具有一定借鑒價值，感興趣的朋友可以參考下，希望大家閱讀完這篇文章之后大有收獲，下面讓小編帶著大家一起了解一下。

前言碎語

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

核心概念？

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

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

Event：定義生產者和消費者之間進行交換的數據類型。

EventFactory：創建事件的工廠類接口，由用戶實現，提供具體的事件

EventHandler：事件處理接口，由用戶實現，用于處理事件。

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

核心架構圖：

實踐Disruptor

改造boot-websocket-log項目，這是一個典型的生產者消費者模式的實例。然后將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();
    }
}

第三步，定義數據處理器

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

第四步，創建Disruptor實操類，定義事件發布方法，發布事件

/**
 * Created by kl on 2018/8/24.
 * Content :Disruptor 環形隊列
 */
@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);
        }
    }
}

感謝你能夠認真閱讀完這篇文章，希望小編分享的“Disruptor高性能線程消息傳遞并發框架的示例分析”這篇文章對大家有幫助，同時也希望大家多多支持億速云，關注億速云行業資訊頻道，更多相關知識等著你來學習!