老余博客上线了!!!

Disruptor 详解 二

热点新闻 老余 79℃ 0评论

Disruptor 的大名从很久以前就听说了,但是一直没有时间;看完以后才发现其内部的思想异常清晰,很容易就能前移到其他的项目,所以仔细了解一下还是很有必要的这。篇博客将主要从源码角度分析,Disruptor 为什么那么快,在此之前可以先查看 Disruptor 详解 一 ,能够...

Disruptor 的大名从很久以前就听说了,但是一直没有时间;看完以后才发现其内部的思想异常清晰,很容易就能前移到其他的项目,所以仔细了解一下还是很有必要的这。篇博客将主要从源码角度分析,Disruptor 为什么那么快,在此之前可以先查看 Disruptor 详解 一 ,能够对 Disruptor 的使用有一个大致的了解;此外 Disruptor 通常会和 ArrayBlockingQueue 做对比,可以参考 JDK源码分析(11)之 BlockingQueue 相关

一、Disruptor 简介

首先可以从下面两张图看到,Disruptor 的内部结构,只这里我偷了一下懒,图中的内容是老版本的,可能和新版本有点不一样但是主要结构还是一样的;

具体使用示例代码我这里就不贴,大家可以看我上一篇博客;

初始化;首先在启动的时候,需要预先初始化 RingBuffer,所以需要传入 EventFactory;这里和 JUC 里面 Queue 很不一样的地方地方是,RingBuffer 中的 Event 不会被取出,每次 publish 的时候都是覆盖之前的内容,所以 RingBuffer 这里是不会产生 GC 的;而生产者和消费者都持有一个 Sequence,指示当前的处理位置,当需要获取 Event 的时候,可以直接使用 sequence & ringBuffer.size - 1 除留余数法快速找到对应的数组位置;

private void fill(EventFactory<E> eventFactory) {
  for (int i = 0; i < bufferSize; i++) {
    entries[BUFFER_PAD + i] = eventFactory.newInstance();
  }
}

生产者;同时可以指定 Disruptor 是单生产者还是多生产者:

  • ProducerType.SINGLE – SingleProducerSequencer :因为只有一个生产者,所以者更新 sequence 的时候是不需要加锁的;
  • ProducerType.MULTI – MultiProducerSequencer :多个生产者的时候,使用乐观锁机制更新 sequence,即 UNSAFE.compareAndSwapLong

当没有空余位置的时候他们都是使用 LockSupport.parkNanos(1L); 来阻塞线程的,如果有需要你也可以改成其他的等待模式;

// RingBuffer
// 首先通过 Sequencer 拿到下一个可用的序列
public long next() { return sequencer.next(); }

// 然后用除留余数发拿到对应的数组元素
public E get(long sequence) { return elementAt(sequence); }

// 这里是使用 UNSAFE 直接获取内存对象
protected final E elementAt(long sequence) {
  return (E) UNSAFE.getObject(entries, REF_ARRAY_BASE + ((sequence & indexMask) << REF_ELEMENT_SHIFT));
}

// 最后将拿到的数组元素修改为新的 Event,再发布
public void publish(long sequence) { sequencer.publish(sequence); }

// 这里所有关于生产者并发的问题都封装到了 Sequencer 里面,后面最详细讲到

消费者;正因为上面说的 RingBuffer 中的对象不对像普通的 Queue 一样,真正取出,所以在 Disruptor 中可以很容易做到,同一个消息同时被多个消费者获取的逻辑;这里的关键就在于 每个消费者所持有的 Sequence;

  • 当消息可以被重复消费的时候,每个消费者不需要管其他的消费者,每次获取新任务的时候,只需要和生产者的 Sequence 比较就可以了,获取成功后更新自己,这样每个消费者就可以互不影响了;
  • 当消息不能被重复消费的时候,所有的消费者共享一个 Sequence,当发生竞争的时候使用指定的 WaitStrategy 解决冲突;

等待策略;Disruptor 提供了很多从等待策略,这里需要根据实际的业务需求选择使用;同时和 JDK 中的队列相比,无论是阻塞队列还是并发队列,其控制并发的方式都是固定的,而在 Disruptor 中则可以很容易的定制这些策略,从这一点来看也可以说是实现了策略模式;

  • BlockingWaitStrategy: 和 ArrayBlockingQueue 一样使用加锁的方式
  • BusySpinWaitStrategy: Busy Spin strategy 自旋等待
  • LiteBlockingWaitStrategy: BlockingWaitStrategy 的变种,也是使用加锁方式
  • PhasedBackoffWaitStrategy:两段式策略
  • SleepingWaitStrategy: 这是一个在性能和 CPU 占用率做了平衡的一种策略,初始自旋,然后 Yield,最后 Sleep
  • TimeoutBlockingWaitStrategy:同名字
  • YieldingWaitStrategy: 使用自旋、Yield 方式

以上这些就是 Disruptor 的大致框架性内容了,另外有两点是 Disruptor 很快的重要原因;

  • 缓存的引用
  • 并发的处理

二、Disruptor 对缓存的应用

首先计算机中各级存储器的速度差异巨大,数量级描述大致如下:

存储器 容量 速度
寄存器 * / B 1 ns
一级 Cache * / KB 5 ~ 10 ns
二级 Cache * / KB – M 40 ~ 60 ns
内存 */ M – G 100 ~ 150 ns
硬盘 * / G – T 3 ~ 15 ms

根据上图的数据,直观的反应如果想加快软件的运行速度,当然是尽量利用上层的缓存体系;在 JVM 中缓存不是以单字节存在的,而是以缓存行的形式,通常是 2 的整数幂个连续字节,一般为 32-256 个字节。最常见的缓存行大小是 64 个字节;

在我们的队列,数则或者 Disruptor 中,理想状态下就是生产者和消费的速度保持相对一致,这样能避免阻塞的发生,其生产者和消费者就分别位于数组的头部和尾部;

但是这样的理想状态很难到达,要么是生产者快一些,要么是消费者快一些,其结果如下图;

所以头和尾通常都位于同一个缓存行中,这样者更新头的时候,将对应的缓存标记为失效,同时尾也被标记为了失效,者就是伪缓存;

下面是一个缓存的测试例子;

public final class FalseSharing implements Runnable {
  private static final int NUM_THREADS = 4; // change
  private static final long ITERATIONS = 500L * 1000L * 1000L;
  private final int arrayIndex;
  private static VolatileLong[] longs = new VolatileLong[NUM_THREADS];

  static {
    for (int i = 0; i < longs.length; i++) {
      longs[i] = new VolatileLong();
    }
  }

  public FalseSharing(final int arrayIndex) {
    this.arrayIndex = arrayIndex;
  }

  public static void main(final String[] args) throws Exception {
    final long start = System.nanoTime();
    runTest();
    System.out.println("duration = " + (System.nanoTime() - start));
  }

  private static void runTest() throws InterruptedException {
    Thread[] threads = new Thread[NUM_THREADS];
    for (int i = 0; i < threads.length; i++) {
      threads[i] = new Thread(new FalseSharing(i));
    }
    for (Thread t : threads) {
      t.start();
    }

    for (Thread t : threads) {
      t.join();
    }
  }

  @Override
  public void run() {
    long i = ITERATIONS + 1;
    while (0 != --i) {
      longs[arrayIndex].value = i;
    }
  }

  public static final class VolatileLong {
    // public long p1, p2, p3, p4, p5, p6; // cache line padding
    public volatile long value = 0L;
    // public long p8, p9, p10, p11, p12, p13, p14, p15; // cache line padding
  }
}

这里不同的机器测试的结果不同,大家可以修改线程数,padding 数,和 padding 的先后顺序;会得到不同的结果;

我测试的结果:
无 padding :17988876300
有 padding :4667271000
可以看到是查了一个数量级

这样的缓存填充,在 Disruptor 中随处可见:

abstract class RingBufferPad {
  protected long p1, p2, p3, p4, p5, p6, p7;
}

public final class RingBuffer<E> extends RingBufferFields<E> implements Cursored, EventSequencer<E>, EventSink<E> {
  public static final long INITIAL_CURSOR_VALUE = Sequence.INITIAL_VALUE;
  protected long p1, p2, p3, p4, p5, p6, p7;
  ...
}

三、Disruptor 的并发处理

并发的处理,同样的 Disruptor 中随处可见,虽然在平时写代码的时候也会注意,但是当状态变量多了以后,代码就会变得很复杂,不容易读懂;而在 Disruptor 中由 Sequence 串联起来的各个部分,以及策略模式的应用,使得每部分的处理一样的清晰;这里的内容太多了就不一一分析了,比如 MultiProducerSequencer 和 SingleProducerSequencer;

// SingleProducerSequencer
public long next(int n) {
  if (n < 1) throw new IllegalArgumentException("n must be > 0");
  long nextValue = this.nextValue;
  long nextSequence = nextValue + n;
  long wrapPoint = nextSequence - bufferSize;
  long cachedGatingSequence = this.cachedValue;

  if (wrapPoint > cachedGatingSequence || cachedGatingSequence > nextValue) {
    cursor.setVolatile(nextValue);  // StoreLoad fence

    long minSequence;
    while (wrapPoint > (minSequence = Util.getMinimumSequence(gatingSequences, nextValue))) {
        LockSupport.parkNanos(1L); // TODO: Use waitStrategy to spin?
    }

    this.cachedValue = minSequence;
  }

  this.nextValue = nextSequence;
  return nextSequence;
}

// MultiProducerSequencer 
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;

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

总结

对 Disruptor 源码查看的最大感觉是,习以为常的结构设计模式,都可以有更精妙的写法,如果 Sequence 承担的各部分逻辑串联的角色,整体的消费者生产者模式,消费者部分可以看成是观察者模式,也可以看出是事件监听模式,以及并发控制的策略模式;两外就是包括伪缓存在内的各细节优化;

转载请注明:老余博客 » Disruptor 详解 二

读后有收获可以请作者喝咖啡:

喜欢 (0)or分享 (0)
发表我的评论
取消评论
表情

Hi,您需要填写昵称和邮箱!

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址