/**
 * 缓存行测试
 *
 * @author tangjiacheng
 * @since 2021/8/26 21:30
 */
public class CacheLineTest {

    static  long[][] arr;

    public static void main(String[] args) {
        arr = new long[1024 * 1024][];
        // 初始化数组内存，保证数组分配一块连续内存
        for (int i = 0; i < 1024 * 1024; i++) {
            arr[i] = new long[8];
            for (int j = 0; j < 8; j++) {
                arr[i][j] = 0L;
            }
        }
        long sum = 0L;
        // 每次取 8 个 long 类型变量
        long marked = System.currentTimeMillis();
        for (int i = 0; i < 1024 * 1024; i += 1) {
            for (int j = 0; j < 8; j++) {
                sum = arr[i][j];
            }
        }
        System.out.println("Loop times: " + (System.currentTimeMillis() - marked) + " ms");
        // 每次取 1024 * 1024 个 long 类型变量 
        marked = System.currentTimeMillis();
        for (int i = 0; i < 8; i += 1) {
            for (int j = 0; j < 1024 * 1024; j++) {
                sum = arr[j][i];
            }
        }
        System.out.println("Loop times: " + (System.currentTimeMillis() - marked) + " ms");
    }
}

class LhsPadding {
    protected long p1;
    protected long p2;
    protected long p3;
    protected long p4;
    protected long p5;
    protected long p6;
    protected long p7;

    LhsPadding() {
    }
}

class Value extends LhsPadding {
    protected volatile long value;

    Value() {
    }
}

class RhsPadding extends Value {
    protected long p9;
    protected long p10;
    protected long p11;
    protected long p12;
    protected long p13;
    protected long p14;
    protected long p15;

    RhsPadding() {
    }
}

public class Sequence extends RhsPadding {
    static final long INITIAL_VALUE = -1L;
    private static final Unsafe UNSAFE = Util.getUnsafe();
    private static final long VALUE_OFFSET;

    public Sequence() {
        this(-1L);
    }

    public Sequence(long initialValue) {
        UNSAFE.putOrderedLong(this, VALUE_OFFSET, initialValue);
    }
    //省略 getter, setter 等方法
    static {
        try {
            VALUE_OFFSET = UNSAFE.objectFieldOffset(Value.class.getDeclaredField("value"));
        } catch (Exception var1) {
            throw new RuntimeException(var1);
        }
    }
}

/**
 * Disruptor 测试 demo
 *
 * @author tangjiacheng
 * @since 2021/8/27 11:55
 */
public class DisruptorTest {

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

            private int value;

            public int get() {
                return value;
            }

            public void set(int value) {
                this.value = value;
            }

        }

        // 生产者的线程工厂
        ThreadFactory threadFactory = r -> new Thread(r, "simpleThread");

        // RingBuffer生产工厂,初始化RingBuffer的时候使用
        EventFactory<Event> factory = Event::new;

        // 处理Event的handler
        EventHandler<Event> handler = (event, sequence, endOfBatch) -> System.out.println("event: " + event.get());

        // 阻塞策略
        BlockingWaitStrategy strategy = new BlockingWaitStrategy();

        // 指定RingBuffer的大小
        int bufferSize = 16;

        // 创建disruptor，采用单生产者模式
        Disruptor<Event> disruptor = new Disruptor(factory, bufferSize, threadFactory, ProducerType.SINGLE, strategy);

        // 设置EventHandler
        disruptor.handleEventsWith(handler);

        // 启动disruptor的线程
        disruptor.start();

        RingBuffer<Event> ringBuffer = disruptor.getRingBuffer();

        for (int l = 0; l < 10; l++) {
            // 获取下一个可用位置的下标
            long sequence = ringBuffer.next();
            try {
                // 返回可用位置的元素
                Event event = ringBuffer.get(sequence);
                // 设置该位置元素的值
                event.set(l);
            } finally {
                // 将事件发布给消费者
                ringBuffer.publish(sequence);
            }
            Thread.sleep(10);
        }
    }

}

// SingleProducerSequencer.java

public long next() {
        return this.next(1);
    }

    public long next(int n) {
        // 校验申请 slot 个数
        if (n < 1) {
            throw new IllegalArgumentException("n must be > 0");
        } else {
            // 上次返回的 seq
            long nextValue = this.nextValue;
            // 应该返回的序列值
            long nextSequence = nextValue + (long)n;
            long wrapPoint = nextSequence - (long)this.bufferSize;
            // 获取消费者未消费元素的最小值
            long cachedGatingSequence = this.cachedValue;
            if (wrapPoint > cachedGatingSequence || cachedGatingSequence > nextValue) {
                this.cursor.setVolatile(nextValue);

                long minSequence;
                // 消费者速度小于生产者，生产者线程在此处自旋
                while(wrapPoint > (minSequence = Util.getMinimumSequence(this.gatingSequences, nextValue))) {
                    LockSupport.parkNanos(1L);
                }

                this.cachedValue = minSequence;
            }

            this.nextValue = nextSequence;
            return nextSequence;
        }
    }

// MultiProducerSequencer.java
public long next() {
        return this.next(1);
    }

    public long next(int n) {
        if (n < 1) {
            throw new IllegalArgumentException("n must be > 0");
        } else {
            long current;
            long next;
            do {
                while(true) {
                    current = this.cursor.get();
                    next = current + (long)n;
                    long wrapPoint = next - (long)this.bufferSize;
                    // gatingSequenceCache 实际上是 单生产者的 原子类版本
                    long cachedGatingSequence = this.gatingSequenceCache.get();
                    if (wrapPoint <= cachedGatingSequence && cachedGatingSequence <= current) {
                        break;
                    }

                    long gatingSequence = Util.getMinimumSequence(this.gatingSequences, current);
                    if (wrapPoint > gatingSequence) {
                        LockSupport.parkNanos(1L);
                    } else {
                        this.gatingSequenceCache.set(gatingSequence);
                    }
                }
            } while(!this.cursor.compareAndSet(current, next));

            return next;
        }
    }

// SingleProducerSequencer.java
public void publish(long sequence) {
        // 更新游标
        this.cursor.set(sequence);
        // 给消费者发送信号
        this.waitStrategy.signalAllWhenBlocking();
    }

// MultiProducerSequencer.java
public void publish(long sequence) {
        // 额外使用 availableBuffer 数组，标记 slot 是否被成功发布
        this.setAvailable(sequence);
        this.waitStrategy.signalAllWhenBlocking();
    }