深度详解高性能队列：Disruptor

女爷i 2024-03-26 22:00 60阅读 0赞

### 前言 ###

并发框架**Disruptor**是一个高性能队列，其凭借**无锁**，**消除伪共享**等策略极大提升了队列性能，本篇文章将基于示例和源码，对**Disruptor**高性能队列的使用和原理进行学习。

**Disruptor**版本：**3.4.0**

### 正文 ###

#### 一. Disruptor结构分析和组件介绍 ####

**Disruptor**中的核心组件是**RingBuffer**，基于**RingBuffer**的生产者消费者模型，如下所示。

![format_png][]

**RingBuffer**中有一个**entries**字段，是一个**Object**数组结构，**RingBuffer**使用**entries**来存储元素，队列工作过程就是生产者将数据写入到**RingBuffer**的元素中，消费者从**RingBuffer**获取元素中的数据。下面给出一种最简单的使用场景，即单生产者场景。

**Disruptor**没有对队列中的元素类型做定义，需要使用者自行定义元素类型，本示例中将队列的元素定义为**TestEvent**，如下所示。

public class TestEvent {
    
        private String id;
    
        public String getId() {
            return id;
        }
    
        public void setId(String id) {
            this.id = id;
        }
    
    }
    复制代码

然后定义**TestEventHandler**实现**EventHandler**接口，每个**TestEventHandler**都需要注册到**Disruptor**中，**Disruptor**会基于每个注册的**TestEventHandler**来创建**BatchEventProcessor**作为消费者，每个**BatchEventProcessor**消费到元素后会将元素交给其持有的**TestEventHandler**来处理。**TestEventHandler**实现如下所示。

public class TestEventHandler implements EventHandler<TestEvent> {
    
        private final String consumerId;
    
        public TestEventHandler(String consumerId) {
            this.consumerId = consumerId;
        }
    
        @Override
        public void onEvent(TestEvent testEvent, long sequence, boolean endOfBatch) {
            System.out.println("Consumer-EventHandler-" + this.consumerId + " consumed message: " + testEvent.getId());
        }
    
    }
    复制代码

再然后定义**TestEventFactory**实现**EventFactory**接口，用于帮助**Disruptor**在初始化**RingBuffer**时一次性将元素全部创建出来并填充满元素数组。**TestEventFactory**实现如下所示。

public class TestEventFactory implements EventFactory<TestEvent> {
    
        @Override
        public TestEvent newInstance() {
            return new TestEvent();
        }
    
    }
    复制代码

最后是生产者，同样的，**Disruptor**没有对生产者做定义，本示例中自行封装的生产者如下所示。

public class TestEventProducer {
    
        private final RingBuffer<TestEvent> ringBuffer;
    
        public TestEventProducer(RingBuffer<TestEvent> ringBuffer) {
            this.ringBuffer = ringBuffer;
        }
    
        public void onData(String data) {
            long sequence = ringBuffer.next();
            try {
                TestEvent testEvent = ringBuffer.get(sequence);
                testEvent.setId(data);
            } finally {
                ringBuffer.publish(sequence);
            }
        }
    
    }
    复制代码

下面是一个测试程序，结合上述定义好的类，对生产和消费的简单工作流程进行演示。

public class MyTest {
    
        private static final String CONSUMER_ID_1 = "1";
        private static final String CONSUMER_ID_2 = "2";
        private static final String CONSUMER_ID_3 = "3";
    
        private static final int RING_BUFFER_SIZE = 1024 * 1024;
        private static final int WAIT_MS = 1000;
        private static final int BATCH_NUM = 3;
    
        public static void main(String[] args) throws Exception {
            TestEventFactory factory = new TestEventFactory();
            // 创建Disruptor，指定生产者类型为单生产者
            Disruptor<TestEvent> disruptor = new Disruptor<>(factory, RING_BUFFER_SIZE,
                    Executors.defaultThreadFactory(), ProducerType.SINGLE, new YieldingWaitStrategy());
            // 向Disruptor注册TestEventHandler，即注册消费者
            disruptor.handleEventsWith(new TestEventHandler(CONSUMER_ID_1),
                    new TestEventHandler(CONSUMER_ID_2), new TestEventHandler(CONSUMER_ID_3));
            // 开启Disruptor，基于线程池将消费者运行起来
            disruptor.start();
            Thread.sleep(WAIT_MS);
            // 创建生产者，并将数据写入RingBuffer的元素中
            RingBuffer<TestEvent> ringBuffer = disruptor.getRingBuffer();
            TestEventProducer testEventProducer = new TestEventProducer(ringBuffer);
            for (int i = 0; i < BATCH_NUM; i++) {
                testEventProducer.onData(String.valueOf(i));
            }
            Thread.sleep(WAIT_MS);
            // 关闭Disruptor
            disruptor.shutdown();
        }
    
    }
    复制代码

结合上述示例，在**Disruptor**框架中有如下几个关键角色。

*  **`Disruptor`对象**。可以理解为**Disruptor**框架中的锚点，其持有一个**RingBuffer**对象，一个线程池**Executor**对象以及一个**ConsumerRepository**对象，生产者生产的数据会存放在**RingBuffer**中的元素中，同时当向**Disruptor**注册事件处理器时**Disruptor**会基于注册的事件处理器创建消费者并添加到**ConsumerRepository**中；
 *  **`RingBuffer`对象**。**Disruptor**框架中的核心对象，其持有一个**Object**数组用于存放元素以及一个**Sequencer**对象实现对生产者的同步控制；
 *  **`Sequencer`对象**。其实际是一个接口，有两个实现类分别为**SingleProducerSequencer**和**MultiProducerSequencer**，代表对单生产者和多生产者的同步控制（可以这么理解，上面示例中定义的生产者**TestEventProducer**如果向**RingBuffer**生产元素，那么就会和其他生产者以及消费者产生并发冲突，**Sequencer**就是用于控制并解决这个并发冲突的）；
 *  **`SequenceBarrier`对象**。其由**Sequencer**创建，并且会由消费者持有，主要用于消费者获取当前可以消费的元素的序号；
 *  **`EventProcessor`对象**。其实际是一个接口，表示消费者，当向**Disruptor**对象注册**EventHandler**对象时，**Disruptor**会基于**EventHandler**创建一个**BatchEventProcessor**对象作为消费者，当向**Disruptor**注册**WorkHandler**对象时，**Disruptor**会基于**WorkHandler**创建一个**WorkProcessor**对象作为消费者，本篇文章提及的消费者全部指**BatchEventProcessor**；
 *  **`Sequence`对象**。每个**EventProcessor**消费者会持有一个**Sequence**，同时**SingleProducerSequencer**持有一个**Sequence**，**MultiProducerSequencer**持有两个**Sequence**

[format_png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/3813a2053b6e4b56ad9929106b94f949.png