disruptor的使用和分析

心已赠人 2024-02-17 17:16 46阅读 0赞

disruptor：3.3.4  
jdk：1.8

##### 介绍： #####

disruptor是典型的生产者和消费者模式，disruptor使用RingBuffer存放数据，sequence管理生产和消费的位置（long类型，递增），生产者要生产数据的时候首先向生产者请求一个sequence，然后在sequence的位置放置数据。

##### 入门： #####

生产的数据模型

public class LogEvent {
    
        private long value;
    
        public void setValue(long value) {
            this.value = value;
        }
    
        public String toString() {
            return this.value + "";
        }
    }

消费者消费数据

public class EventHandler {
    
        public void handleEvent(LogEvent event, long sequence, boolean falg) throws Exception {
            System.out.println("handle event " + event + " cur thread " + Thread.currentThread().getId());
        }
    }

初始化：

void sigleProducerSingleConsume() throws Exception {
            int bufferSize = 8; // RingBuffer的大小，2的n次方
            ThreadFactory threadFactory = Executors.defaultThreadFactory();
            EventHandler handler = new EventHandler();
            Disruptor<LogEvent> disruptor = new Disruptor<>(LogEvent::new, bufferSize, threadFactory, ProducerType.SINGLE, new BlockingWaitStrategy());
            disruptor.handleEventsWith(handler::handleEvent);
            RingBuffer<LogEvent> buffer = disruptor.start();
            for(int i=0; i<10; i++) {
                long sequence = buffer.next(); // 申请位置
                try {
                    LogEvent event = buffer.get(sequence);
                    event.setValue(i); // 放置数据
                } finally {
                    buffer.publish(sequence); // 提交，如果不提交完成事件会一直阻塞
                }
            }
            disruptor.shutdown();
        }

##### 原理解析： #####

###### 多个生产者模式： ######

Disruptor<LogEvent> disruptor = new Disruptor<>(LogEvent::new, bufferSize, threadFactory, ProducerType.MULTI, new BlockingWaitStrategy());

初始化方法：

public Disruptor(
                final EventFactory<T> eventFactory,
                final int ringBufferSize,
                final ThreadFactory threadFactory,
                final ProducerType producerType,
                final WaitStrategy waitStrategy)
        {
            this(RingBuffer.create(
                                   producerType, eventFactory, ringBufferSize, waitStrategy),
                    new BasicExecutor(threadFactory));
        }

创建RingBuffer：

public static <E> RingBuffer<E> create(
            ProducerType producerType,
            EventFactory<E> factory,
            int bufferSize,
            WaitStrategy waitStrategy)
        {
            switch (producerType)
            {
                case SINGLE:
                    return createSingleProducer(factory, bufferSize, waitStrategy);
                case MULTI:
                    return createMultiProducer(factory, bufferSize, waitStrategy);
                default:
                    throw new IllegalStateException(producerType.toString());
            }
        }

public static <E> RingBuffer<E> createMultiProducer(
            EventFactory<E> factory,
            int bufferSize,
            WaitStrategy waitStrategy)
        {
            MultiProducerSequencer sequencer = new MultiProducerSequencer(bufferSize, waitStrategy);
    
            return new RingBuffer<E>(factory, sequencer);
        }

这里可以看到，使用MultiProducerSequencer 创建了RingBuffer，当我们使用单个生产者模式的时候可以看到使用SingleProducerSequencer创建了RingBuffer，所以，disruptor是使用一个sequence来维护生产者用到的index 的。

让我们看看MultiProducerSequencer 的代码：

public MultiProducerSequencer(int bufferSize, final WaitStrategy waitStrategy)
        {
            super(bufferSize, waitStrategy);
            availableBuffer = new int[bufferSize];
            indexMask = bufferSize - 1;
            indexShift = Util.log2(bufferSize);
            initialiseAvailableBuffer();
        }

要注意availableBuffer ，这个数组中保存了生产者在每个位置添加数据的信息。  
indexShift ：bufferSize >> indexShift + 1 时为0  
initialiseAvailableBuffer()初始化availableBuffer 所有元素为-1

然后让我们看添加数据的处理：

long sequence = buffer.next(); // 申请位置
    try {
        LogEvent event = buffer.get(sequence);
        event.setValue(i); // 放置数据
    } finally {
        buffer.publish(sequence); // 提交，如果不提交完成事件会一直阻塞
    }

取得可用的位置

@Override
        public long next()
        {
            return next(1);
        }
    
        /**
         * @see Sequencer#next(int)
         */
        @Override
        public long next(int n)
        {
            if (n < 1)
            {
                throw new IllegalArgumentException("n must be > 0");
            }
    
            long current;
            long next;
    
            do
            {
                current = cursor.get(); // 
                next = current + n; // 获取下一个可用位置，最好不要用next(n)来获取位置，很容易产生死循环
    
                long wrapPoint = next - bufferSize; // 
                long cachedGatingSequence = gatingSequenceCache.get(); // 保存了消费者处理过的最小位置
            // 保证生产者没有超过消费者一圈。
                if (wrapPoint > cachedGatingSequence || cachedGatingSequence > current)
                {
                    long gatingSequence = Util.getMinimumSequence(gatingSequences, current);
    
                    if (wrapPoint > gatingSequence)
                    {
                        LockSupport.parkNanos(1); // TODO, should we spin based on the wait strategy?
                        continue;
                    }
    
                    gatingSequenceCache.set(gatingSequence);
                }
                else if (cursor.compareAndSet(current, next))
                {
                    break;
                }
            }
            while (true);
    
            return next;
        }

中心思想就是获取当前可用的位置，这里没有用到锁，而是使用了cas。  
gatingSequence中保存了各个消费者处理过的当前位置。

这样当前生产者就可以使用这个位置了，最后要记得publish，告知消费者去读取数据。

@Override
        public void publish(final long sequence)
        {
            setAvailable(sequence); // 
            waitStrategy.signalAllWhenBlocking(); // 消费者可以读了
        }

setAvailable是一个很巧妙的设计，如果是多个生产者，availableBuffer中的值是这样变化的：  
\[-1, -1, -1, -1\]  
\[0, -1, -1, -1\]  
\[0, 0, 0, 0\]  
至此，四个位置已经被写入数据了  
\[1, 0, 0, 0\]  
\[1, 1, 0, 0\]  
\[1, 1, 0, 1\]  
\[1, 1, 1, 1\]  
看第三种情形，producer1占据了第三个位置，但是producer2先一步在第四个位置写入了数据，consumer是可以跳过第三个位置读取第四个位置的数据的。但是这种情况会极少出现，第三个位置没有没consumer读取的话，producer2是不能拿到第四个位置的。

private void setAvailable(final long sequence)
        {
            setAvailableBufferValue(calculateIndex(sequence), calculateAvailabilityFlag(sequence));
        }

private int calculateAvailabilityFlag(final long sequence)
        {
            return (int) (sequence >>> indexShift);
        }
    
        private int calculateIndex(final long sequence)
        {
            return ((int) sequence) & indexMask;
        }

private void setAvailableBufferValue(int index, int flag)
        {
            long bufferAddress = (index * SCALE) + BASE;
            UNSAFE.putOrderedInt(availableBuffer, bufferAddress, flag);
        }