360doc文章提取器

Disruptor进阶

文章来源:文章来源 抓取时间:2019-01-28 16:12:49 浏览量:15 作者:web127 返回文章列表

Intruduction

Disruptor是Java实现的用于线程间通信的消息组件。其核心是一个Lock-free的Ringbuffer。我使用BlockingQueue与进Disruptor行了简单的对比测试,结果表明使用Disruptor来进行线程间通信效率会提高将近一倍。而LMAX给出的数据是,使用Disruptor能够在一个线程里每秒处理6百万订单。Disruptor为什么会如此快呢?通过参考Martin Fowler(Disruptor的开发者之一)的技术博客和Disruptor的源代码,可以总结出以下四条原因:

Lock vs CAS

Disruptor使用CAS而不是Lock。关于CAS(compare and swap)请参考WIKIPEDIA相关条目Compare and swap。与大部分并发队列使用的Lock相比,CAS显然要快很多。CAS是CPU级别的指令,更加轻量,不需要像Lock一样需要OS的支持,所以每次调用不需要kernel entry,也不需要context switch。当然,使用CAS的代价是Disruptor实现的复杂程度也相对提高了。

避免伪共享

现代计算机体系架构在各个层次都使用的Cache来提高效率,然而在多核体系结构中,对Cache的不恰当使用极易造成伪共享,使性能下降。关于伪共享请参考WIKIPEDIA相关条目False sharing。为了避免伪共享带来的性能下降,Disruptor对一切可能存在伪共享的地方使用Padding将两个不想关的内存隔离到两个缓存行上。可能存在伪共享的地方包括两个不相关的线程共享变量之间,线程私有变量和线程共享变量之间等。下面分别举例子说明。

在Disruptor的实现中,有一个多线程共享的计数组件Sequence,对Sequence的操作可以说是整个Disruptor的核心,关于Sequence,在下文介绍各个组件的时候还要详细说明。这里主要说明它是怎样避免伪共享的。主要代码如下:

static final long INITIAL_VALUE = -1L;  private static final Unsafe UNSAFE;  private static final long VALUE_OFFSET;    static  {      UNSAFE = Util.getUnsafe();      final int base = UNSAFE.arrayBaseOffset(long[].class);      final int scale = UNSAFE.arrayIndexScale(long[].class);      VALUE_OFFSET = base + (scale * 7);  }    private final long[] paddedValue = new long[15];    . . .   . . .    public long get()  {      return UNSAFE.getLongVolatile(paddedValue, VALUE_OFFSET);  }  

Sequence定义了一个长度为15的long类型数组,但仅使用数组第八个元素进行计数,数组其他部分连同对象的头作为padding部分,保证在以64byte作为Cache缓存行大小的CPU中,计数元素(数组第八个元素)不会与其他变量存在于同一个缓存行中。关于Java中对象在内存中具体怎样布局,可以参考深入理解Java虚拟机:JVM高级特性与最佳实践

另一个例子是关于线程私有变量的:

private static class Padding  {      /** Set to -1 as sequence starting point */      public long nextValue = Sequence.INITIAL_VALUE, cachedValue = Sequence.INITIAL_VALUE, p2, p3, p4, p5, p6, p7;  }    private final Padding pad = new Padding();  

这段代码用在单生产者的应用场景中。在这种应用场景下,这个计数器不需要是线程安全的,使用Sequence过于heavy了,但仍然需要通过padding将其与其他线程共享的变量隔离开来。变量p2到p7以及对象头的作用就是这个。

Linked Queue vs Array Ringbuffer

Disruptor选择使用Array Ringbuffer来构造lock-free队列,而不是选择Linked Queue。

首先,数组是预分配的,这样不仅避免了Java GC带来的运行开销,而且因为在ringbuffer上进行的操作是顺序执行的,所以对缓存来说更加友好,保证了缓存命中率。使用Disruptor的时候,为了更好的利用Ringbuffer的这个优点,需要尽量将Ringbuffer的元素设计的可重用,生产者在生产消息或产生事件的时候对Ringbuffer元素中的属性进行更新,而不是替换Ringbuffer中的元素。

其次,数组在定位元素的时候是使用索引,而链表在定位元素的时候使用对象引用(地址)。在lock-free队列中使用链表需要考虑使用Double-CAS等方式来克服ABA问题(关于double-CAS和ABA问题,请参coolshell上关于无锁队列的文章),而在数组中,因为元素是预分配的,所以不存在ABA问题。Disruptor使用递增的Sequence来标示不同时刻访问的相同元素,比如一个消费者的Sequence等于i的时候表示在访问Ringbuffer的某个位置的元素,在下一次访问这个位置的元素的时候,Seqence等于i + buffer_size。在需要访问数组元素的时候,只需要将序号对数组大小取余就可以得到数组索引。每次对ringbuffer的访问都会导致相应的Sequence增加。需要注意的是,由于Sequence是递增的,所以在到达最大值以后,会溢出,变成最小的负数,但这通常不是问题,因为要使long类型递增到溢出,即使每秒钟1000 000 000次递增,也需要上百年时间。

无时不刻的缓存

为了高效,Disruptor可谓无所不用其极,它绝不会放过任何利用缓存的机会,看一个例子。

public long next(int n)  {      if (n < 1)      {          throw new IllegalArgumentException("n must be > 0");      }        long nextValue = pad.nextValue;        long nextSequence = nextValue + n;      long wrapPoint = nextSequence - bufferSize;      long cachedGatingSequence = pad.
    下载DOC文档