这篇文章主要介绍了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
核心架构图:
改造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高性能线程消息传递并发框架的示例分析”这篇文章对大家有帮助,同时也希望大家多多支持亿速云,关注亿速云行业资讯频道,更多相关知识等着你来学习!
免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。