Disruptor框架学习(2)--为啥这么快

public class CacheHit {

    //二维数组：
    private static long[][] longs;

    //一维数组长度：
    private static int length = 1024*1024;

    public static void main(String [] args) throws InterruptedException {
        //创建二维数组,并赋值：
        longs = new long[length][];
        for(int x = 0 ;x < length;x++){
            longs[x] = new long[6];
            for(int y = 0 ;y<6;y++){
                longs[x][y] = 1L;
            }
        }
        cacheHit();
        cacheMiss();
    }
    //缓存命中：
    private static void cacheHit() {
        long sum = 0L;
        long start = System.nanoTime();
        for(int x=0; x < length; x++){
            for(int y=0;y<6;y++){
                sum += longs[x][y];
            }
        }
        System.out.println("命中耗时："+(System.nanoTime() - start));
    }
    //缓存未命中：
    private static void cacheMiss() {
        long sum = 0L;
        long start = System.nanoTime();
        for(int x=0;x < 6;x++){
            for(int y=0;y < length;y++){
                sum += longs[y][x];
            }
        }
        System.out.println("未命中耗时："+(System.nanoTime() - start));
    }
}

未命中耗时：43684518
命中耗时：19244507

public class FalseShare implements Runnable{

    //线程数、数组大小：
    public static int NUM_THREADS = 4; // change
    
    //数组迭代的次数：
    public final static long ITERATIONS = 500L * 1000L * 1000L;
    
    //线程需要处理的数组元素角标：
    private final int handleArrayIndex;
    
    //操作数组：
    private static VolatileLong[] longs = new VolatileLong[NUM_THREADS];
    
    //对数组的元素进行赋值：
    static{
        for (int i = 0; i < longs.length; i++) {
            longs[i] = new VolatileLong();
        }
    }
    
    public FalseShare(final int handleArrayIndex) {
        this.handleArrayIndex = handleArrayIndex;
    }

    //启动线程，每一个线程操作一个数组的元素，一一对应：
    public static void main(final String[] args) throws Exception {
        //程序睡眠必须加上：
        Thread.sleep(10000);
        
        final long start = System.nanoTime();
        
        Thread[] threads = new Thread[NUM_THREADS];
        for (int i = 0; i < threads.length; i++) {
            threads[i] = new Thread(new FalseShare(i));
        }
        for (Thread t : threads) {
            t.start();
        }
        for (Thread t : threads) {
            t.join();
        }
        System.out.println(System.nanoTime() - start);
    }

    //对数组的元素进行操作：
    public void run() {
        long i = ITERATIONS;
        while (0 != --i) {
            longs[handleArrayIndex].value = i;
        }
    }

    //数组元素：
    public final static class VolatileLong {
        public volatile long value = 0L;
        public long p1, p2, p3, p4, p5; //代码1
        public int p6;//代码1
    }
}

未注释代码1：19830512      18472356     19993249    19841462
注释代码1：  21141471      25611265     19939633    29976847

(1)I--本地读请求：CPU读取变量x，如果其他核中的缓存有变量x，且缓存行的状态为M，则将该核心的变量x更新到内存，本核心的再从内存中读取取数据，加载到缓存行中，两个核心的缓存行状态都变成S；如果其他核心的缓存行状态为S或者E，本核心从内存中杜取数据，之后所有核心中的包含变量x的缓存行状态都变成S。

(2)I--本地读请求：CPU读取变量x，如果其他核中的缓存没有变量x，则本核心从内存中读取变量x，存入本核心的缓存行当中，该缓存行状态变成E；

(3)I--本地写请求：CPU读取写入变量x，如果其他核中没有此变量，则从内存中读取，在本核心中修改，此缓存行状态变为M；如果其他缓存行中有变量x，并且状态为M,则需要先将其他核心中的变量x写回内存，本核心再从内存中读取；如果其他缓存行中有变量x，并且状态为E/S，则将其他核心中的缓存行状态置为I，本核心在从内存中读取变量x，之后将本核心的缓存行置为M；

注意，一个缓存除在Invalid状态外都可以满足CPU的读请求，一个invalid的缓存行必须从主存中读取（变成S或者 E状态）来满足该CPU的读请求。

(4)S--远程写请求：多个核心共享变量X，其他核心将变量x修改，本核心中的缓存行不能使用，状态变为I；

(5)S--本地读请求：多个核心共享变量X，本核心读取本缓存中的变量x，状态不变；

(6)S--远程读请求：多个核心共享变量X，其他核心要读取变量X，从主内存中读取变量x，状态置为S，本核心状态S不变；

(7)S--本地写请求：多个核心共享变量X，本核心修改本缓存行中的变量x，必须先将其他核心中所拥有变量x的缓存行状态变成I，本核心缓存行状态置为M；该操作通常使用RequestFor Ownership (RFO)广播的方式来完成；

(8)E--远程读请求：只有本核心拥有变量x，其他核心也要读取变量x,从内存中读取变量x，并将所有拥有变量x的缓存行置为S状态；

(9)E--本地读请求：只有本核心拥有变量x，本核心需要读取变量x，读取本地缓存行中的变量x即可，状态不变依旧为E；

(10)E--远程写请求：只有本核心拥有变量x，其他核心需要修改变量x，其他核心从内存中读取变量x，进行修改，状态变成M，而本核心中缓存行变为状态I；

(11)E--本地写请求：只有本核心拥有变量x，本核心修改本缓存行中的变量x，状态置为M；

(12)M--本地写请求：只有本核心中拥有变量x，本核心进行修改x操作，缓存行状态不变；

(13)M--本地读请求：只有本核心中拥有变量x，本核心进行读取x操作，缓存行状态不变；

(14)M--远程读请求：只有本核心中拥有变量x，其他核心需要读取变量x,先将本核心中的变量x写回到内存中，在将本缓存行状态置为S，其他核心拥有变量x的缓存行状态也变为S；

(15)M--远程写请求：只有本核心中拥有变量x，其他和核心需要修改变量x，先将本核心中的变量x写回内存，再将本核心中缓存行置为I。其他核心的在从缓存行中读取变量x，修改后置为M；

update XXX_TABLE SET MONEY = 100 AND VERSION = 11 WHERE ID = 1 AND VERSION = 10;

public class CasTest implements Runnable{

    private int memoryValue = 1;

    private int expectValue;

    private int updateValue;

    public CasTest(int expectValue,int updateValue){
        this.expectValue = expectValue;
        this.updateValue = updateValue;
    }

    public void run() {
        if(memoryValue==expectValue){
            this.memoryValue = updateValue;
            System.out.println("修改成功");
        }else {
            System.out.println("修改失败");
        }
    }

    public static void main(String[] agrs) throws InterruptedException {
        CasTest casTest1 = new CasTest(1,2);
        Thread t1 = new Thread(casTest1);
        t1.start();
        
        Thread t2= new Thread(casTest1);
        t2.start();

        t1.join();
        t2.join();
    }
}

public class AtomicInteger extends Number implements java.io.Serializable {

    private static final long serialVersionUID = 6214790243416807050L;

    private static final Unsafe unsafe = Unsafe.getUnsafe();

    public final boolean compareAndSet(int expect, int update) {
        return unsafe.compareAndSwapInt(this, valueOffset, expect, update);
    }
}

public final native boolean compareAndSwapInt(Object var1, long var2, int var4, int var5);

public final class MultiProducerSequencer extends AbstractSequencer{
    @Override
    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){
                    waitStrategy.signalAllWhenBlocking();
                    LockSupport.parkNanos(1); // TODO, should we spin based on the wait strategy?
                    continue;
                }
                gatingSequenceCache.set(gatingSequence);

            //对current,next进行compareAndSet，cursor就是序列号对象：
            } else if (cursor.compareAndSet(current, next)){
                break;
            }
        }while (true);
        return next;
    }
}

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