disruptor 史上最全之3： 8大使用场景详细图解

Posted 2022-10-26 架构师-尼恩

文章很长，而且持续更新，建议收藏起来，慢慢读！疯狂创客圈总目录 博客园版 为您奉上珍贵的学习资源 ：

• 免费赠送 :《尼恩Java面试宝典》持续更新+ 史上最全 + 面试必备 2000页+ 面试必备 + 大厂必备 +涨薪必备
• 免费赠送 经典图书:《Java高并发核心编程（卷1）》 面试必备 + 大厂必备 +涨薪必备 加尼恩免费领
• 免费赠送 经典图书:《Java高并发核心编程（卷2）》 面试必备 + 大厂必备 +涨薪必备 加尼恩免费领
• 免费赠送 经典图书:《Netty Zookeeper Redis 高并发实战》 面试必备 + 大厂必备 +涨薪必备 加尼恩免费领
• 免费赠送 经典图书:《SpringCloud Nginx高并发核心编程》 面试必备 + 大厂必备 +涨薪必备 加尼恩免费领
• 免费赠送 资源宝库： Java 必备 百度网盘资源大合集 价值>10000元 加尼恩领取

---

disruptor 史上最全 系列文章：

作为Java领域最高性能的 队列，没有之一， 大家不光要懂，而是 需要深入骨髓的搞懂。

所以，给大家奉上了下面的三篇文章，并且配备了视频进行 详细介绍：

作为Java领域最高性能的 队列，没有之一， 大家不光要懂，而是 需要深入骨髓的搞懂。

所以，给大家奉上了下面的三篇文章，并且配备了视频进行 详细介绍：

• 1 disruptor 史上最全 之1：伪共享 原理&性能对比实战
• 2 disruptor 史上最全 之2：使用和原理 图解
• 3 disruptor 史上最全 之3： 使用场景详细分析 图解
• 4 disruptor 史上最全 之4： 架构师视角的源码深入分析

总的Disruptor 的使用场景

Disruptor 它可以用来作为高性能的有界内存队列， 适用于两大场景：

• 生产者消费者场景
• 发布订阅 场景

生产者消费者场景。Disruptor的最常用的场景就是“生产者-消费者”场景，对场景的就是“一个生产者、多个消费者”的场景，并且要求顺序处理。

备注，这里和JCTool 的 MPSC 队列，刚好相反， MPSC 使用于多生产者，单消费者场景

发布订阅 场景：Disruptor也可以认为是观察者模式的一种实现， 实现发布订阅模式。

当前业界开源组件使用Disruptor的包括Log4j2、Apache Storm等，

Disruptor 使用细分场景

Disruptor是一个优秀的并发框架，可以使用在多个生产者单消费者场景

• 单生产者多消费者场景
• 多生产者单消费者场景
• 单生产者多消费者场景
• 多个消费者串行消费场景
• 菱形方式执行场景
• 链式并行执行场景
• 多组消费者相互隔离场景
• 多组消费者航道执行模式

单生产者多消费者并行场景

在并发系统中提高性能最好的方式之一就是单一写者原则，对Disruptor也是适用的。

如果在生产者单消费者 需求中仅仅有一个事件生产者，那么可以设置为单一生产者模式来提高系统的性能。

ProducerType 的类型

ProducerType 定义了生产者的类型， 两类

在这种场景下，ProducerType 的类型的 SINGLE

单生产者多消费者并行场景的参考代码

参考的代码如下：

执行结果：

以上用例的具体减少，请参见 尼恩《100wqps 日志平台实操，视频》

多生产者单消费者场景

该场景较为简单，就是多个生产者，单个消费者

其实，消费者也可以是多个

ProducerType 的类型

ProducerType 定义了生产者的类型， 两类

在这种场景下，ProducerType 的类型的 MULTI

多生产者场景的要点

在代码编写维度，多生产者单消费者场景的要点如下：

• 创建Disruptor 的时候，将ProducerType.SINGLE改为ProducerType.MULTI，

• 编写多线程生产者的相关代码即可。

多生产者场景的参考代码

参考的代码如下：

运行的结果如下

以上用例的具体减少，请参见 尼恩《100wqps 日志平台实操，视频》

单生产者多消费者竞争场景

该场景中，生产者为一个，消费者为多个，多个消费者之间， 存在着竞争关系，

也就是说，对于同一个事件event ，多个消费者 不重复消费

disruptor如何设置多个竞争消费者？

首先，得了解一下，disruptor框架的两个设置消费者的方法

大概有两点：

• 消费者需要 实现 WorkHandler 接口，而不是 EventHandler 接口
• 使用 handleEventsWithWorkerPool 设置 disruptor的 消费者，而不是 handleEventsWith 方法

在disruptor框架调用start方法之前，有两个方法设置消费者：

• disruptor.handleEventsWith(EventHandler … handlers)，将多个EventHandler的实现类传入方法，封装成一个EventHandlerGroup，实现多消费者消费。
• disruptor.handleEventsWithWorkerPool(WorkHandler … handlers)，将多个WorkHandler的实现类传入方法，封装成一个EventHandlerGroup实现多消费者消费。

那么，以上的Disruptor类的handleEventsWith，handleEventsWithWorkerPool方法的联系及区别是什么呢？
相同的在于：

两者共同点都是，将多个消费者封装到一起，供框架消费事件。

第一个不同点在于：

对于某一条事件 event，

handleEventsWith 方法返回的EventHandlerGroup，Group中的每个消费者都会对 event 进行消费，各个消费者之间不存在竞争。

handleEventsWithWorkerPool方法返回的EventHandlerGroup，Group的消费者对于同一条事件 event 不重复消费；也就是，如果c0消费了事件m，则c1不再消费事件m。

另外一个不同：

在设置消费者的时候，Disruptor类的handleEventsWith，handleEventsWithWorkerPool方法所传入的形参不同。对于独立消费的消费者，应当实现EventHandler接口。对于不重复消费的消费者，应当实现WorkHandler接口。

因此，根据消费者集合是否独立消费事件，可以对不同的接口进行实现。也可以对两种接口同时实现，具体消费流程由disruptor的方法调用决定。

演示代码如下:

执行结果

以上用例的具体减少，请参见 尼恩《100wqps 日志平台实操，视频》

多个消费者串行消费场景

在 多个消费者串行消费场景中，多个消费者，可以按照次序，消费消息。

比如：一个用户注册的Event，需要有一个Handler来存储信息，一个Hanlder来发邮件等等。

多个消费者串行消费场景案例

执行结果

菱形方式执行场景

场景特点

先并发，后串行

菱形方式执行场景案例

执行结果

链式并行执行场景

场景特点

多组消费者形成 并行链，特点是：

• 链内 串行

• 链间 并行

场景案例

执行结果

多组消费者相互隔离场景

场景特点

多组消费者 相互隔离，特点是：

• 组内 相互竞争

• 组间 相互隔离

场景案例

执行结果

多组消费者航道执行模式

场景特点

多组消费者形成 并行链，特点是：

• 组内 相互竞争

• 组之间串行依次执行

场景案例

组之间串行依次执行,组内有多个实例竞争执行

执行效果

六边形执行顺序

这是一种比较复杂的场景

场景特点

单边内部是有序的

边和边之间是并行的

参考代码

   @org.junit.Test
    public void testHexagonConsumerDisruptorWithMethodRef() throws InterruptedException 
        // 消费者线程池
        Executor executor = Executors.newCachedThreadPool();
        // 环形队列大小，2的指数
        int bufferSize = 1024;
        // 构造  分裂者 （事件分发者）
        Disruptor<LongEvent> disruptor = new Disruptor<LongEvent>(LongEvent::new, bufferSize,
                executor,
                ProducerType.SINGLE,  //多个生产者
                new YieldingWaitStrategy());

        EventHandler consumer1 = new LongEventHandlerWithName("consumer 1");
        EventHandler consumer2 = new LongEventHandlerWithName("consumer 2");
        EventHandler consumer3 = new LongEventHandlerWithName("consumer 3");
        EventHandler consumer4 = new LongEventHandlerWithName("consumer 4");
        EventHandler consumer5 = new LongEventHandlerWithName("consumer 5");
        // 连接 消费者 处理器
        // 可以使用lambda来注册一个EventHandler

        disruptor.handleEventsWith(consumer1,consumer2);
        disruptor.after(consumer1).handleEventsWith(consumer3);
        disruptor.after(consumer2).handleEventsWith(consumer4);
        disruptor.after(consumer3,consumer4).handleEventsWith(consumer5);
        // 开启 分裂者（事件分发）
        disruptor.start();
        // 获取环形队列，用于生产 事件
        RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();
        //1生产者，并发生产数据
        LongEventProducerWithTranslator producer = new LongEventProducerWithTranslator(ringBuffer);
        Thread thread = new Thread() 
            @Override
            public void run() 
                for (long i = 0; true; i++) 
                    producer.onData(i);
                    ThreadUtil.sleepSeconds(1);
                
            
        ;
        thread.start();
        ThreadUtil.sleepSeconds(5);
    

执行结果

参考文献

https://blog.csdn.net/hxg117/article/details/78064632

http://openjdk.java.net/projects/jdk8/features
http://beautynbits.blogspot.co.uk/2012/11/the-end-for-false-sharing-in-java.html
http://openjdk.java.net/jeps/142
http://mechanical-sympathy.blogspot.co.uk/2011/08/false-sharing-java-7.html
http://stackoverflow.com/questions/19892322/when-will-jvm-use-intrinsics
https://blogs.oracle.com/dave/entry/java_contented_annotation_to_help

https://www.jianshu.com/p/b38ffa33d64d

https://blog.csdn.net/MrYushiwen/article/details/123171635

https://blog.csdn.net/everyok/article/details/88889057