温馨提示×

温馨提示×

您好,登录后才能下订单哦!

密码登录×
登录注册×
其他方式登录
点击 登录注册 即表示同意《亿速云用户服务条款》

Disruptor高性能线程消息传递并发框架的示例分析

发布时间:2022-03-01 14:18:49 来源:亿速云 阅读:127 作者:小新 栏目:开发技术

这篇文章主要介绍了Disruptor高性能线程消息传递并发框架的示例分析,具有一定借鉴价值,感兴趣的朋友可以参考下,希望大家阅读完这篇文章之后大有收获,下面让小编带着大家一起了解一下。

    前言碎语

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

    Disruptor高性能线程消息传递并发框架的示例分析

    Disruptor高性能线程消息传递并发框架的示例分析

    核心概念?

    这么性能炸裂的框架肯定要把玩一番,试用前,我们先了解下disruptor的主要的概念,然后结合楼主的weblog项目(之前使用的BlockingQueue),来实践下

    RingBuffer:环形的缓冲区,消息事件信息的载体。曾经 RingBuffer 是 Disruptor 中的最主要的对象,但从3.0版本开始,其职责被简化为仅仅负责对通过 Disruptor 进行交换的数据(事件)进行存储和更新。在一些更高级的应用场景中,Ring Buffer 可以由用户的自定义实现来完全替代。

    Event:定义生产者和消费者之间进行交换的数据类型。

    EventFactory:创建事件的工厂类接口,由用户实现,提供具体的事件

    EventHandler:事件处理接口,由用户实现,用于处理事件。

    目前为止,我们了解以上核心内容即可,更多的详情,可以移步wiki文档:https://github.com/LMAX-Exchange/disruptor

    核心架构图:

    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高性能线程消息传递并发框架的示例分析”这篇文章对大家有帮助,同时也希望大家多多支持亿速云,关注亿速云行业资讯频道,更多相关知识等着你来学习!

    向AI问一下细节

    免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。

    AI