disruptor使用笔记

发表于 | 更新于 | 分类于 | 评论数： | 阅读次数：

前言

最近做一些日志的采集工作 jdk的队列 有点顶不住了
然后看到 很多知名的中间件 工具 都用这个来替代
由于这个东西用的人很多 文档非常丰富 这里只做例子 实际架构图等等 去官网看即可

参考文档:
美团:https://tech.meituan.com/2016/11/18/disruptor.html
官方:http://lmax-exchange.github.io/disruptor/
动态消费者博客:https://zhuanlan.zhihu.com/p/100386603###
https://www.cnblogs.com/luozhiyun/p/11631305.html

基础概念
  • nP-nC n个生产者-n个消费者
  • Ring Buffer
    Ring Buffer在3.0版本以前被认为是Disruptor的核心组件，但是在之后的版本中只是负责存储和更新数据。在一些高级使用案例中用户也能进行自定义
  • Sequence
    Disruptor使用一组Sequence来作为一个手段来标识特定的组件的处理进度( RingBuffer/Consumer )。每个消费者和Disruptor本身都会维护一个Sequence。虽然一个 AtomicLong 也可以用于标识进度，但定义 Sequence 来负责该问题还有另一个目的，那就是防止不同的 Sequence 之间的CPU缓存伪共享(Flase Sharing)问题。
  • Sequencer
    Sequencer是Disruptor的真正核心。此接口有两个实现类 SingleProducerSequencer、MultiProducerSequencer ，它们定义在生产者和消费者之间快速、正确地传递数据的并发算法。
  • Sequence Barrier
    保持Sequencer和Consumer依赖的其它Consumer的 Sequence 的引用。除此之外还定义了决定 Consumer 是否还有可处理的事件的逻辑。
  • Wait Strategy
    Wait Strategy决定了一个消费者怎么等待生产者将事件（Event）放入Disruptor中。
  • Event
    从生产者到消费者传递的数据叫做Event。它不是一个被 Disruptor 定义的特定类型，而是由 Disruptor 的使用者定义并指定。
  • EventProcessor
    持有特定的消费者的Sequence，并且拥有一个主事件循环（main event loop）用于处理Disruptor的事件。其中BatchEventProcessor是其具体实现，实现了事件循环（event loop），并且会回调到实现了EventHandler的已使用过的实例中。
  • EventHandler
    由用户实现的接口，用于处理事件，是 Consumer 的真正实现
  • Producer
    生产者，只是泛指调用 Disruptor 发布事件的用户代码，Disruptor 没有定义特定接口或类型
    示例1-直接使用
    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
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    60
    61
    62
    63
    64
    65
    66
    67
    68
    69
    70
    71
    72
    73
    74
    75
    76
    77
    78
    79
    80
    81
    82
    83
    84
    package com.ming;

    import com.lmax.disruptor.BlockingWaitStrategy;
    import com.lmax.disruptor.EventFactory;
    import com.lmax.disruptor.EventHandler;
    import com.lmax.disruptor.RingBuffer;
    import com.lmax.disruptor.dsl.Disruptor;
    import com.lmax.disruptor.dsl.ProducerType;
    import org.junit.Test;

    import java.util.concurrent.ThreadFactory;

    /**
     * 测试 disruptor
     *
     * @author ming
     * @date 2020-10-27 10:30
     */
    public class TestDisruptor {
        /**
         * 基于美团文档的 示例  增加一个消费者
         *
         * @author ming
         * @date 2020-10-27 15:17
         */
        @Test
        public void test() throws InterruptedException {
    // 队列中的元素
            class Element {

                private int value;

                public int get() {
                    return value;
                }

                public void set(int value) {
                    this.value = value;
                }

            }

            // 生产者的线程工厂
            ThreadFactory threadFactory = r -> new Thread(r, "simpleThread");

            // RingBuffer生产工厂,初始化RingBuffer的时候使用
            EventFactory<Element> factory = Element::new;

            // 处理Event的handler
            EventHandler<Element> handler1 = (element, sequence, endOfBatch) -> System.out.println("Element1: " + element.get());
            EventHandler<Element> handler2 = (element, sequence, endOfBatch) -> System.out.println("Element2: " + element.get());

            // 阻塞策略
            BlockingWaitStrategy strategy = new BlockingWaitStrategy();

            // 指定RingBuffer的大小
            int bufferSize = 16;

            // 创建disruptor，采用单生产者模式
            Disruptor<Element> disruptor = new Disruptor(factory, bufferSize, threadFactory, ProducerType.SINGLE, strategy);
            // 设置EventHandler
            disruptor.handleEventsWith(handler1);
            disruptor.handleEventsWith(handler2);

            // 启动disruptor的线程
            disruptor.start();

            RingBuffer<Element> ringBuffer = disruptor.getRingBuffer();

            for (int l = 0; true; l++) {
                // 获取下一个可用位置的下标
                long sequence = ringBuffer.next();
                try {
                    // 返回可用位置的元素
                    Element event = ringBuffer.get(sequence);
                    // 设置该位置元素的值
                    event.set(l);
                } finally {
                    ringBuffer.publish(sequence);
                }
                Thread.sleep(10);
            }
        }
    }
示例2-单例使用

包装成一个单例子 方便其他地方调用 个人感觉强行写成单例 。。。有点丑陋

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
package com.ming.core.disruptor;

/**
 * 测试disruptor的element
 *
 * @author ming
 * @date 2020-10-27 16:17
 */
public class Element {

    private Integer value;

    public Integer get() {
        return value;
    }

    public void set(Integer value) {
        this.value = value;
    }

    public void clear() {
        value = null;
    }
}

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
49
50
51
52
53
54
55
56
57
58
59
package com.ming.core.disruptor;

import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;

import java.util.concurrent.ThreadFactory;

/**
 * 测试 disruptor单例
 *
 * @author ming
 * @date 2020-10-27 16:16
 */
public class ElementDisruptorSingleton {
    private static volatile Disruptor<Element> elementDisruptor;

    private ElementDisruptorSingleton() {
    }


    /**
     * 单例对象
     *
     * @return elementDisruptor
     * @author ming
     * @date 2020-10-27 16:24
     */
    public static Disruptor<Element> getInstance() {
        if (null == elementDisruptor) {
            synchronized (Disruptor.class) {
                if (null == elementDisruptor) {
                    // 生产者的线程工厂
                    ThreadFactory threadFactory = r -> new Thread(r, "simpleThread");
                    // RingBuffer生产工厂,初始化RingBuffer的时候使用
                    EventFactory<Element> factory = Element::new;
                    // 处理Event的handler
                    EventHandler<Element> handler1 = (element, sequence, endOfBatch) -> System.out.println("Element1: " + element.get());
                    EventHandler<Element> handler2 = (element, sequence, endOfBatch) -> System.out.println("Element2: " + element.get());
                    // 阻塞策略
                    BlockingWaitStrategy strategy = new BlockingWaitStrategy();
                    // 指定RingBuffer的大小
                    int bufferSize = 16;
                    // 创建disruptor，采用单生产者模式
                    Disruptor<Element> disruptor = new Disruptor(factory, bufferSize, threadFactory, ProducerType.SINGLE, strategy);
                    // 设置EventHandler 并且后置清理消费过的数据
                    disruptor.handleEventsWith(handler1);
                    disruptor.handleEventsWith(handler2);
                    elementDisruptor = disruptor;
                    // 启动disruptor的线程
                    elementDisruptor.start();
                }
            }
        }
        return elementDisruptor;
    }
}
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
package com.ming.core.disruptor;

import com.lmax.disruptor.RingBuffer;

/**
 * 测试disruptor
 *
 * @author ming
 * @date 2020-10-27 16:18
 */
public class TestDisruptor {
    public static void main(String[] args) {
        RingBuffer<Element> ringBuffer = ElementDisruptorSingleton.getInstance().getRingBuffer();
        for (int l = 0; true; l++) {
            // 获取下一个可用位置的下标
            long sequence = ringBuffer.next();
            try {
                // 返回可用位置的元素
                Element event = ringBuffer.get(sequence);
                // 设置该位置元素的值
                event.set(l);
            } finally {
                ringBuffer.publish(sequence);
            }
            try {
                Thread.sleep(10);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}
示例3-spring中使用

spring中 注册到ioc容器中

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
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
package com.ming.core.disruptor;

import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.concurrent.ThreadFactory;

@Component
public class ElementDisruptorService {
    private volatile Disruptor<Element> elementDisruptor;

    @PostConstruct
    public void init() {
        //初始化disruptor
        // 生产者的线程工厂
        ThreadFactory threadFactory = r -> new Thread(r, "simpleThread");
        // RingBuffer生产工厂,初始化RingBuffer的时候使用
        EventFactory<Element> factory = Element::new;
        // 处理Event的handler
        EventHandler<Element> handler1 = (element, sequence, endOfBatch) -> System.out.println("Element1: " + element.get());
        EventHandler<Element> handler2 = (element, sequence, endOfBatch) -> System.out.println("Element2: " + element.get());
        // 阻塞策略
        BlockingWaitStrategy strategy = new BlockingWaitStrategy();
        // 指定RingBuffer的大小
        int bufferSize = 16;
        // 创建disruptor，采用单生产者模式
        elementDisruptor = new Disruptor(factory, bufferSize, threadFactory, ProducerType.SINGLE, strategy);
        // 设置EventHandler 并且后置清理消费过的数据
        elementDisruptor.handleEventsWith(handler1);
        elementDisruptor.handleEventsWith(handler2);
        // 启动disruptor的线程
        elementDisruptor.start();
    }

    @PreDestroy
    public void destroy() {
        //销毁 disruptor
        elementDisruptor.shutdown();
    }

    public void sendMessage() {
        RingBuffer<Element> ringBuffer = elementDisruptor.getRingBuffer();
        for (int l = 0; true; l++) {
            // 获取下一个可用位置的下标
            long sequence = ringBuffer.next();
            try {
                // 返回可用位置的元素
                Element event = ringBuffer.get(sequence);
                // 设置该位置元素的值
                event.set(l);
            } finally {
                ringBuffer.publish(sequence);
            }
            try {
                Thread.sleep(10);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

优化

单生产者和多生产者

如果确定只有一个线程生产 将disruptor设置为单生产者来提高性能
ProducerType.SINGLE

1
Disruptor<Element> disruptor = new Disruptor(factory, bufferSize, threadFactory, ProducerType.SINGLE, strategy);

等待策略 strategy

  • BlockingWaitStrategy
    Disruptor的默认策略是BlockingWaitStrategy。在BlockingWaitStrategy内部是使用锁和condition来控制线程的唤醒。BlockingWaitStrategy是最低效的策略，但其对CPU的消耗最小并且在各种不同部署环境中能提供更加一致的性能表现

  • SleepingWaitStrategy
    SleepingWaitStrategy 的性能表现跟 BlockingWaitStrategy 差不多，对 CPU 的消耗也类似，但其对生产者线程的影响最小，通过使用LockSupport.parkNanos(1)来实现循环等待。一般来说Linux系统会暂停一个线程约60µs，这样做的好处是，生产线程不需要采取任何其他行动就可以增加适当的计数器，也不需要花费时间信号通知条件变量。但是，在生产者线程和使用者线程之间移动事件的平均延迟会更高。它在不需要低延迟并且对生产线程的影响较小的情况最好。一个常见的用例是异步日志记录。

  • YieldingWaitStrategy
    YieldingWaitStrategy是可以使用在低延迟系统的策略之一。YieldingWaitStrategy将自旋以等待序列增加到适当的值。在循环体内，将调用Thread.yield（），以允许其他排队的线程运行。在要求极高性能且事件处理线数小于 CPU 逻辑核心数的场景中，推荐使用此策略；例如，CPU开启超线程的特性。

  • BusySpinWaitStrategy
    性能最好，适合用于低延迟的系统。在要求极高性能且事件处理线程数小于CPU逻辑核心树的场景中，推荐使用此策略；例如，CPU开启超线程的特性。

及时清理ring buffer中的对象
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
     class Element {

            private int value;

            public int get() {
                return value;
            }

            public void set(int value) {
                this.value = value;
            }

            public void clear() {
                value = 0;
            }
        }
...
        disruptor.handleEventsWith(handler1).then(((event, sequence, endOfBatch) -> event.clear()));

总结

disruptor是用来替换jdk 原本的队列
性能会高出很多
不过还是基于内存的 只能在单节点中做一些操作 替换原本使用jdk队列的地方

------ 本文结束 ------

版权声明
由ming创作并维护,博客采用CC协议。
本文首发于ming 博客（ https://blog.xujiuming.com ），版权所有，转载请注明出处！