高性能无锁队列 Disruptor 外围原理剖析及其在i主题业务中的运行

2024-08-15 06:51:31 本文首先引见了 Disruptor 高性能内存队列的基本概念、经常使用 Demo、高性能原理及源码剖析，最后经过两个例子引见了 Disruptor 在i主题业务中的运行。

一、i主题及 Disruptor 简介

i主题是 vivo 旗下的一款主题商店 app，用户可以经过下载主题、壁纸、字体等，成功对手机界面格调的一键改换和自定义。

Disruptor 是英国外汇买卖公司 LMAX 开发的一个高性能的内存队列（用于系统外部线程间传递信息，不同于 RocketMQ、Kafka这种散布式信息队列），基于 Disruptor 开发的系统复线程能撑持每秒600万订单。目前，包括 ApacheStorm、Camel、Log4j 2在内的很多出名名目都运行了 Disruptor 以失掉高性能。在 vivo外部它也有不少运行，比如自定义监控中经常使用 Disruptor 队列来暂存经过监控 SDK上报的监控数据，i主题中也经常使用它来统计本地内存目的数据。

接上去从 Disruptor 和 JDK 内置队列的对比、Disruptor 外围概念、Disruptor 经常使用Demo、Disruptor外围源码、Disruptor 高性能原理、Disruptor 在 i主题业务中的运行几个角度来引见 Disruptor。

二、和 JDK 中内置的队列对比

上方来看下 JDK 中内置的队列和 Disruptor 的对比。队列的底层成功普通分为三种：数组、链表和堆，其中堆普通是为了成功带有优先级特性的队列，暂不思考。另外，像 ConcurrentLinkedQueue 、LinkedTransferQueue属于无界队列，在稳固性要求特意高的系统中，为了防止消费者速渡过快，造成内存溢出，只能选用有界队列。这样 JDK 中剩下可选的线程安保的队列还有ArrayBlockingQueue

由于 LinkedBlockingQueue 是基于链表成功的，由于链表存储的数据在内存里不延续，关于高速缓存并不友好，而且LinkedBlockingQueue 是加锁的，性能较差。ArrayBlockingQueue有雷同的疑问，它也须要加锁，另外，ArrayBlockingQueue 存在伪共享疑问，也会造成性能变差。而当天要引见的 Disruptor是基于数组的有界无锁队列，合乎空间部分性原理，可以很好的应用 CPU 的高速缓存，同时它防止了伪共享，大大优化了性能。

三、Disruptor 外围概念

如下图，从数据流转的角度先对 Disruptor 有一个直观的概念。Disruptor 支持单（多）消费者、单（多）消费者形式。消费时支持广播消费（HandlerA会消费处置一切信息，HandlerB 也会消费处置一切信息）、集群消费（HandlerA 和 HandlerB各消费部分信息），HandlerA 和HandlerB 消费成功后会把信息交给 HandlerC 继续处置。

上方联合 Disruptor 官网的架构图引见下 Disruptor 的外围概念：

四、Disruptor经常使用Demo

Event 是详细的数据实体，消费者消费 Event ，存入 RingBuffer，消费者从 RingBuffer 中消费它启动逻辑处置。Event 就是一个普通的 Java 对象，无需成功 Disruptor 内定义的接口。

public class OrderEvent {private long value;public long getValue() {return value;}public void setValue(long value) {this.value = value;}}

用于创立 Event 对象。

public class OrderEventFactory implements EventFactory<OrderEvent> {public OrderEvent newInstance() {return new OrderEvent();}}

可以看到，生成者关键是持有 RingBuffer 对象启动数据的颁布。这里有几个点须要留意：

public class OrderEventProducer {private RingBuffer<OrderEvent> ringBuffer;public OrderEventProducer(RingBuffer<OrderEvent> ringBuffer) {this.ringBuffer = ringBuffer;}public void sendData(ByteBuffer>

消费者可以成功 EventHandler 接口，定义自己的处置逻辑。

public class OrderEventHandler implements EventHandler<OrderEvent> {public void onEvent(OrderEvent event,long sequence,boolean endOfBatch) throws Exception {System.out.println("消费者: " + event.getValue());}}

public static void main(String[] args) {OrderEventFactory orderEventFactory = new OrderEventFactory();int ringBufferSize = 4;ExecutorService executor = Executors.newFixedThreadPool(1);/*** 1. 实例化disruptor对象1) eventFactory: 信息(event)工厂对象2) ringBufferSize: 容器的长度3) executor:4) ProducerType: 单消费者还是多消费者5) waitStrategy: 期待战略*/Disruptor<OrderEvent> disruptor = new Disruptor<OrderEvent>(orderEventFactory,ringBufferSize,executor,ProducerType.SINGLE,new BlockingWaitStrategy());// 2. 参与消费者的监听disruptor.handleEventsWith(new OrderEventHandler());// 3. 启动disruptordisruptor.start();// 4. 失掉实践存储数据的容器: RingBufferRingBuffer<OrderEvent> ringBuffer = disruptor.getRingBuffer();OrderEventProducer producer = new OrderEventProducer(ringBuffer);ByteBuffer bb = ByteBuffer.allocate(8);for (long i = 0; i < 5; i++) {bb.putLong(0, i);producer.sendData(bb);}disruptor.shutdown();executor.shutdown();}

五、Disruptor 源码剖析

本文剖析时以单（多）消费者、单消费者为例启动剖析。

首先是经过传入的参数创立 RingBuffer，将创立好的 RingBuffer 与传入的 executor 交给 Disruptor 对象持有。

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

接上去剖析 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());}}

不论是单消费者还是多消费者，最终都会创立一个 RingBuffer 对象，只是传给 RingBuffer 的 Sequencer 对象不同。可以看到，RingBuffer 外部最终创立了一个Object 数组来存储 Event 数据。这里有几点须要留意：

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

RingBufferFields(EventFactory<E> eventFactory,Sequencer sequencer){// 省略部分代码...// 额外创立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++){// BUFFER_PAD + i为真正的数组索引entries[BUFFER_PAD + i] = eventFactory.newInstance();}}

参与消费者的外围代码如下所示，外围就是为将一个

而后参与到consumerRepository 中，后续启动 Disruptor 时，会遍历 consumerRepository 中的一切 BatchEventProcessor（成功了Runnable 接口），将 BatchEventProcessor 义务提交到线程池中。

public final EventHandlerGroup<T> handleEventsWith(final EventHandler<? super T>... handlers){// 经过disruptor对象间接调用handleEventsWith方法时传的是空的Sequence数组return createEventProcessors(new Sequence[0], handlers);}

EventHandlerGroup<T> createEventProcessors(final Sequence[] barrierSequences,final EventHandler<? super T>[] eventHandlers) {// 搜集参与的消费者的序号final Sequence[] processorSequences = new Sequence[eventHandlers.length];// 本批次消费由于参与在同一个节点之后, 因此共享该屏障final SequenceBarrier barrier = ringBuffer.newBarrier(barrierSequences);// 为每个EventHandler创立一个BatchEventProcessorfor (int i = 0, eventHandlersLength = eventHandlers.length;i < eventHandlersLength; i++) {final EventHandler<? super T> eventHandler = eventHandlers[i];final BatchEventProcessor<T> batchEventProcessor =new BatchEventProcessor<>(ringBuffer, barrier, eventHandler);if (exceptionHandler != null){batchEventProcessor.setExceptionHandler(exceptionHandler);}// 参与到消费者信息仓库中consumerRepository.add(batchEventProcessor, eventHandler, barrier);processorSequences[i] = batchEventProcessor.getSequence();}// 降级网关序列(消费者只有要关注一切的末端消费者节点的序列)updateGatingSequencesForNextInChain(barrierSequences, processorSequences);return new EventHandlerGroup<>(this, consumerRepository, processorSequences);}

创立完 Disruptor 对象之后，可以经过 Disruptor 对象参与 EventHandler，这里有一须要留意：经过 Disruptor对象间接调用 handleEventsWith 方法时传的是空的 Sequence 数组，这是什么意思？可以看到createEventProcessors 方法接纳该空 Sequence 数组的字段名是barrierSequences，翻译成中文就是栅栏序号。怎样了解这个字段？

比如经过如下代码给 Disruptor 参与了两个handler，记为 handlerA 和 handlerB，这种是串行消费，关于一个Event，handlerA 消费完后才干轮到 handlerB 去消费。关于 handlerA来说，它没有前置消费者（生成者消费到哪里，消费者就可以消费到哪里），因此它的 barrierSequences 是一个空数组。而关于handlerB 来说，它的前置消费者是 handlerA，因此它的 barrierSequences 就是A的消费进展，也就是说handlerB 的消费进展是要小于 handlerA 的消费进展的。

disruptor.handleEventsWith(handlerA).handleEventsWith(handlerB);

假设是经过如下形式参与的 handler，则 handlerA 和handlerB 会消费一切 Event 数据，相似 MQ 信息中的广播消费，而 handlerC 的 barrierSequences 数组就是蕴含了 handlerA 的消费进展和 handlerB 的消费进展，这也是为什么barrierSequences 是一个数组，后续 handlerC在消费数据时，会取A和B消费进展的较小值启动判别，比如A消费到进展6，B消费到进展4，那么C只能去消费下标为3的数据，这也是barrierSequences 的含意。

disruptor.handleEventsWith(handlerA, handlerB).handleEventsWith(handlerC);

Disruptor的启动逻辑比拟繁复，就是遍历consumerRepository 中搜集的 EventProcessor（成功了Runnable接口），将它提交到创立 Disruptor 时指定的executor中，EventProcessor 的 run 方法会启动一个while 循环，不时尝试从 RingBuffer 中失掉数据启动消费。

disruptor.start();

public RingBuffer<T> start() {checkOnlyStartedOnce();for (final ConsumerInfo consumerInfo : consumerRepository) {consumerInfo.start(executor);}return ringBuffer;}public void start(final Executor executor) {executor.execute(eventprocessor);}

在剖析 Disruptor 的颁布数据的源码前，先来回忆下颁布数据的全体流程。

public void sendData(ByteBuffer>

next 方法自动放开一个序号。nextValue 示意已调配的序号，nextSequence 示意在此基础上再放开n个序号（此处n为1），cachedValue 示意缓存的消费者的最小消费进展。

假定有一个 size 为8的 RingBuffer，下标为6的数据曾经颁布好（nextValue为6），消费者不时未开启消费（cachedValue 和

cachedGatingSequence 为-1），此时消费者想继续颁布数据，调用 next() 方法放开失掉序号为7的位置(nextSequence为7)，计算失掉的 wrapPoint 为7-8=-1，此时 wrapPoint 等于

cachedGatingSequence，可以继续颁布数据，如左图。最后将 nextValue 赋值为7，示意序号7的位置曾经被消费者占用了。

接着消费者继续调用 next() 方法放开序号为0的数据，此时 nextValue为7，nextSequence 为8，wrapPoint 等于0， 由于消费者迟迟未消费

（cachedGatingSequence为-1），此时 wrapPoint 大于了 cachedGatingSequence，因此 next 方法的if判别成立，会调用LockSupport.parkNanos 阻塞期待消费者启动消费。其中 getMinimumSequence方法是失掉多个消费者的最小消费进展。

public long next() {return next(1);}

public long next(int n) {/*** 已调配的序号的缓存(已调配到这里), 初始-1. 可以看该方法的前往值nextSequence,* 接上去消费者就会该往该位置写数据, 它赋值给了nextValue, 所以下一次性调用next方* 法时, nextValue位置就是示意曾经消费好了数据, 接上去要放开nextSequece的数据*/long nextValue = this.nextValue;// 本次放开调配的序号long nextSequence = nextValue + n;// 造成环路的点：环形缓冲区或者追尾的点 = 等于本次放开的序号-环形缓冲区大小// 假设该序号大于最慢消费者的进展, 那么示意追尾了, 须要期待long wrapPoint = nextSequence - bufferSize;// 上次缓存的最小网关序号(消费最慢的消费者的进展)long cachedGatingSequence = this.cachedValue;// wrapPoint > cachedGatingSequence 示意消费者追上消费者发生环路(追尾), 即缓冲区已满,// 此时须要失掉消费者们最新的进展, 以确定能否队列满if (wrapPoint > cachedGatingSequence || cachedGatingSequence > nextValue) {// 拔出StoreLoad内存屏障/栅栏, 保障可见性。// 由于publish经常使用的是set()/putOrderedLong, 并不保障其余消费者能及时看见颁布的数据// 当我再次放开更多的空间时, 必定保障消费者能消费颁布的数据cursor.setVolatile(nextValue);long minSequence;// minSequence是多个消费者的最小序号, 要等一切消费者消费完了才干继续消费while (wrapPoint > (minSequence = Util.getMinimumSequence(gatingSequences,nextValue))) {LockSupport.parkNanos(1L);}// 缓存消费者们最新的消费进展this.cachedValue = minSequence;}// 这里只写了缓存, 并未写volatile变量, 由于只是预调配了空间然而并未被颁布数据,// 不须要让其余消费者感知到。消费者只会感知到真正被颁布的序号this.nextValue = nextSequence;return nextSequence;}

间接经过 Unsafe 工具类失掉指定序号的 Event 对象，此时失掉的是空对象，因此接上去须要对该 Event 对象启动业务赋值，赋值成功后调用 publish 方法启动最终的数据颁布。

OrderEvent event = ringBuffer.get(sequence);

public E get(long sequence) {return elementAt(sequence);}

protected final E elementAt(long sequence) {return (E) UNSAFE.getObject(entries,REF_ARRAY_BASE +((sequence & indexMask) << REF_ELEMENT_SHIFT));}

消费者失掉到可用序号后，首先对该序号处的空 Event 对象启动业务赋值，接着调用 RingBuffer 的 publish 方法颁布数据，RingBuffer 会委托给其持有的sequencer（单消费者和多消费者对应不同的 sequencer）对象启动真正颁布。单消费者的颁布逻辑比拟繁难，降级下 cursor进展（cursor 示意消费者的消费进展，该位置已实践颁布数据，而 next 方法中的 nextSequence示意消费者放开的最大序号，或者还未实践颁布数据），接着唤醒期待的消费者。

waitStrategy 有不同的成功，因此唤醒逻辑也不尽相反，如驳回 BusySpinWaitStrategy战略时，消费者失掉不到数据时自旋期待，而后继续判别能否有新数据可以消费了，因此 BusySpinWaitStrategy 战略的signalAllWhenBlocking 就是一个空成功，啥也不做。

ringBuffer.publish(sequence);

public void publish(long sequence) {sequencer.publish(sequence);}

public void publish(long sequence) {// 更重消费者进展cursor.set(sequence);// 唤醒期待的消费者waitStrategy.signalAllWhenBlocking();}

前面提到，Disruptor 启动时，会将封装 EventHandler 的EventProcessor（此处以 BatchEventProcessor为例）提交到线程池中运转，BatchEventProcessor 的 run 方法会调用 processEvents 方法不时尝试从RingBuffer 中失掉数据启动消费，上方剖析下 processEvents 的逻辑（代码做了精简）。它会开启一个 while 循环，调用sequenceBarrier.waitFor方法失掉最大可用的序号，比如失掉序号一节所提的，消费者继续消费，消费者不时未消费，此时消费者曾经将整个 RingBuffer数据都消费满了，消费者不可再继续消费，消费者此时会阻塞。假定这时刻消费者开局消费，因此 nextSequence 为0，而

availableSequence 为7，此时消费者可以批量消费，将这8条已消费者的数据所有消费完，消费成功后降级下消费进展。降级消费进展后，消费者经过Util.getMinimumSequence 方法就可以感知到最新的消费进展，从而不再阻塞，继续颁布数据了。

private void processEvents() {T event = null;// sequence记载消费者的消费进展, 初始为-1long nextSequence = sequence.get() + 1L;// 死循环，因此不会让出线程，须要独立的线程(每一个EventProcessor都须要独立的线程)while (true) {// 经过屏障失掉到的最大可用序号final long availableSequence = sequenceBarrier.waitFor(nextSequence);// 批量消费while (nextSequence <= availableSequence) {event =>

上方剖析下 SequenceBarrier 的 waitFor 方法。首先它会调用 waitStrategy 的 waitFor 方法失掉最大可用序号，以 BusySpinWaitStrategy 战略为例，它的 waitFor 方法的三个参数的含意区分是：

由于 dependentSequence 分为两种状况，所以 waitFor 的逻辑也可以分为两种状况探讨：

在 waitStrategy 的 waitFor 方法前往，失掉最大可用的序号 availableSequence 后，最后须要再调用下 sequencer 的

getHighestPublishedSequence失掉真正可用的最大序号，这和消费者模型有相关，假设是单消费者，由于数据是延续颁布的，间接前往传入的 availableSequence。而假设是多消费者，由于多消费者是有多个线程在消费数据，颁布的数据是不延续的，因此须要经过

getHighestPublishedSequence 方法失掉已颁布的且延续的最大序号，由于失掉序号启动消费时须要是顺序的，不能腾跃。

public long waitFor(final long sequence)throws AlertException, InterruptedException, TimeoutException {/*** sequence: 消费者希冀失掉的序号* cursorSequence: 消费者的序号* dependentSequence: 消费者须要依赖的序号*/long availableSequence = waitStrategy.waitFor(sequence,cursorSequence,dependentSequence, this);if (availableSequence < sequence) {return availableSequence;}// 目的sequence曾经颁布了, 这里失掉真正的最大序号(和消费者模型无关)return sequencer.getHighestPublishedSequence(sequence, availableSequence);}

public long waitFor(final long sequence, Sequence cursor, final Sequence dependentSequence,final SequenceBarrier barrier) throws AlertException, InterruptedException {long availableSequence;// 确保该序号曾经被我前面的消费者消费(协调与其余消费者的相关)while ((availableSequence = dependentSequence.get()) < sequence) {barrier.checkAlert();// 自旋期待ThreadHints.onSpinWait();}return availableSequence;}

六、Disruptor 高性能原理剖析

前文剖析源码时引见到，RingBuffer 外部保养了一个 Object 数组（也就是真正存储数据的容器），在 RingBuffer 初始化时该 Object数组就曾经经常使用EventFactory 初始化了一些空 Event，后续就不须要在运转时来创立了，防止频繁GC。

另外，RingBuffe的数组中的元素是在初始化时一次性性所有创立的，所以这些元素的内存地址大略率是延续的。消费者在消费时，是遵照空间部分性原理的。消费完第一个Event 时，很快就会消费第二个 Event，而在消费第一个 Event 时，CPU 会把内存中的第一个 Event 的前面的 Event 也加载进Cache 中，这样当消费第二个 Event 时，它曾经在 CPU Cache 中了，所以就不须要从内存中加载了，这样也可以大大优化性能。

如下代码所示，定义了一个 Pointer 类，它有2个 long 类型的成员变量x、y，而后在 main 方法中其中2个线程区分对同一个 Pointer对象的x和y自增 100000000 次，最后统计下方法耗时，在我本机电脑上测试屡次，平均约为3600ms。

public class Pointer {volatile long x;volatile long y;@Overridepublic String toString() {return new StringJoiner(", ", Pointer.class.getSimpleName() + "[", "]").add("x=" + x).add("y=" + y).toString();}}

public static void main(String[] args) throws InterruptedException {Pointer pointer = new Pointer();int num = 100000000;long start = System.currentTimeMillis();Thread t1 = new Thread(() -> {for(int i = 0; i < num; i++){pointer.x++;}});Thread t2 = new Thread(() -> {for(int i = 0; i < num; i++){pointer.y++;}});t1.start();t2.start();t1.join();t2.join();System.out.println(System.currentTimeMillis() - start);System.out.println(pointer);}

接着将 Pointer 类修正如下：在变量x和y之间拔出7个 long 类型的变量，仅此而已，接着继续经过上述的 main方法统计耗时，平均约为500ms。可以看到，修正前的耗时是修正后（防止了伪共享）的7倍多。那么什么是伪共享，为什么防止了伪共享能有这么大的性能优化呢？

public class Pointer {volatile long x;long p1, p2, p3, p4, p5, p6, p7;volatile long y;@Overridepublic String toString() {return new StringJoiner(", ", Pointer.class.getSimpleName() + "[", "]").add("x=" + x).add("y=" + y).toString();}}

6.2.2、防止伪共享为什么可以优化性能

内存的访问速度是远远慢于 CPU 的，为了高效应用 CPU，在 CPU 和内存之间加了缓存，称为 CPU Cache。为了提高性能，须要更多地从 CPU Cache里失掉数据，而不是从内存中失掉数据。CPU Cache 加载内存里的数据，是以缓存行（通常为64字节）为单位加载的。Java 的 long类型是8字节，因此一个缓存行可以寄存8个 long 类型的变量。

然而，这种加载带来了一个坏处，如上述例子所示，假定有一个 long 类型的变量x，另外还有一个 long 类型的变量y紧挨着它，那么当加载x时也会加载y。假设此时 CPU Core1的线程在对x启动修正，另一个 CPU Core2 的线程却在对y启动读取。者修正x时，会把x和y同时加载到 CPU Core1 对应的CPU Cache 中，降级完后x和其它一切蕴含x的缓存行都将失效。而当 CPU Core2的线程读取y时，发现这个缓存行曾经失效了，须要从主内存中从新加载。

这就是伪共享，x和y不相干，然而却由于x的降级造成须要从新从主内存读取，拖慢了程序性能。处置方法之一就是如上述示例中所做，在x和y之间填充7个 long 类型的变量，保障x和y不会被加载到同一个缓存行中去。Java8中也参与了新的注解@Contended（JVM加上启动参数-XX:-RestrictContended 才会失效），也可以防止伪共享。

Disruptor 中经常使用 Sequence 类的 value 字段来示意消费/消费进展，可以看到在该字段前后各填充了7个 long 类型的变量，来防止伪共享。另外，向 RingBuffer 外部的数组、

SingleProducerSequencer 等也经常使用了该技术。

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

消费者消费数据时，须要入队。消费者消费数据时，须要出队。入队时，不能笼罩没有消费的元素。出队时，不能读取没有写入的元素。因此，Disruptor 中须要保养一个入队索引（消费者数据消费到哪里，对应 AbstractSequencer 中的 cursor）和一个出队索引（一切消费者中消费进展最小的序号）。

Disruptor 中最复杂的是入队操作，上方以多消费者（MultiProducerSequencer）的 next(n) 方法（放开n个序号）为例剖析下Disruptor 是如何成功无锁操作的。代码如下所示，判别下能否有足够的序号（空余位置），假设没有，就让出 CPU经常使用权，而后从新判别。假设有，则经常使用 CAS 设置 cursor（入队索引）。

public long next(int n) {do {// cursor相似于入队索引, 指的是上次消费到这里current = cursor.get();// 目的是再消费n个next = current + n;// 前文剖析过, 用于判别消费者能否曾经追上消费进展, 消费者能否放开到n个序号long wrapPoint = next - bufferSize;// 失掉缓存的上一次性的消费进展long cachedGatingSequence = gatingSequenceCache.get();// 第一步：空间无余就继续期待if (wrapPoint > cachedGatingSequence || cachedGatingSequence > current) {// 从新计算下一切消费者里的最小消费进展long gatingSequence = Util.getMinimumSequence(gatingSequences, current);// 依然没有足够的空间, 让出CPU经常使用权if (wrapPoint > gatingSequence) {LockSupport.parkNanos(1);continue;}// 降级下最新的最小的消费进展gatingSequenceCache.set(gatingSequence);}// 第二步：看见空间足够时尝试CAS竞争空间else if (cursor.compareAndSet(current, next)) {break;}} while (true);return next;}

这个比拟好了解，在前文剖析消费数据的逻辑时引见了，消费者会失掉下最大可用的序号，而后批量消费这些信息。

七、Disruptor 在i主题业务中的经常使用

很多开源名目都经常使用了 Disruptor，比如日志框架 Log4j2 经常使用它来成功异步日志。HBase、Storm 等名目中也经常使用了到了 Disruptor。vivo 的 i主题业务也经常使用了 Disruptor，上方繁难引见下它的2个经常使用场景。

业务监控系统关于企业来说十分关键，可以协助企业及时发现和处置疑问，可以繁难的检测业务目的数据，改良业务决策，从而保障业务的可继续开展。i主题经常使用 Disruptor（多消费者单消费者）来暂存待上报的业务目的数据，而后有定时义务不时提取数据上报到监控平台，如下图所示。

i主题业务中少量经常使用了本地缓存，为了统计本地缓存中key 的个数（去重）以及每种缓存形式 key 的数量，思考经常使用 Disruptor来暂存并消费处置数据。由于业务代码里很多中央触及到本地缓存的访问，也就是说，消费者是多线程的。思考到消费处置比拟繁难，而假设经常使用多线程消费的话又触及到加锁同步，因此消费者驳回复线程形式。

全体流程如下图所示，首先在缓存访问工具类中参与缓存访问统计上报的调用，缓存访问数据进入到 RingBuffer 后，复线程消费者经常使用 HyperLogLog 来去重统计不同key的个数，经常使用正则婚配来统计每种形式key的数量。而后有异步义务定时失掉统计结果，启动展现。

须要留意的是，由于 RingBuffer 队列大小是固定的，假设消费者消费过快而消费者消费不上来，假设经常使用 next方法放开序号，假设残余空间不够会造成消费者阻塞，因此倡导经常使用 tryPublishEvent 方法去颁布数据，它外部是经常使用 tryNext方法放开序号，该方法假设放开不到可用序号会抛出意外，这样消费者感知到了就可以做兼容处置，而不是阻塞期待。

八、经常使用倡导

九、总结

本文首先经过对比 JDK 中内置的线程安保的队列和Disruptor 的特点，引入了高性能无锁内存队列 Disruptor。接着引见了 Disruptor的外围概念和基本经常使用，使读者对 Disruptor 建设后来步的意识。接着从源码和原理角度引见了 Disruptor的外围成功以及高性能原理（空间预调配、防止伪共享、无锁、支持批量消费）。其次，联合i主题业务引见了 Disruptor在通常中的运行。最后，基于上述原理剖析及运行实战，总结了一些 Disruptor 最佳通常战略。

参考文章：