当CPU写入一个地址中的数据时，同样会先将数据写入 L1 Cache 中，然后再根据缓存一致性协议将数据写入 L2 Cache、L3 Cache 以及主内存中。具体写入过程与缓存一致性协议相关，有可能只写入 L1 Cache 中，也有可能需要将数据写入 L2 Cache、L3 Cache 以及主内存中。写入过程中也可能会使用缓存行失效、写回等技术来提高效率。

思考：这种缓存架构在多线程访问的时候存在什么问题？

CPU多核缓存架构缓存一致性问题分析

场景一

缓存一致性协议不能使用的特殊情况：

        当操作的数据不能被缓存在处理器内部，或操作的数据跨多个缓存行时 ，则处理器会调用总线锁定。 不能被缓存在处理器内部的数据通常指的是不可缓存的设备内存（如显存、网络接口卡的缓存 等），这些设备内存一般不受缓存一致性协议的管辖，处理器无法将其缓存到自己的缓存行中。

        有些处理器不支持缓存锁定。 早期的Pentium系列处理器并不支持缓存锁定机制。在这些处理器上，只能使用总 线锁定来实现原子操作。但现代的处理器通常都支持缓存锁定机制，因此应该尽量使用缓存锁定来实现原子操作，以获得更好的性能。

总线窥探

        当特定数据被多个缓存共享时，处理器修改了共享数据的值，更改必须传播到所有其他具有该数据 副本的缓存中。这种更改传播可以防止系统违反缓存一致性。 数据变更的通知可以通过总线窥探来完成。所有的窥探者都在监视总线上的每一个事务。如果一个修改共享缓存块的事务出现在总线上，所有的窥探者都会检查他们的缓存是否有共享块的相同副本。 如果缓存中有共享块的副本，则相应的窥探者执行一个动作以确保缓存一致性。 这个动作可以是刷新缓存块或使缓存块失效。它还涉及到缓存块状态的改变，这取决于缓存一致性协议（cache coherence protocol）。

MESI协议

        MESI协议 是一个基于写失效的缓存一致性协议，是支持回写（write-back）缓存的最常用协议。 也称作伊利诺伊协议 (Illinois protocol，因为是在伊利诺伊大学厄巴纳-香槟分校被发明的)。

缓存行有4种不同的状态:

public class OrderEventHandler implements EventHandler<OrderEvent> {

 @Override
 public void onEvent(OrderEvent event, long sequence, boolean endOfBatch) throws
Exception {
 // TODO 消费逻辑
 System.out.println("消费者获取数据value:"+
event.getValue()+",name:"+event.getName());
 }
 }

public class DisruptorDemo {

    public static void main(String[] args) throws Exception {

        //创建disruptor
        Disruptor<OrderEvent> disruptor = new Disruptor<>(
                new OrderEventFactory(),
                1024 * 1024,
                Executors.defaultThreadFactory(),
                ProducerType.SINGLE, //单生产者
                new YieldingWaitStrategy() //等待策略
        );
        //设置消费者用于处理RingBuffer的事件
        disruptor.handleEventsWith(new OrderEventHandler());
        disruptor.start();

        //创建ringbuffer容器
        RingBuffer<OrderEvent> ringBuffer = disruptor.getRingBuffer();
        //创建生产者
        OrderEventProducer eventProducer = new OrderEventProducer(ringBuffer);
        //发送消息
        for (int i = 0; i < 100; i++) {
            eventProducer.onData(i, "Fox" + i);
        }

        disruptor.shutdown();
    }
}

如果消费者是多个，只需要在调用 handleEventsWith 方法时将多个消费者传递进去

//设置多消费者,消息会被重复消费
 disruptor.handleEventsWith(new OrderEventHandler(), new OrderEventHandler());

//设置多消费者,消费者要实现WorkHandler接口，一条消息只会被一个消费者消费
 disruptor.handleEventsWithWorkerPool(new OrderEventHandler(), new OrderEventHandler());

public class OrderEventHandler implements EventHandler<OrderEvent>,
        WorkHandler<OrderEvent> {

    @Override
    public void onEvent(OrderEvent event, long sequence, boolean endOfBatch) throws
            Exception {
        // TODO 消费逻辑
        System.out.println("消费者"+ Thread.currentThread().getName()
                +"获取数据value:"+ event.getValue()+",name:"+event.getName());
    }

    @Override
    public void onEvent(OrderEvent event) throws Exception {
        // TODO 消费逻辑
        System.out.println("消费者"+ Thread.currentThread().getName()
                +"获取数据value:"+ event.getValue()+",name:"+event.getName());
    }
}

多生产者多消费者模式

public class DisruptorDemo2 {

    public static void main(String[] args) throws Exception {

        //创建disruptor
        Disruptor<OrderEvent> disruptor = new Disruptor<>(
                new OrderEventFactory(),
                1024 * 1024,
                Executors.defaultThreadFactory(),
                ProducerType.MULTI, //多生产者
                new YieldingWaitStrategy() //等待策略
        );

        //设置消费者用于处理RingBuffer的事件
        //disruptor.handleEventsWith(new OrderEventHandler());
        //设置多消费者,消息会被重复消费
        //disruptor.handleEventsWith(new OrderEventHandler(),new OrderEventHandler());
        //设置多消费者,消费者要实现WorkHandler接口，一条消息只会被一个消费者消费
        disruptor.handleEventsWithWorkerPool(new OrderEventHandler(), new
                OrderEventHandler());

        //启动disruptor
        disruptor.start();

        //创建ringbuffer容器
        RingBuffer<OrderEvent> ringBuffer = disruptor.getRingBuffer();

        new Thread(() -> {
            //创建生产者
            OrderEventProducer eventProducer = new OrderEventProducer(ringBuffer);
            // 发送消息
            for (int i = 0; i < 100; i++) {
                eventProducer.onData(i, "Fox" + i);
            }
        }, "producer1").start();

        new Thread(() -> {
            //创建生产者
            OrderEventProducer eventProducer = new OrderEventProducer(ringBuffer);
            // 发送消息
            for (int i = 0; i < 100; i++) {
                eventProducer.onData(i, "monkey" + i);
            }
        }, "producer2").start();


        //disruptor.shutdown();

    }
}

3.RingBuffer

3.1 ringbuffer到底是什么

它是一个环（首尾相接的环），你可以把它用做在不同上下文（线程）间传递数据的buffer。

基本来说，ringbuffer拥有一个序号，这个序号指向数组中下一个可用的元素。（如下图右边的图片表示序号，这个序号指向数组的索引4的位置。）

随着你不停地填充这个buffer（可能也会有相应的读取），这个序号会一直增长，直到绕过这个环。

要找到数组中当前序号指向的元素，可以通过sequence & （array length－1） = array index，比如一共有8槽，3&（8－1）=3，HashMap就是用这个方式来定位数组元素的，这种方式比取模的速度更快。

常用的队列之间的区别

• 没有尾指针。只维护了一个指向下一个可用位置的序号。
• 不删除buffer中的数据，也就是说这些数据一直存放在buffer中，直到新的数据覆盖他们

3.2 ringbuffer采用这种数据结构原因

• 因为它是数组，所以要比链表快，数组内元素的内存地址的连续性存储的。这是对CPU缓存友好的—也就是说，在硬件级别，数组中的元素是会被预加载的，因此在ringbuffer当中，cpu无需时不时去主存加载数组中的下一个元素。因为只要一个元素被加载到缓存行，其他相邻的几个元素也会被加载进同一个缓存行。
• 其次，你可以为数组预先分配内存，使得数组对象一直存在（除非程序终止）。这就意味着不需要花大量的时间用于垃圾回收。此外，不像链表那样，需要为每一个添加到其上面的对象创造节点对象—对应的，当删除节点时，需要执行相应的内存清理操作。

3.3 如何从Ringbuffer读取

消费者(Consumer)是一个想从Ring Buffer里读取数据的线程，它可以访问ConsumerBarrier对象——这个对象由RingBuffer创建并且代表消费者与RingBuffer进行交互。就像Ring Buffer显然需要一个序号才能找到下一个可用节点一样，消费者也需要知道它将要处理的序号——每个消费者都需要找到下一个它要访问的序号。在上面的例子中，消费者处理完了Ring Buffer里序号8之前（包括8）的所有数据，那么它期待访问的下一个序号是9。

消费者可以调用ConsumerBarrier对象的waitFor()方法，传递它所需要的下一个序号.

final long availableSeq = consumerBarrier.waitFor(nextSequence);

ConsumerBarrier返回RingBuffer的最大可访问序号——在上面的例子中是12。ConsumerBarrier有一个WaitStrategy方法来决定它如何等待这个序号.

接下来

接下来，消费者会一直逛来逛去，等待更多数据被写入 Ring Buffer。并且，写入数据后消费者会收到通知——节点 9，10，11 和 12 已写入。现在序号 12 到了，消费者可以指示 ConsumerBarrier 去拿这些序号里的数据了。

在Disruptor中采用了数组的方式保存了我们的数据，上面我们也介绍了采用数组保存我们访问时很好的利用缓存，但是在Disruptor中进一步选择采用了环形数组进行保存数据，也就是RingBuffer。在这里先说明一下环形数组并不是真正的环形数组，在RingBuffer中是采用取余的方式进行访问的，比如数组大小为 10，0访问的是数组下标为0这个位置，其实10，20等访问的也是数组的下标为0的这个位置。

实际上，在这些框架中取余并不是使用%运算，都是使用的&与运算，这就要求你设置的大小一般是2的N次方也就是，10,100,1000等等，这样减去1的话就是，1，11，111，就能很好的使用index & (size -1),这样利用位运算就增加了访问速度。
如果在Disruptor中你不用2的N次方进行大小设置，他会抛出buffersize必须为2的N次方异常。

• Producer会向这个RingBuffer中填充元素，填充元素的流程是首先从RingBuffer读取下一个Sequence，之后在这个Sequence位置的槽填充数据，之后发布。
• Consumer消费RingBuffer中的数据，通过SequenceBarrier来协调不同的Consumer的消费先后顺序，以及获取下一个消费位置Sequence。
• Producer在RingBuffer写满时，会从头开始继续写替换掉以前的数据。但是如果有SequenceBarrier指向下一个位置，则不会覆盖这个位置，阻塞到这个位置被消费完成。Consumer同理，在所有Barrier被消费完之后，会阻塞到有新的数据进来。

3.4 一个生产者

生产者单线程写数据的流程比较简单:

1. 申请写入m个元素；
2. 若是有m个元素可以写入, 则返回最大的序列号. 这儿主要判断是否会覆盖未读的元素

3. 若是返回的正确, 则生产者开始写入元素.