7

Disruptor高性能队列实现原理

 3 years ago
source link: https://segmentfault.com/a/1190000039970225
Go to the source link to view the article. You can view the picture content, updated content and better typesetting reading experience. If the link is broken, please click the button below to view the snapshot at that time.
neoserver,ios ssh client

一、Disruptor简介

  1. Disruptor是英国外汇交易公司LMAX开发的一个低延迟高性能=无锁的有界循环数组。基于Disruptor开发的系统单线程能支撑每秒600万订单，目前已经开源的并发框架。Log4j2底层使用的并发框架

  2. Disruptor设计特点

    1. 环形数据结构：底层使用的是数组而非连表
    2. 元素位置定位：数组的长度是2^n，下标是递增的，可以通过位运算快速定位
    3. 无锁设计，生产者或消费者需要先申请位置，申请成功以后才能读写，申请过程中通过CAS保证线程安全。
  3. 用于解决单机多线程之间的数据交换，而非类似于kafka的分布式队列。

二、JDK里的队列解决方案

队里有界性锁底层结构ArrayBlockingQueue有界有锁数组LinkedBlockingQueue有界有锁链表ConcurrentLinkedQueue无界无锁链表

在高并发且要求较高的稳定性的系统场景下，非了防止生产者速度过快，只能选有界队列；同时，为了减少Java的垃圾回收对系统性能的影响尽量选择“数组”作为队列的底层结构，符合条件只有一个：ArrayBlockingQueue

2.1 ArrayBlockingQueue的问题

  1. 加锁：不加锁的性能 > CAS操作的性能 > 加锁的性能。

    2.1.2 伪共享

    伪共享：缓存系统中是以缓存行（cache line）为单位存储的，当多线程修改互相独立的变量时，如果这些变量共享同一个缓存行，就会无意中影响彼此的性能，

  2. CPU 和主内存之间有好几层缓存，距离CPU越近，缓存空间越小，速度越快。
  3. CPU运算时，优先从最近的缓存寻找数据，找不到时再往上层去找。

image

  1. 缓存系中以 缓存行(cache line) 为单位存储,一个缓存行有64字节，可以存储8个long类型数据。当cpu访问一个long类型的数组，当数组中的一个值被加载到缓存中，它会额外加载另外7个。当数组的一个值失效，则整个缓存行失效，它将换出其他7个值。

  2. ArrayBlockingQueue有三个成员变量：

      • takeIndex：需要被取走的元素下标
      • putIndex：可被元素插入的位置的下标
      • count：队列中元素的数量

这三个变量很容易放到一个缓存行中，但是之间修改没有太多的关联。所以每次修改，都会使之前缓存的数据失效，从而不能完全达到共享的效果。

public class ArrayBlockingQueue<E> {

    /** The queued items */
    final Object[] items;

    /** items index for next take, poll, peek or remove */
    int takeIndex;

    /** items index for next put, offer, or add */
    int putIndex;

    /** Number of elements in the queue */
    int count;

}
  1. 伪共享解决思路：增大数组元素的间隔使得由不同线程存取的元素位于不同的缓存行上，以空间换时间。

// value1和value2可能会产生伪共享
class ValueNoPadding {
    protected volatile long value1 = 0L;
    protected volatile long value2 = 0L;
}

// value1和value2中间插入无用值 p1~p14 
class ValuePadding {
    protected long p1, p2, p3, p4, p5, p6, p7;
    protected volatile long value1 = 0L;
    protected long p9, p10, p11, p12, p13, p14;
    protected volatile long value2 = 0L;
}

三、Disruptor

RingBuffer

  1. ringBuffer是一个环，用做在不同线程间传递数据的空间
  2. ringBuffer拥有一个序号，整个序号是递增的，用于指向下一个可用元素。
  3. 队列空间在创建时就固定不再改变，可用降低GC的压力
    image

  1. 准备数据容器

    // 数据容器,存放生产和消费的数据内容
public class LongEvent {
 private long value;
}

  2. 准备数据容器的生产工厂，用于RingBuffer初始化时的数据填充

    // 数据容器生产工厂
public class LongEventFactory implements EventFactory<LongEvent> {
 public LongEvent newInstance() {
    return new LongEvent();
 }
}

  3. 准备消费者

    //消费者
public class LongEventConsumer implements EventHandler<LongEvent> {

 /**
  *
  * @param longEvent
  * @param sequence 当前的序列
  * @param endOfBatch 是否是最后一个数据
  * @throws Exception
  */
 @Override
 public void onEvent(LongEvent longEvent, long sequence, boolean endOfBatch) throws Exception {
     String str = String.format("long event : %s l:%s b:%s", longEvent.getValue(), sequence, endOfBatch);
     System.out.println(str);
 }
}

  4. 生产线程、主线程

    public class Main {

 public static void main(String[] args) throws Exception {
     
     // 线程工厂
     ThreadFactory threadFactory = (r) -> new Thread(r);

     // disruptor-创建一个disruptor
     // 设置数据容器的工厂类,ringBuffer的初始化大小,消费者线程的工厂类
     Disruptor<LongEvent> disruptor = new Disruptor<LongEvent>(new LongEventFactory(), 8, threadFactory);
     // disruptor-设置消费者
     disruptor.handleEventsWith(new LongEventConsumer());
     disruptor.start();
     
     // 获取disruptor的RingBuffer
     RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();

     // 主线程开始生产
     for (long l = 0; l <= 8; l++) {
         long nextIndex = ringBuffer.next();
         
         LongEvent event = ringBuffer.get(nextIndex);
         event.setValue(l);
         ringBuffer.publish(nextIndex);
         
         Thread.sleep(1000);
     }
 }
}

单生产者生产数据的流程

  1. 生产者线程申请写入M个数据
  2. disruptor从当前指针cursor顺序去找M个可写空间，返回找到的可用空间的最大序号
  3. 通过CAS比对返回的序号和申请的序号是否一致，判断是否会覆盖未读的元素，若返回正确，直接写入数据

image

多生产者生产数据的流程

  1. 引入一个与ringBuffer大小相同的buff：availableBuffer用于记录ringBuffer每一个空间的使用情况，若生产者写入数据，则将对应availableBuffer位置标记为写入成功，若消费者读取了数据，则将对应的availableBuffer位置标记为空闲。
  2. 多个生产者分配空间时，使用CAS给每一个线程获取不同的数组空间进行操作。
  3. 多个消费者在消费数据时，顺序的从availableBuffer搜索一段连续可读的空间，并返回该空间的最大序列号，并读取数据，同时将availableBuffer的对应的位置进行标记空闲。

image

Disruptor 解决伪共享与线程可见性问题


// 数据左右两边插入多余变量隔离真正的变量
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
{
    static final long INITIAL_VALUE = -1L;
    private static final Unsafe UNSAFE;
    private static final long VALUE_OFFSET;

 
    public Sequence(final long initialValue)
    {
        UNSAFE.putOrderedLong(this, VALUE_OFFSET, initialValue);
    }

    public long get()
    {
        return value;
    }
    // 使用UNSAFE操作直接修改内存值
    public void set(final long value)
    {
        UNSAFE.putOrderedLong(this, VALUE_OFFSET, value);
    }
}

四、参考文献


report

Recommend

About Joyk


Aggregate valuable and interesting links.
Joyk means Joy of geeK