前言碎语
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，完成功能，有兴趣的可以对比下。
第一步，定义事件类型
|
1 2 3 4 5 6 7 8 9 10 11 12 13 |
/** * 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; } } |
第二步，定义事件工厂
|
1 2 3 4 5 6 7 8 9 10 |
/** * Created by kl on 2018/8/24. * Content :进程日志事件工厂类 */ public class LoggerEventFactory implements EventFactory{ @Override public LoggerEvent newInstance() { return new LoggerEvent(); } } |
第三步，定义数据处理器
|
1 2 3 4 5 6 7 8 9 10 11 12 13 |
/** * 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实操类，定义事件发布方法，发布事件
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 |
/** * 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的使用，启动项目后就会不断的发布日志事件，处理器会将事件内容通过websocket传送到前端页面上展示，
boot-websocket-log项目地址： https://gitee.com/kailing/boot-websocket-log
Disruptor是高性能的进程内线程间的数据交换框架，特别适合日志类的处理。Disruptor也是从 https://github.com/alipay/sofa-tracer 了解到的，这是蚂蚁金服 团队开源的分布式链路追踪项目，其中日志处理部分就是使用了Disruptor。
以上就是浅析Disruptor高性能线程消息传递并发框架的详细内容，更多关于Disruptor线程消息传递并发框架的资料请关注其它相关文章！
原文链接：http://www.kailing.pub/article/index/arcid/208.html
查看更多关于浅析Disruptor高性能线程消息传递并发框架的详细内容...