浅析Disruptor高性能线程消息传递并发框架

浅析Disruptor高性能线程消息传递并发框架

目录前言碎语核心概念？实践Disruptor文末结语

前言碎语

Disruptor是英国LMAX公司开源的高性能的线程间传递消息的并发框架，和jdk中的BlockingQueue非常类似，但是性能却是BlockingQueue不能比拟的，下面是官方给出的一分测试报告，可以直观的看出两者的性能区别：

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

核心概念？

这么性能炸裂的框架肯定要把玩一番，试用前，我们先了解下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 bhttp://y 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 thhttp://e 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的使用，启动项目后就会不断的发布日志事件，处理器会将事件内容通过websocket传送到前端页面上展示，

boot-websocket-log项目地址：https://gitee.com/kailing/boot-websocket-log

Disruptor是高性能的进程内线程间的数据交换框架，特别适合日志类的处理。Disruptor也是从https://github.com/alipay/sofa-tracer了解到的，这是蚂蚁金服 团队开源的分布式链路追踪项目，其中日志处理部分就是使用了Disruptor。

以上就是浅析Disruptor高性能线程消息传递并发框架的详细内容，更多关于Disruptor线程消息传递并发框架的资料请关注我们其它相关文章！

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。