构建高性能内存队列：Disruptor 永远滴神~

Java中有哪些行列

• ArrayBlockingQueue 运用ReentrantLock
• LinkedBlockingQueue 运用ReentrantLock
• ConcurrentLinkedQueue 运用CAS
• 等等

咱们清楚运用锁的性能比较低，尽量运用无锁设计。接下来就咱们来认识下Disruptor。

Disruptor简单运用

github地址：github.com/LMAX-Exchan…

先简单介绍下：

• Disruptor它是一个开源的并发结构，并获得2011 Duke’s程序结构创新奖【Oracle】，能够在无锁的情况下完成网络的Queue并发操作。英国外汇交易公司LMAX开发的一个高性能行列，号称单线程能支撑每秒600万订单~
• 日志结构Log4j2 异步形式采用了Disruptor来处理
• 局限呢，他便是个内存行列，也便是说无法支撑分布式场景。

简单运用

数据传输目标

@Data
public class EventData {
    private Long value;
}

顾客

public class EventConsumer implements WorkHandler<EventData> {
    /**
     * 消费回调
     * @param eventData
     * @throws Exception
     */
    @Override
    public void onEvent(EventData eventData) throws Exception {
        Thread.sleep(5000);
        System.out.println(Thread.currentThread() + ", eventData:" + eventData.getValue());
    }
}

出产者

public class EventProducer {
    private final RingBuffer<EventData> ringBuffer;
    public EventProducer(RingBuffer<EventData> ringBuffer) {
        this.ringBuffer = ringBuffer;
    }
    public void sendData(Long v){
        // cas展位
        long next = ringBuffer.next();
        try {
            EventData eventData = ringBuffer.get(next);
            eventData.setValue(v);
        } finally {
            // 通知等待的顾客
            System.out.println("EventProducer send success, sequence:"+next);
            ringBuffer.publish(next);
        }
    }
}

测验类

public class DisruptorTest {
    public static void main(String[] args) {
        // 2的n次方
        int bufferSize = 8;
        Disruptor<EventData> disruptor = new Disruptor<EventData>(
                () -> new EventData(), // 事情工厂
                bufferSize,            // 环形数组巨细
                Executors.defaultThreadFactory(),       // 线程池工厂
                ProducerType.MULTI,    // 支持多事情发布者
                new BlockingWaitStrategy());    // 等待战略
        // 设置顾客
        disruptor.handleEventsWithWorkerPool(
                new EventConsumer(),
                new EventConsumer(),
                new EventConsumer(),
                new EventConsumer());
        disruptor.start();
        RingBuffer<EventData> ringBuffer = disruptor.getRingBuffer();
        EventProducer eventProducer = new EventProducer(ringBuffer);
        long i  = 0;
        for(;;){
            i++;
            eventProducer.sendData(i);
            try {
                Thread.sleep(1500);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

中心组件

根据上面简单比如来看的确很简单，Disruptor帮咱们封装好了出产消费模型的完成，接下来咱们来看下他是根据哪些中心组件来支撑起一个高性能无锁行列呢？

RingBuffer： 环形数组，底层运用数组entries，在初始化时填充数组，防止不断新建目标带来的开销。后续只会对entries做更新操作

Sequencer： 中心管家

• 界说出产同步的完成：SingleProducerSequencer单出产、MultiProducerSequencer多出产

• 当时写的进展Sequence cursor

• 一切顾客进展的数组Sequence[] gatingSequences

• MultiProducerSequencer可用区availableBuffer【运用空间交换查询功率】

Sequence： 自身便是一个序号器用来标识处理进展，也能够作为是一个atomicInteger; 还有别的一个特色，为了处理伪同享问题而引进的：缓存行填充。这个在后面介绍。

workProcessor: 处理Event的循环，在循环中获取Disruptor的事情，然后把事情分配给各个handler

EventHandler: 担任事务逻辑的handler,自己完成。

WaitStrategy： 顾客 怎么等待 事情的战略，界说了如下战略

• leepingWaitStrategy：自旋 + yield + sleep

• BlockingWaitStrategy：加锁，合适CPU资源紧张（不需要切换线程），体系吞吐量无要求的

• YieldingWaitStrategy：自旋 + yield + 自旋

• BusySpinWaitStrategy：自旋，削减线程之前切换

• PhasedBackoffWaitStrategy：自旋 + yield + 自界说战略

关注大众号：码猿技能专栏 每天守时推送更多精彩内容

带着问题来解析代码？

1、多出产者怎么保证消息出产不会相互掩盖。【怎么达到互斥效果】

每个线程获取不同的一段数组空间，然后通过CAS判别这段空间是否现已分配出去。

接下来咱们看下多出产类MultiProducerSequencer中next办法【获取出产序号】

// 顾客上一次消费的最小序号 // 后续第二点会讲到
private final Sequence gatingSequenceCache = new Sequence(Sequencer.INITIAL_CURSOR_VALUE);
// 当时进展的序号
protected final Sequence cursor = new Sequence(Sequencer.INITIAL_CURSOR_VALUE);
// 一切顾客的序号 //后续第二点会讲到
protected volatile Sequence[] gatingSequences = new Sequence[0];
 public long next(int n)
    {
        if (n < 1)
        {
            throw new IllegalArgumentException("n must be > 0");
        }
        long current;
        long next;
        do
        {
            // 当时进展的序号，Sequence的value具有可见性，保证多线程间线程之间能感知到可请求的最新值
            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);
                // 阐明大于1圈，并没有多余空间能够请求
                if (wrapPoint > gatingSequence)
                {
                    LockSupport.parkNanos(1); // TODO, should we spin based on the wait strategy?
                    continue;
                }
                // 更新最小值到Sequence的value中
                gatingSequenceCache.set(gatingSequence);
            }
            // CAS成功后更新当时Sequence的value
            else if (cursor.compareAndSet(current, next))
            {
                break;
            }
        }
        while (true);
        return next;
    }

2、出产者向序号器请求写的序号，如序号正在被消费，Sequencer是怎么知道哪些序号是能够被写入的呢？【未消费则被掩盖怎么处理】

从gatingSequences中取得最小的序号，出产者最多能写到这个序号的后一位。浅显来讲便是请求的序号不能大于最小顾客序号一圈【请求到最大序列号-buffersize 要小于/等于 最小消费的序列号】的时分， 才干请求到当时写的序号

public final EventHandlerGroup<T> handleEventsWithWorkerPool(final WorkHandler<T>... workHandlers)
{
    return createWorkerPool(new Sequence[0], workHandlers);
}
EventHandlerGroup<T> createWorkerPool(
    final Sequence[] barrierSequences, final WorkHandler<? super T>[] workHandlers)
{
    final SequenceBarrier sequenceBarrier = ringBuffer.newBarrier(barrierSequences);
    final WorkerPool<T> workerPool = new WorkerPool<>(ringBuffer, sequenceBarrier, exceptionHandler, workHandlers);
    consumerRepository.add(workerPool, sequenceBarrier);
    final Sequence[] workerSequences = workerPool.getWorkerSequences();
    updateGatingSequencesForNextInChain(barrierSequences, workerSequences);
    return new EventHandlerGroup<>(this, consumerRepository, workerSequences);
}
    private void updateGatingSequencesForNextInChain(final Sequence[] barrierSequences, final Sequence[] processorSequences)
{
    if (processorSequences.length > 0)
    {
        // 顾客启动后就会将一切顾客存放入AbstractSequencer中gatingSequences
        ringBuffer.addGatingSequences(processorSequences);
        for (final Sequence barrierSequence : barrierSequences)
        {
            ringBuffer.removeGatingSequence(barrierSequence);
        }
        consumerRepository.unMarkEventProcessorsAsEndOfChain(barrierSequences);
    }
}

3、在多出产者情况下，出产者是请求到一段可写入的序号，然后再写入这些序号中，那么顾客是怎么感知哪些序号是能够被消费的呢？【借问提1图阐明】

这个前提是多出产者情况下，第一点咱们说过每个线程获取不同的一段数组空间，那么现在单单通过序号现已不够用了，MultiProducerSequencer运用了int 数组 【availableBuffer】来标识当时序号是否可用。当出产者成功出产事情后会将availableBuffer中当时序列号置为1标识能够读取。

如此顾客能够读取的的最大序号便是咱们availableBuffer中第一个不可用序号-1。

初始化availableBuffer流程

public MultiProducerSequencer(int bufferSize, final WaitStrategy waitStrategy)
{
    super(bufferSize, waitStrategy);
    // 初始化可用数组
    availableBuffer = new int[bufferSize];
    indexMask = bufferSize - 1;
    indexShift = Util.log2(bufferSize);
    initialiseAvailableBuffer();
}
// 初始化默认availableBuffer为-1
private void initialiseAvailableBuffer()
{
    for (int i = availableBuffer.length - 1; i != 0; i--)
    {
        setAvailableBufferValue(i, -1);
    }
    setAvailableBufferValue(0, -1);
}
// 出产者成功出产事情将可用区数组置为1
public void publish(final long sequence)
{
    setAvailable(sequence);
    waitStrategy.signalAllWhenBlocking();
}
private void setAvailableBufferValue(int index, int flag)
{
    long bufferAddress = (index * SCALE) + BASE;
    UNSAFE.putOrderedInt(availableBuffer, bufferAddress, flag);
}

顾客消费流程

WorkProcessor类中消费run办法
public void run()
    {
        boolean processedSequence = true;
        long cachedAvailableSequence = Long.MIN_VALUE;
        long nextSequence = sequence.get();
        T event = null;
        while (true)
        {
            try
            {
                // 先通过cas获取消费事情的占有权
                if (processedSequence)
                {
                    processedSequence = false;
                    do
                    {
                        nextSequence = workSequence.get() + 1L;
                        sequence.set(nextSequence - 1L);
                    }
                    while (!workSequence.compareAndSet(nextSequence - 1L, nextSequence));
                }
                // 数据就绪，能够消费
                if (cachedAvailableSequence >= nextSequence)
                {
                    event = ringBuffer.get(nextSequence);
                    // 触发回调函数
                    workHandler.onEvent(event);
                    processedSequence = true;
                }
                else
                {
                    // 获取能够被读取的下标
                    cachedAvailableSequence = sequenceBarrier.waitFor(nextSequence);
                }
            }
        // ....省略
        }
        notifyShutdown();
        running.set(false);
    }
    public long waitFor(final long sequence)
        throws AlertException, InterruptedException, TimeoutException
    {
        checkAlert();
        // 这个值获取的current write 下标，能够以为大局消费下标。此处与每一段的write1和write2下标区分开
        long availableSequence = waitStrategy.waitFor(sequence, cursorSequence, dependentSequence, this);
        if (availableSequence < sequence)
        {
            return availableSequence;
        }
        // 通过availableBuffer筛选出第一个不可用序号 -1
        return sequencer.getHighestPublishedSequence(sequence, availableSequence);
    }
    public long getHighestPublishedSequence(long lowerBound, long availableSequence)
    {
        // 从current read下标开端， 循环至 current write,假如碰到availableBuffer 为-1 直接返回
        for (long sequence = lowerBound; sequence <= availableSequence; sequence++)
        {
            if (!isAvailable(sequence))
            {
                return sequence - 1;
            }
        }
        return availableSequence;
    }

处理伪同享问题

什么是伪同享问题呢？

为了进步CPU的速度，Cpu有高速缓存Cache,该缓存最小单位为缓存行CacheLine,他是从主内存复制的Cache的最小单位，通常是64字节。一个Java的long类型是8字节，因此在一个缓存行中能够存8个long类型的变量。假如你拜访一个long数组，当数组中的一个值被加载到缓存中，它会额定加载别的7个。因此你能十分快地遍历这个数组。

关注大众号：码猿技能专栏 每天守时推送更多精彩内容

伪同享问题是指，当多个线程同享某份数据时，线程1或许拉到线程2的数据在其cache line中，此刻线程1修正数据，线程2取其数据时就要重新从内存中拉取，两个线程相互影响，导致数据虽然在cache line中，每次却要去内存中拉取。

Disruptor是怎么处理的呢？

在value前后一致都加入7个Long类型进行填充，线程拉取时，不论怎么都会占满整个缓存

回顾总结：Disuptor为何能称之为高性能的无锁行列结构呢？

• 缓存行填充，防止缓存频繁失效。【java8中也引进@sun.misc.Contended注解来防止伪同享】
• 无锁竞赛：通过CAS 【二阶段提交】
• 环形数组：数据都是掩盖，防止GC
• 底层更多的运用位运算来提升功率