二十. Disruptor队列原理简析

布满荆棘的人生 2024-03-30 17:38 37阅读 0赞

### 前言 ###

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

`Disruptor`版本：**3.4.0**

### 正文 ###

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

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

![在这里插入图片描述][8f7a69a8cb0d40ca9ea4ec38c6ac1465.jpeg_pic_center]

`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`。`Sequence`的使用者都是使用`Sequence`来维护自己的读/写序号。

#### 二. Disruptor对象的创建 ####

`Disruptor`对象的构造方法如下所示。

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));
    }

在创建`Disruptor`时会一并将`RingBuffer`创建出来，看一下`RingBuffer`的`create()`方法，如下所示。

public static <E> RingBuffer<E> create(
    		ProducerType producerType,
    		EventFactory<E> factory,
    		int bufferSize,
    		WaitStrategy waitStrategy) {
        
    	// 创建RingBuffer时会一并将Sequencer创建出来
    	// 如果指定的生产者类型为SINGLE，则创建SingleProducerSequencer
    	// 如果指定的生产者类型为MULTI，则创建MultiProducerSequencer
    	switch (producerType) {
        
    		case SINGLE:
    			return createSingleProducer(factory, bufferSize, waitStrategy);
    		case MULTI:
    			return createMultiProducer(factory, bufferSize, waitStrategy);
    		default:
    			throw new IllegalStateException(producerType.toString());
    	}
    }

本篇文章中的**producerType**为**SINGLE**，所以这里分析`createSingleProducer()`方法，如下所示。

public static <E> RingBuffer<E> createSingleProducer(
    		EventFactory<E> factory,
    		int bufferSize,
    		WaitStrategy waitStrategy) {
        
    	// 先创建SingleProducerSequencer
    	SingleProducerSequencer sequencer = new SingleProducerSequencer(bufferSize, waitStrategy);
    	// 然后基于SingleProducerSequencer创建RingBuffer
    	return new RingBuffer<E>(factory, sequencer);
    }

在`createSingleProducer()`方法中，会先创建`SingleProducerSequencer`，然后再基于这个`SingleProducerSequencer`创建`RingBuffer`，下面看一下`RingBuffer`的构造方法。

RingBuffer(
    		EventFactory<E> eventFactory,
    		Sequencer sequencer) {
        
    	super(eventFactory, sequencer);
    }

继续跟进`RingBuffer`父类`RingBufferFields`的构造方法，如下所示。

RingBufferFields(
    		EventFactory<E> eventFactory,
    		Sequencer sequencer) {
        
    	this.sequencer = sequencer;
    	this.bufferSize = sequencer.getBufferSize();
    
    	// 队列容量不能小于等于0
    	if (bufferSize < 1) {
        
    		throw new IllegalArgumentException("bufferSize must not be less than 1");
    	}
    	// 队列容量需要为2的幂次方，方便进行模运算
    	if (Integer.bitCount(bufferSize) != 1) {
        
    		throw new IllegalArgumentException("bufferSize must be a power of 2");
    	}
    
    	this.indexMask = bufferSize - 1;
    	// 初始化元素数组，空间多了BUFFER_PAD * 2
    	// 这是为了避免数组的首和尾的有效元素和其它无关数据加载到同一个缓存行从而出现伪共享
    	this.entries = new Object[sequencer.getBufferSize() + 2 * BUFFER_PAD];
    	// 预先将元素数组中的元素对象全部创建出来
    	fill(eventFactory);
    }
    
    private void fill(EventFactory<E> eventFactory) {
        
    	for (int i = 0; i < bufferSize; i++) {
        
    		entries[BUFFER_PAD + i] = eventFactory.newInstance();
    	}
    }

上述构造方法中，首先会对**bufferSize**进行校验，这个**bufferSize**实际就是元素数组的大小，因为`RingBuffer`是一个环形存储结构，存储的元素放在元素数组中，所以每添加一个元素时，这个元素在数组中的下标索引的取值是这个元素的序号对`bufferSize`取模（元素的序号由`Sequencer`中的`Sequence`记录，是一直增加的），那么这里的校验规则就是**bufferSize**需要是大于等于1且满足2的幂次方，之所以需要满足2的幂次方，就是因为对满足2的幂次方的**bufferSize**可以使用与上**bufferSize - 1**的方式来取模，这样的方式取模更快，这和`HashMap`中的取模方式是一致的。

其次，可以注意到存储元素的元素数组的实际大小为**sequencer.getBufferSize() + 2 \* BUFFER\_PAD**，这样做的原因是要在元素数组的首和尾额外分别创建一个缓存行大小的填充空间，这样就可以避免元素数组的有效元素与其它无关数据被加载到同一个缓存行从而出现伪共享的情况。

在上述构造方法的最后，还会将元素数组预热，即提前将所有有效元素对象创建出来，这些元素对象会一直存在以达到复用的效果，可以有效的解决频繁对元素对象**GC**的问题。

#### 三. 消费者消费逻辑 ####

本篇文章的示例中，在创建出`Disruptor`对象后，此时会调用`Disruptor`的`handleEventsWith()`方法来注册事件处理器，同时每个事件处理器会对应创建一个消费者，下面看一下`handleEventsWith()`方法的实现，如下所示。

public final EventHandlerGroup<T> handleEventsWith(final EventHandler<? super T>... handlers) {
        
    	return createEventProcessors(new Sequence[0], handlers);
    }
    
    EventHandlerGroup<T> createEventProcessors(
    		final Sequence[] barrierSequences,
    		final EventHandler<? super T>[] eventHandlers) {
        
    	checkNotStarted();
    
    	final Sequence[] processorSequences = new Sequence[eventHandlers.length];
    	// 通过RingBuffer创建SequenceBarrier
    	final SequenceBarrier barrier = ringBuffer.newBarrier(barrierSequences);
    
    	// 遍历每个事件处理器
    	for (int i = 0, eventHandlersLength = eventHandlers.length; i < eventHandlersLength; i++) {
        
    		final EventHandler<? super T> eventHandler = eventHandlers[i];
    
    		// 创建消费者，注意这里传入了SequenceBarrier
    		final BatchEventProcessor<T> batchEventProcessor =
    				new BatchEventProcessor<>(ringBuffer, barrier, eventHandler);
    
    		if (exceptionHandler != null) {
        
    			batchEventProcessor.setExceptionHandler(exceptionHandler);
    		}
    
    		// 将消费者相关对象添加到consumerRepository中
    		consumerRepository.add(batchEventProcessor, eventHandler, barrier);
    		// 每个消费者都持有一个Sequence对象
    		// 这里将消费者的Sequence对象添加到processorSequences数组中
    		processorSequences[i] = batchEventProcessor.getSequence();
    	}
    
    	updateGatingSequencesForNextInChain(barrierSequences, processorSequences);
    	
    	// 创建EventHandlerGroup并返回
    	return new EventHandlerGroup<>(this, consumerRepository, processorSequences);
    }

在上述的`handleEventsWith()`方法中会调用到`createEventProcessors()`方法，`createEventProcessors()`方法中做的第一件关键事情就是创建`SequenceBarrier`对象，这个`SequenceBarrier`会被这一批创建出来的消费者共同持有，用于消费者来判断当前是否可以消费元素数据。`createEventProcessors()`方法做的第二件事情就是遍历这一批的所有事件处理器（即传入的所有`EventHandler`），基于每个事件处理器都会创建一个消费者`BatchEventProcessor`对象，同时每个`BatchEventProcessor`对象的相关信息都会添加到`Disruptor`的**consumerRepository**字段中，**consumerRepository**字段是一个`ConsumerRepository`对象，其类图如下所示。

![在这里插入图片描述][4fca620c784b47f7a698bc7c7827b15c.jpeg_pic_center]

主要目的就是用于`Disruptor`持有消费者的引用。

上面分析完了消费者的创建，继续往下分析前，先看一下本示例中的消费者`BatchEventProcessor`的类图。

![在这里插入图片描述][f8aa78968b274ffe9eae230552a04afa.jpeg_pic_center]

首先应该知道`BatchEventProcessor`实现了`Runnable`接口，那么每个消费者都应该由`Disruptor`将其放入一个线程中运行起来，其次每个`BatchEventProcessor`都持有一个`SequenceBarrier`对象，一个`EventHandler`对象以及一个`Sequence`对象，这三个对象在消费者消费元素数据的过程中都会发挥重要作用，简要来说，`SequenceBarrier`对象会将消费者与生产者关联起来，`EventHandler`对象会实际的处理消费到的元素，`Sequence`会记录消费者消费到了哪里。

本篇文章的示例中，`Disruptor`创建完并且向`Disruptor`注册完`EventHandler`后，接下来就是调用`Disruptor`的`start()`方法将整个队列启动起来，下面看一下`Disruptor`的`start()`方法，如下所示。

public RingBuffer<T> start() {
        
    	checkOnlyStartedOnce();
    	// 遍历每一个消费者对应的ConsumerInfo，并调用其start()方法
    	for (final ConsumerInfo consumerInfo : consumerRepository) {
        
    		consumerInfo.start(executor);
    	}
    
    	return ringBuffer;
    }

前面提到过`Disruptor`通过一个`ConsumerRepository`对象持有消费者的相关信息，这里`Disruptor`通过`ConsumerRepository`遍历每一个消费者对应的`ConsumerInfo`并调用其`start()`方法，本示例中的`ConsumerInfo`实际类型为`EventProcessorInfo`，下面看一下其`start()`方法的实现，如下所示。

public void start(final Executor executor) {
        
    	executor.execute(eventprocessor);
    }

实际就是将每个消费者作为`Runnable`扔进线程池中，那么消费者的启动以及如何工作的逻辑，肯定就在消费者`BatchEventProcessor`的`run()`方法中，如下所示。

public void run() {
        
    	if (!running.compareAndSet(false, true)) {
        
    		throw new IllegalStateException("Thread is already running");
    	}
    	sequenceBarrier.clearAlert();
    
    	notifyStart();
    
    	T event = null;
    	// 消费者的初始序号是-1，而-1是消费不到数据的
    	// 所以消费者消费的第一个元素序号是初始序号+1，即序号0
    	long nextSequence = sequence.get() + 1L;
    	try {
        
    		while (true) {
        
    			try {
        
    				// 调用SequenceBarrier的waitFor()方法获取当前实际可以消费到的元素的最大序号
    				final long availableSequence = sequenceBarrier.waitFor(nextSequence);
    				if (batchStartAware != null) {
        
    					batchStartAware.onBatchStart(availableSequence - nextSequence + 1);
    				}
    
    				// 循环的消费元素，直到消费完序号为availableSequence的元素
    				while (nextSequence <= availableSequence) {
        
    					event = dataProvider.get(nextSequence);
    					// 每消费一个元素，就将该元素交由事件处理器来处理
    					eventHandler.onEvent(event, nextSequence, nextSequence == availableSequence);
    					nextSequence++;
    				}
    
    				// 更新当前消费者的消费进度，即将当前消费者的序号设置为最后一次消费的元素的序号
    				sequence.set(availableSequence);
    			} catch (final TimeoutException e) {
        
    				notifyTimeout(sequence.get());
    			} catch (final AlertException ex) {
        
    				if (!running.get()) {
        
    					break;
    				}
    			} catch (final Throwable ex) {
        
    				exceptionHandler.handleEventException(ex, nextSequence, event);
    				sequence.set(nextSequence);
    				nextSequence++;
    			}
    		}
    	} finally {
        
    		notifyShutdown();
    		running.set(false);
    	}
    }

单看`BatchEventProcessor`的`run()`方法的逻辑其实很简单，就是循环的获取当前可以消费的元素的最大序号，只要获取到可以消费的元素的最大序号，就依次消费直到消费到这个最大序号下的元素，每消费到一个元素就将这个元素传入当前消费者持有的事件处理器进行处理，也就是调用`EventHandler`的`onEvent()`方法来处理消费到的元素。那么问题就来了，因为`RingBuffer`是一个环形的结构，生产者生产数据填充元素的时候，如果`RingBuffer`满了，且不加以同步控制，那么会按照后添加的数据覆盖先添加的数据来处理，所以当消费者在获取到了一个可以消费的元素的最大序号后，就可能会出现生产者新生产的数据覆盖掉当前消费者还未消费的数据的情况，所以肯定是需要进行同步控制的，这个同步控制在消费者和生产者都有相应的实现，本小节主要分析消费者的同步控制，主要逻辑在`SequenceBarrier`的`waitFor()`方法中，要分析`SequenceBarrier`的`waitFor()`方法，需要先分析`SequenceBarrier`的创建，其创建的时机在上面已经有提及，就是在`Disruptor`的`createEventProcessors()`方法中，相关的代码片段如下所示。

// 本示例中的barrierSequences是一个空数组
    final SequenceBarrier barrier = ringBuffer.newBarrier(barrierSequences);

所以`SequenceBarrier`就是在注册事件监听器的时候，通过`RingBuffer`的`newBarrier()`方法创建出来的，下面看一下`newBarrier()`方法的实现，如下所示。

public SequenceBarrier newBarrier(Sequence... sequencesToTrack) {
        
    	return sequencer.newBarrier(sequencesToTrack);
    }

本篇文章中讨论的`Sequencer`实际均为`SingleProducerSequencer`，所以继续跟进`SingleProducerSequencer`的`newBarrier()`方法，如下所示。

public SequenceBarrier newBarrier(Sequence... sequencesToTrack) {
        
    	// 这里的waitStrategy在本示例中为YieldingWaitStrategy
    	// cursor为Sequencer持有的一个Sequence
    	return new ProcessingSequenceBarrier(this, waitStrategy, cursor, sequencesToTrack);
    }

上述方法中调用了`ProcessingSequenceBarrier`的构造方法，并且传入的**cursor**参数为`Sequencer`持有的用于记录生产者发布了的元素的序号的`Sequence`。继续跟进`ProcessingSequenceBarrier`的构造方法，如下所示。

ProcessingSequenceBarrier(
    		final Sequencer sequencer,
    		final WaitStrategy waitStrategy,
    		final Sequence cursorSequence,
    		final Sequence[] dependentSequences) {
        
    	// SequenceBarrier持有Sequencer的引用
    	this.sequencer = sequencer;
    	// SequenceBarrier持有一个等待策略，决定消费者在等待可消费元素时的策略
    	this.waitStrategy = waitStrategy;
    	// SequenceBarrier持有生产者最新发布元素的序号
    	this.cursorSequence = cursorSequence;
    	// 本示例中满足0 == dependentSequences.length
    	// 所以dependentSequence取为cursorSequence
    	if (0 == dependentSequences.length) {
        
    		dependentSequence = cursorSequence;
    	} else {
        
    		dependentSequence = new FixedSequenceGroup(dependentSequences);
    	}
    }

通过上面的构造方法可知，`ProcessingSequenceBarrier`持有`Sequencer`的引用，持有一个等待策略`WaitStrategy`，持有两个`Sequence`，在本篇文章示例中两个`Sequence`是完全相同的，均为生产者最新发布元素的序号，`ProcessingSequenceBarrier`的类图如下所示。

![在这里插入图片描述][6e3b6bafe0b047e0b4ebe908fc9d5f9d.jpeg_pic_center]

下面开始分析消费者获取当前实际可以消费到的元素的最大序号的实现，即`SequenceBarrier`也就是`ProcessingSequenceBarrier`的`waitFor()`方法，如下所示。

public long waitFor(final long sequence)
    		throws AlertException, InterruptedException, TimeoutException {
        
    	checkAlert();
    
    	// 调用等待策略的waitFor()方法得到一个可用序号
    	long availableSequence = waitStrategy.waitFor(sequence, cursorSequence, dependentSequence, this);
    
    	// 如果可用序号小于目标序号，则直接返回可用序号
    	// 说明当前最多只能消费到可用序号的元素
    	if (availableSequence < sequence) {
        
    		return availableSequence;
    	}
    	
    	// 如果可用序号大于等于目标序号，则调用Sequencer来得到当前最大的已发布的序号
    	// 本示例中这里的Sequencer实际为SingleProducerSequencer
    	// SingleProducerSequencer的策略就是直接返回可用序号
    	return sequencer.getHighestPublishedSequence(sequence, availableSequence);
    }
    
    // SingleProducerSequencer#getHighestPublishedSequence
    public long getHighestPublishedSequence(long lowerBound, long availableSequence) {
        
    	return availableSequence;
    }

在`ProcessingSequenceBarrier`的`waitFor()`方法中，先通过等待策略拿到一个可用序号**availableSequence**，然后判断可用序号与目标序号**sequence**（这里目标序号就是消费者想要消费到的序号）的大小关系，如果可用序号小于目标序号，表明当前消费者想要消费到的数据还没有被生产者生产（未发布），此时直接返回可用序号，如果可用序号大于等于目标序号，则调用`Sequencer`来得到当前最大的已发布的序号，本示例中的`Sequencer`实际为`SingleProducerSequencer`，而`SingleProducerSequencer`的策略就是将可用序号返回，即当出现当前生产者发布的元素已经多于消费者想要消费的元素时，允许消费者消费到最新发布的元素。

下面再继续跟进等待策略是如何获取可用序号的，本示例中的等待策略为`YieldingWaitStrategy`，`YieldingWaitStrategy`的`waitFor()`方法如下所示。

public long waitFor(
    		final long sequence, Sequence cursor, final Sequence dependentSequence, final SequenceBarrier barrier)
    		throws AlertException, InterruptedException {
        
    	long availableSequence;
    	// SPIN_TRIES = 100
    	int counter = SPIN_TRIES;
    
    	// 通过dependentSequence来拿到可用序号
    	// 这里的dependentSequence就是生产者最新发布的元素的序号
    	// 循环的获取生产者最新发布的元素序号直到最新发布的元素序号大于等于目标序号为止
    	// 前100次循环不放弃时间片，从第101次开始，每次循环需要放弃时间片
    	while ((availableSequence = dependentSequence.get()) < sequence) {
        
    		counter = applyWaitMethod(barrier, counter);
    	}
    
    	// 返回时可用序号一定是大于等于目标序号的，否则就会一直在上面循环
    	return availableSequence;
    }
    
    private int applyWaitMethod(final SequenceBarrier barrier, int counter)
    		throws AlertException {
        
    	barrier.checkAlert();
    
    	if (0 == counter) {
        
    		Thread.yield();
    	} else {
        
    		--counter;
    	}
    
    	return counter;
    }

`YieldingWaitStrategy`的`waitFor()`方法就是循环的判断生产者当前最新发布的元素序号是否大于等于消费者的目标序号，如果满足就返回这个最新发布的元素序号作为可用序号，如果不满足就一直循环的判断直到满足为止，并且前100次循环不放弃时间片，从第101次开始，每次循环都需要放弃时间片。

至此消费者的消费逻辑分析完毕。

#### 四. 生产者生产逻辑 ####

在本篇文章示例中，自定义了一个生产者叫做`TestEventProducer`，再贴出其实现如下。

public class TestEventProducer {
        
    
        private final RingBuffer<TestEvent> ringBuffer;
    
        public TestEventProducer(RingBuffer<TestEvent> ringBuffer) {
        
            this.ringBuffer = ringBuffer;
        }
    
        public void onData(String data) {
        
    		// 从RingBuffer中申请空间
    		// 元素数组sequence位置的元素是可用的
    		// 可用的意思就是可以生产数据到这个元素上并发布
            long sequence = ringBuffer.next();
            try {
        
    			// 从元素数组中将sequence位置的元素获取出来
                TestEvent testEvent = ringBuffer.get(sequence);
    			// 将生产的data数据设置到sequence位置的元素中
                testEvent.setId(data);
            } finally {
        
    			// 发布元素
    			// 也就是消费者可以消费了
    			// 不发布消费者就不能消费
                ringBuffer.publish(sequence);
            }
        }
    
    }

如上所示，`Disruptor`的生产者生产元素的步骤可以概括如下。

1.  从`RingBuffer`中申请空间，即获取到一个可以设置数据的元素的序号；
2.  根据步骤1中获取到的索引将可以设置数据的元素从数组中获取出来；
3.  为步骤2中获取出来的元素设置数据；
4.  发布步骤3中的元素，即这个元素允许被消费者消费了。

所以，关键步骤就是生产者如何申请空间，下面来分析一下`RingBuffer`的`next()`方法，如下所示。

public long next() {
        
    	return sequencer.next();
    }

可知`RingBuffer`的`next()`方法会调用到其持有的`Sequencer`的`next()`方法，本示例中这里的`Sequencer`为`SingleProducerSequencer`，其`next()`方法如下所示。

public long next() {
        
    	return next(1);
    }
    
    public long next(int n) {
        
    	if (n < 1) {
        
    		throw new IllegalArgumentException("n must be > 0");
    	}
    
    	// nextValue初始值是-1
    	// nextValue可以理解为最近一次写入的元素的序号
    	long nextValue = this.nextValue;
    
    	// nextSequence为本次申请的元素的序号
    	long nextSequence = nextValue + n;
    	
    	// wrapPoint表示生产者可能追尾消费最慢的消费者的点
    	long wrapPoint = nextSequence - bufferSize;
    	
    	// cachedGatingSequence表示上一次缓存的最慢消费者消费到的元素序号
    	// cachedValue初始值是-1
    	long cachedGatingSequence = this.cachedValue;
    
    	// 当wrapPoint大于cachedGatingSequence时，表示发生追尾
    	// 此时需要获取最新的最慢消费者的消费进度
    	if (wrapPoint > cachedGatingSequence || cachedGatingSequence > nextValue) {
        
    		// 调用UNSAFE的putLongVolatile()方法，对cursor插入StoreLoad屏障，让cursor的值对消费者可见
    		// 这样做的目的是让消费者及时见到cursor的值并消费发布的元素数据，因为下面会重新获取最慢消费者的消费进度
    		// 这里的cursor表示生产者最近一次发布的元素的序号
    		cursor.setVolatile(nextValue);
    
    		long minSequence;
    		// 先重新拿一次最慢消费者的消费进度，并判断是否还会追尾
    		// 如果还是会追尾，那么就睡眠1纳秒后再重复上面的判断步骤，直到不追尾为止
    		while (wrapPoint > (minSequence = Util.getMinimumSequence(gatingSequences, nextValue))) {
        
    			LockSupport.parkNanos(1L);
    		}
    
    		this.cachedValue = minSequence;
    	}
    	// 先将nextSequence更新给nextValue
    	this.nextValue = nextSequence;
    
    	// 返回nextSequence，生产者会生产数据填充到nextSequence对应的元素中
    	// 最后会发布元素，即将nextSequence赋值给cursor，所以cursor的更新晚于nextValue
    	return nextSequence;
    }

在上述`SingleProducerSequencer`的`next()`方法中，有五个重要变量，这里先给出其含义，如下所示。

1.  **nextValue**表示最近一次写入的元素的序号；
2.  **nextSequence**表示本次想要申请写入的元素的序号；
3.  **cursor**表示最近一次发布的元素的序号，发布了的元素才可以被消费者消费；
4.  **wrapPoint**表示生产者和消费者可能会追尾的点；
5.  **cachedGatingSequence**表示上一次缓存的最慢消费者消费的元素序号。

在`SingleProducerSequencer`的`next()`方法中，首先会将**nextValue**加**n**得到**nextSequence**，然后用**nextSequence**减去环形数组的大小**bufferSize**来得到可能会追尾的点**wrapPoint**，得到**wrapPoint**后，就会将**wrapPoint**与**cachedGatingSequence**比较大小，只要**wrapPoint**大于**cachedGatingSequence**，就会发生追尾，由于**cachedGatingSequence**是上一次缓存的最慢消费者的消费进度，那么此时就需要等待最慢消费者去消费元素从而缓存最新的最慢进度，直到不发生追尾。

注意到在等待最慢消费者去消费元素前，还执行了一个`cursor.setVolatile(nextValue)`的操作，这是因为**cursor**在更新的时候，调用的是**UNSAFE**的`putOrderedLong()`方法来更新**cursor**的值，而**UNSAFE**的`putOrderedLong()`方法插入的屏障类型是**StoreStore**屏障，该屏障不能保证消费者能及时看见**cursor**的最新值，所以需要在等待最慢消费者消费元素前先调用**UNSAFE**的`putLongVolatile()`方法，对**cursor**插入**StoreLoad**屏障，让**cursor**的值对消费者可见，确保消费者能尽快的消费到最新一次发布的元素。

最后，当成功申请到要写入元素的序号后，会先将**nextSequence**赋值给**nextValue**，但是实际记录生产者发布了的元素的**cursor**还没更新，而在我们自定义的生产者`TestEventProducer`的最后有这么一行代码。

ringBuffer.publish(sequence);

这里的**sequence**就是上面申请到的**nextSequence**，那么其实就是调用`RingBuffer`的`publish()`方法来发布元素，看一下其实现，如下所示。

public void publish(long sequence) {
        
    	// 更新cursor
    	cursor.set(sequence);
    	waitStrategy.signalAllWhenBlocking();
    }

至此，生产者的生产逻辑也分析完毕。

现在对`Disruptor`中的生产者和消费者进行一个简单小节。

*  对于**消费者**来说（这里仅针对`BatchEventProcessor`），消费者会循环的通过`SequenceBarrier`来拿到当前最大可以消费的序号并消费，如果拿不到则根据传入的等待策略**waitStrategy**进行等待，那实际上就是消费者会通过`SequenceBarrier`来拿到生产者当前已经发布的元素的序号**cursor**，从而得到最大可以消费的序号，所以消费者是通过`SequenceBarrier`来完成与生产者的关联。
 *  对于**生产者**来说，生产数据前需要先申请空间，也就是申请可用的元素的序号并通过可用元素的序号将可用元素获取出来，然后生产数据填充可用元素并发布这个元素，这是一个**两阶段提交**，第一阶段是申请可用元素序号，会基于所有消费者的**Sequence**（也就是消费者当前消费到了的元素的序号）得到最慢消费者的消费进度，从而判断申请的序号是否会导致发生追尾，如果会发生追尾，则等待最慢消费者消费，直到不发生追尾为止，如果不会发生追尾，那么就成功申请到可用元素序号，第二阶段是生产数据填充可用元素，然后发布这个序号，此时会更新生产者的**cursor**，此时消费者就能去消费新发布的元素。

最后以图例的方式，对消费者和生产者的一个工作模式进行说明。

**`消费者消费数据`**

![在这里插入图片描述][892cbeee111e4bc6ad02fda8b7170bdc.jpeg_pic_center]

如图，当前消费者消费到的元素的序号为9（**sequence**），那么本次消费的目标序号就是10（**nextSequence**），由于生产者已经发布的元素的序号为12（**cursor**），所以消费者本次能够一直消费到序号为12的元素。

**`生产者生产数据-不发生追尾`**

![在这里插入图片描述][6615f6fcf9d943e9bd3899e83f541dd1.jpeg_pic_center]

如图，生产者当前已经发布的元素的序号为13（**cursor**），那么本次申请的序号为14（**nextSequence**），由于环形数组大小为16（**bufferSize**），所以追尾点为-2（**wrapPoint**），同时最慢消费者的消费进度为2（**cachedGatingSequence**），所以不会发生追尾。

**`生产者生产数据-发生追尾`**

![在这里插入图片描述][ec15438a45d34e1f95c20e0b063bfe1a.jpeg_pic_center]

如图，生产者当前已经发布的元素序号为18（**cursor**），那么本次申请的序号为19（**nextSequence**），由于环形数组大小为16（**bufferSize**），所以追尾点为3（**wrapPoint**），同时最慢消费者的消费进度为2（**cachedGatingSequence**），所以会发生追尾（**wrapPoint > cachedGatingSequence**）。

### 总结 ###

本篇文章初步对`Disruptor`的工作原理进行了分析，并结合一个单生产者示例，深入源码对消费者如何消费数据，生产者如何生产数据以及并发控制思想进行了学习。

最后给出如下总结，回答`Disruptor`为什么快。

**1. 合理利用数据填充，避免了伪共享发生**  
以`Disruptor`中使用频次最高的组件`Sequence`进行举例说明。`Sequence`的实现如下所示。

class LhsPadding {
        
        protected long p1, p2, p3, p4, p5, p6, p7;
    }
    
    class Value extends LhsPadding {
        
        protected volatile long value;
    }
    
    class RhsPadding extends Value {
        
        protected long p9, p10, p11, p12, p13, p14, p15;
    }
    
    
    public class Sequence extends RhsPadding {
        
        ......
    }

可知`Sequence`本质就是对一个`volatile`的长整型变量**value**的一层包装，但是除了**value**之外，还分别在**value**的左右填充了8个长整型变量，因此**value**只会和从不更新的填充变量在一个缓存行上，避免了其它会更新的变量的更新操作导致**value**的缓存失效。这样的一个空间换时间的操作，极大提升了对缓存的利用，使得高频使用数据的读取更加的高效。

**2. 提前初始化元素对象并反复利用**  
`Disruptor`存储元素使用的是一个环形`Object`数组，在一开始就会将这个数组中的所有元素对象全部初始化出来，并且这些对象会反复利用，避免了元素对象的频繁创建和**GC**。

**3. 两阶段提交和CAS操作替代锁**  
无论是消费者消费数据，还是生产者生产数据，均没有使用重量级锁来进行并发控制，而是基于两阶段提交和**CAS**操作来实现了并发控制，减少了线程切换导致的性能开销。

[8f7a69a8cb0d40ca9ea4ec38c6ac1465.jpeg_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/072141732cb64e0dbc82590d57a2c789.jpeg
[4fca620c784b47f7a698bc7c7827b15c.jpeg_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/770e198e6a34417a83c9cc0f9ab380d6.jpeg
[f8aa78968b274ffe9eae230552a04afa.jpeg_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/5566b9e348d94c09ab0788a84432ce87.jpeg
[6e3b6bafe0b047e0b4ebe908fc9d5f9d.jpeg_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/22bd0ae9d31e4fdaae5c8a62bdad081d.jpeg
[892cbeee111e4bc6ad02fda8b7170bdc.jpeg_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/2cf1e5555c7c43c5893b85ff98bdd2eb.jpeg
[6615f6fcf9d943e9bd3899e83f541dd1.jpeg_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/54684ec9654542d9826cd87020f6692d.jpeg
[ec15438a45d34e1f95c20e0b063bfe1a.jpeg_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/9b2cd0e22816458182893369fdb62e17.jpeg