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 :进程日志事件处理器 */@Componentpublic 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 环形队列 */@Componentpublic 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高性能线程消息传递并发框架的示例分析"这篇文章对大家有帮助，同时也希望大家多多支持，关注行业资讯频道，更多相关知识等着你来学习!