//以AtomicInteger为例，这里是Unsafe类  
//AtomicInteger::compareAndSet 实际上是调用的一个Unsafe类的本地方法，只需要知道这个方法是在JNI里是借助于一个CPU指令完成的，是原子性的就可以了 	
public final native boolean compareAndSwapInt(Object var1, long var2, int var4, int var5);
//AtomicInteger::getAndSet 就是去不断自旋，判断内存值是否等于新值，直到设置成功退出循环，返回旧值
public final int getAndSetInt(Object var1, long var2, int var4) {
    int var5;
    do {
        var5 = this.getIntVolatile(var1, var2);
    } while(!this.compareAndSwapInt(var1, var2, var5, var4));

    return var5;
}

//非公平锁  NonfairSync内部类使用Sync内部类的方法
final boolean nonfairTryAcquire(int acquires) {
  final Thread current = Thread.currentThread();
  int c = getState();
  if (c == 0) {
    if (compareAndSetState(0, acquires)) {
      setExclusiveOwnerThread(current);
      return true;
    }
  }
  else if (current == getExclusiveOwnerThread()) {
    int nextc = c + acquires;
    if (nextc < 0) // overflow
      throw new Error("Maximum lock count exceeded");
    setState(nextc);
    return true;
  }
  return false;
}

//公平锁  FairSync内部类
protected final boolean tryAcquire(int acquires) {
  final Thread current = Thread.currentThread();
  int c = getState();
  if (c == 0) {
    if (!hasQueuedPredecessors() &&   //只有这里有区别，公平锁如果有队列，就会放入队列，而非公平锁只要CAS成功就插队
        compareAndSetState(0, acquires)) {
      setExclusiveOwnerThread(current);
      return true;
    }
  }
  else if (current == getExclusiveOwnerThread()) {
    int nextc = c + acquires;
    if (nextc < 0)
      throw new Error("Maximum lock count exceeded");
    setState(nextc);
    return true;
  }
  return false;
}
}

//ReentrantReadWriteLock内部静态类Sync
//写锁尝试
protected final boolean tryAcquire(int acquires) {
    Thread current = Thread.currentThread();
    int c = getState();
    int w = exclusiveCount(c); //取写锁的个数
    if (c != 0) {
        if (w == 0 || current != getExclusiveOwnerThread()) //存在读锁,返回失败
            return false;
        if (w + exclusiveCount(acquires) > MAX_COUNT) //写锁数量超过2^16，返回失败
            throw new Error("Maximum lock count exceeded");
        setState(c + acquires);
        return true;
    }
    if (writerShouldBlock() ||  //如果当且写线程数为0，并且当前线程需要阻塞那么就返回失败
        !compareAndSetState(c, c + acquires)) //如果通过CAS增加写线程数失败也返回失败
        return false;
    setExclusiveOwnerThread(current);
    return true;
}
//读锁尝试
protected final int tryAcquireShared(int unused) {
    Thread current = Thread.currentThread();
    int c = getState();
    if (exclusiveCount(c) != 0 &&
        getExclusiveOwnerThread() != current)
        return -1; // 如果其他线程已经获取了写锁，则当前线程获取读锁失败，进入等待状态
    int r = sharedCount(c);
    if (!readerShouldBlock() &&
        r < MAX_COUNT &&
        compareAndSetState(c, c + SHARED_UNIT)) {
        if (r == 0) {
            firstReader = current;
            firstReaderHoldCount = 1;
        } else if (firstReader == current) {
            firstReaderHoldCount++;
        } else {
            HoldCounter rh = cachedHoldCounter;
            if (rh == null || rh.tid != getThreadId(current))
                cachedHoldCounter = rh = readHolds.get();
            else if (rh.count == 0)
                readHolds.set(rh);
            rh.count++;
        }
        return 1;
    }
    return fullTryAcquireShared(current);
}

class Process extends Thread{
    private Runnable runnable;
    private int millis;
    public Process(int millis,Runnable runnable) { this.millis = millis;this.runnable = runnable; }
    public void run() {
        try {
            Thread.sleep(millis);
            runnable.run();
        } catch (InterruptedException e) {}
    }
}

//当3个栅栏都完成后，await() 执行之后的代码
CountDownLatch latch = new CountDownLatch(3);
Process p1 =  new Process(2000,()-> {System.out.println("p1 finished");latch.countDown();});
Process p2 =  new Process(2000,()-> {System.out.println("p2 finished");latch.countDown();});
Process p3 =  new Process(10000,()-> {System.out.println("p3 finished");latch.countDown();});
Process p4 =  new Process(20000,()-> {System.out.println("p4 finished");latch.countDown();});
Process awaitP =  new Process(1000,()-> {try { latch.await();System.out.println("await p finished");} catch (InterruptedException e){}});

awaitP.start();
p1.start();
p2.start();
p3.start();
p4.start();

//p2 finished
//p1 finished
//p3 finished
//await p finished
//p4 finished


//当然也可以设置超时时间
Process awaitP =  new Process(1000,()-> {try { latch.await(5000, TimeUnit.MILLISECONDS);System.out.println("await p finished");} catch (InterruptedException e){}});

//p1 finished
//p2 finished
//await p finished
//p3 finished
//p4 finished

//当count数量线程都到达barrier后，执行线程后面的代码，同样可以设置超时时间，之后不再叙述
CyclicBarrier barrier = new CyclicBarrier(3);
Process p1 =  new Process(2000,()-> {System.out.println("p1 begins to wait");try { barrier.await();System.out.println("p1 finished");} catch (Exception e){}});
Process p2 =  new Process(2000,()-> {System.out.println("p2 begins to wait");try { barrier.await();System.out.println("p2 finished");} catch (Exception e){}});
Process p3 =  new Process(5000,()-> {System.out.println("p3 begins to wait");try { barrier.await();System.out.println("p3 finished");} catch (Exception e){}});
p1.start();
p2.start();
p3.start();

//p1 begins to wait
//p2 begins to wait
//p3 begins to wait
//p3 finished
//p1 finished
//p2 finished

//用于交换数据
Exchanger<String> exchanger = new Exchanger<>();
Process p1 =  new Process(2000,()-> {String s1 = "p1 data";try { exchanger.exchange(s1);System.out.println("get "+ s1 +" , p1 finished"); } catch (InterruptedException e){}});
Process p2 =  new Process(5000,()-> {String s2 = "p2 data";try { exchanger.exchange(s2);System.out.println("get "+ s2 +" , p2 finished"); } catch (InterruptedException e){}});
p1.start();
p2.start();

//get p2 data , p2 finished
//get p1 data , p1 finished

Phaser phaser = new Phaser();
Process p1 =  new Process(0,()-> {
    try {
        System.out.println("p1 register");
        phaser.register();
        while (true) {
            Thread.sleep(1000);
            System.out.println("p1 stage " + phaser.arriveAndAwaitAdvance());
        }
    } catch (InterruptedException e) {
    }
});
Process p2 =  new Process(0,()-> {
    try {
        System.out.println("p2 register");
        phaser.register();
        while (true) {
            Thread.sleep(2000);
            System.out.println("p2 stage " + phaser.arriveAndAwaitAdvance());
        }
    } catch (InterruptedException e) {
    }
});
Process p3 =  new Process(0,()-> {

    try {
        System.out.println("p3 register");
        phaser.register();
        while (true) {
            Thread.sleep(5000);
            System.out.println("p3 stage " + phaser.arriveAndAwaitAdvance());
        }
    } catch (InterruptedException e) {
    }
});
p1.start();
p2.start();
p3.start();

//p1 register
//p2 register
//p3 register
//p3 stage 1
//p2 stage 1
//p1 stage 1
//p3 stage 2
//p2 stage 2
//p1 stage 2
//...

//使用 parallelStream 并发执行
int[] sum = new int[]{0};
Stream.iterate(1, t-> t +1).limit(100) //1~100 累加
    .parallel()  //并行执行，即多线程执行
    .forEach(num -> sum[0] += num);
System.out.println(sum[0]); //大概率不是5050，因为加法不是原子性的

//单线程执行
int sum2 = Stream.iterate(1, t-> t +1).limit(100)
    .reduce(0,(a,b)-> a+b); //单线程执行
System.out.println(sum2); //5050，没错，因为是单线程执行，结果一定是5050

//并发执行，需要注意这里加法是满足结合律的，在累加里面，0即幺元 a + 0 = a , 0 + a = a 
int sum3 = Stream.iterate(1, t-> t +1).limit(100)
    .parallel()
    .reduce(0,(a,b)-> a+b);
System.out.println(sum3); //一定是5050，这里便是 monoid 的并发应用

Integer max = Arrays.asList(new Integer[]{32,23,66,12,68,11,22})
    .parallelStream()
    .max(Integer::compare) //这里返回的是Optional<Integer> 
    .get();
System.out.println(max); //68