并发专题(7)之JUC并发工具源码分析

一、CountdownLatch源码分析

1.1 CountdownLatch应用

CountDownLatch本身就好像一个计数器，可以让一个线程或多个线程等待其他线程完成后再执行。

public static void main(String[] args) throws InterruptedException, BrokenBarrierException {
    // 声明CountDownLatch，有参构造传入的值，会赋值给state，CountDownLatch基于AQS实现
    // 3 - 1 = 2 - 1 = 1 - 1
    CountDownLatch countDownLatch = new CountDownLatch(3);

    new Thread(() -> {
        System.out.println("111");
        countDownLatch.countDown();
    }).start();

    new Thread(() -> {
        System.out.println("222");
        countDownLatch.countDown();
    }).start();

    new Thread(() -> {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("333");
        countDownLatch.countDown();
    }).start();

    // 主线会阻塞在这个位置，直到CountDownLatch的state变为0
    countDownLatch.await();
    System.out.println("main");
}

1.2 CountdownLatch构造函数源码剖析

// CountDownLatch 的有参构造
public CountDownLatch(int count) {
    // 健壮性校验
    if (count < 0) throw new IllegalArgumentException("count < 0");
    // 构建Sync给AQS的state赋值
    this.sync = new Sync(count);
}

1.3 countDown源码剖析

// countDown本质就是调用了AQS的释放共享锁操作
public void countDown() {
    sync.releaseShared(1);
}
public final boolean releaseShared(int arg) {
    if (tryReleaseShared(arg)) {
        // 唤醒在AQS中等待的线程
        doReleaseShared();
        return true;
    }
    return false;
}

protected boolean tryReleaseShared(int releases) {
    // Decrement count; signal when transition to zero
    for (;;) {
        int c = getState();
        if (c == 0)
            return false;
        int nextc = c-1;
        // 通过CAS修改state的值
        if (compareAndSetState(c, nextc))
            // 判断state的值是否已经等于0，如果是等于0的话，那么直接doReleaseShared唤醒线程
            return nextc == 0;
    }
}

// 如果CountDownLatch中的state已经为0了，那么再次执行countDown跟没执行一样。
// 而且只要state变为0，await就不会阻塞线程。

private void doReleaseShared() {
    for (;;) {
        // 获取头结点
        Node h = head;
        // 判断是否有线程等待
        if (h != null && h != tail) {
            int ws = h.waitStatus;
            // 如果头结点ws为-1说明后继节点存在等待
            if (ws == Node.SIGNAL) {
                if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                    continue;            // loop to recheck cases
                // 唤醒后继线程
                unparkSuccessor(h);
            }
            // 这里是为了修复jdk1.5中的bug，后续会说明
            else if (ws == 0 &&
                     !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                continue;                // loop on failed CAS
        }
        if (h == head)                   // loop if head changed
            break;
    }
}

1..4 await源码剖析

public void await() throws InterruptedException {
    // 调用AQS提供的获取共享锁并且允许中断的方法
    sync.acquireSharedInterruptibly(1);
}
public final void acquireSharedInterruptibly(int arg)
        throws InterruptedException {
    // 允许线程中断
    if (Thread.interrupted())
        throw new InterruptedException();
    if (tryAcquireShared(arg) < 0)
        // 如果返回的是-1，代表state肯定是大于0
        doAcquireSharedInterruptibly(arg);
}
// 判断state是否为0，如果为0则直接结束返回
protected int tryAcquireShared(int acquires) {
    return (getState() == 0) ? 1 : -1;
}

// countDownLatch实现
private void doAcquireSharedInterruptibly(int arg)
    throws InterruptedException {
    // 封装当前线程为node节点，并且加入到aqs队列中
    final Node node = addWaiter(Node.SHARED);
    boolean failed = true;
    try {
        for (;;) {
            final Node p = node.predecessor();
            if (p == head) {
                // 继续判断state是否为0
                int r = tryAcquireShared(arg);
                if (r >= 0) {
                    // 会将当前线程和后面所有排队的线程都唤醒
                    setHeadAndPropagate(node, r);
                    p.next = null; // help GC
                    failed = false;
                    return;
                }
            }
            // 跟lock一样，挂起线程
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                throw new InterruptedException();
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

二、Samaphore源码分析

2.1 Samaphore应用

也是常用的JUC并发工具，一般用于流控。比如有一个公共资源，多线程都可以访问时，可以用信号量做限制。

连接池，内部的链接对象有限，每当有一个线程获取连接对象时，对信号量-1，当这个线程归还资源时对信号量+1。如果线程拿资源时，发现Semaphore内部的资源个数为0，就会被阻塞。例如：Hystrix的隔离策略 - 线程池，信号量。

public static void main(String[] args) throws InterruptedException, BrokenBarrierException {
    // 声明信号量
    Semaphore semaphore = new Semaphore(1);
    // 能否去拿资源
    semaphore.acquire();
    // 拿资源处理业务
    System.out.println("main");
    // 归还资源
    semaphore.release();
}

2.2 有参构造

    public Semaphore(int permits) {
        // 默认非公平锁实现
        sync = new NonfairSync(permits);
    }

    /**
     * Creates a {@code Semaphore} with the given number of
     * permits and the given fairness setting.
     *
     * @param permits the initial number of permits available.
     *        This value may be negative, in which case releases
     *        must occur before any acquires will be granted.
     * @param fair {@code true} if this semaphore will guarantee
     *        first-in first-out granting of permits under contention,
     *        else {@code false}
     */
    public Semaphore(int permits, boolean fair) {
        // 设置资源个数，State其实就是信号量的资源个数
        sync = fair ? new FairSync(permits) : new NonfairSync(permits);
    }

2.3 acquire源码剖析

public void acquire(int permits) throws InterruptedException {
    if (permits < 0) throw new IllegalArgumentException();
    sync.acquireSharedInterruptibly(permits);
}


public final void acquireSharedInterruptibly(int arg)
        throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    if (tryAcquireShared(arg) < 0)
        doAcquireSharedInterruptibly(arg);
}
// 公平锁实现
protected int tryAcquireShared(int acquires) {
    for (;;) {
        // 上来就先判断队列是否有线程在等待，如果有那就取排队
        if (hasQueuedPredecessors())
            return -1;
        int available = getState();
        int remaining = available - acquires;
        // 通过CAS去修改state的值，并且返回还剩余的state
        if (remaining < 0 ||
            compareAndSetState(available, remaining))
            return remaining;
    }
}

protected int tryAcquireShared(int acquires) {
    return nonfairTryAcquireShared(acquires);
}
// 非公平锁实现
final int nonfairTryAcquireShared(int acquires) {
    for (;;) {
        int available = getState();
        int remaining = available - acquires;
        if (remaining < 0 ||
            compareAndSetState(available, remaining))
            return remaining;
    }
}
// 拿不到资源那就挂起线程
private void doAcquireSharedInterruptibly(int arg)
    throws InterruptedException {
    // 将当前线程封装到node并且加入到aqs队列中
    final Node node = addWaiter(Node.SHARED);
    boolean failed = true;
    try {
        for (;;) {
            // 拿到前继节点
            final Node p = node.predecessor();
            if (p == head) {
                // 获取剩余的资源
                int r = tryAcquireShared(arg);
                if (r >= 0) {
                    // 唤醒当前节点以及后继节点
                    setHeadAndPropagate(node, r);
                    p.next = null; // help GC
                    failed = false;
                    return;
                }
            }
            // 挂起线程
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                throw new InterruptedException();
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

2.4 release源码剖析

public void release() {
    sync.releaseShared(1);
}
public final boolean releaseShared(int arg) {
    if (tryReleaseShared(arg)) {
        // 唤醒形成去竞争资源
        doReleaseShared();
        return true;
    }
    return false;
}
protected final boolean tryReleaseShared(int releases) {
    for (;;) {
        int current = getState();
        // state +1
        int next = current + releases;
        // state +1 等于负数，最大值+1为负数表示超过最大值
        if (next < current) // overflow
            throw new Error("Maximum permit count exceeded");
        // 通过CAS一定要把state +1
        if (compareAndSetState(current, next))
            return true;
    }
}

2.5 分析AQS中的PROPAGATE类型节点的作用

在JDK1.5中，使用信号量时，可能会造成在有资源的情况下，后继节点无法呗唤醒，如下流程：

在JDK1.8中，问题被修复，修复方式就是追加了PROPAGATE节点状态来解决。

共享锁在释放资源后，如果头节点为0，无法确认真的没有后继节点。如果头节点为0，需要将头节点的状态修改为-3，当最新拿到锁资源的线程，查看是否有后继节点并且为共享锁，就唤醒排队的线程【这里需要对比JDK1.5中的实现来看才能理解这个场景】。

jdk8多了以下操作。

三、CyclicBarrier源码分析

一般称为栅栏，和CountDownLatch很像。 CountDownLatch在操作时，只能使用一次，也就是state变为0之后，就无法继续玩了。 CyclicBarrier是可以复用的，他的计数器可以归位，然后再处理。而且可以在计数过程中出现问题后，重置当前CyclicBarrier，再次重新操作！

public static void main(String[] args) throws InterruptedException, BrokenBarrierException {
    // 声明栅栏
    CyclicBarrier barrier = new CyclicBarrier(3,() -> {
        System.out.println("打手枪！");
    });

    new Thread(() -> {
        System.out.println("第一位选手到位");
        try {
            barrier.await();
            System.out.println("第一位往死里跑！");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }).start();

    new Thread(() -> {
        System.out.println("第二位选手到位");
        try {
            barrier.await();
            System.out.println("第二位也往死里跑！");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }).start();

    System.out.println("裁判已经到位");
    barrier.await();
}

3.1 有参构造

CyclicBarrier没有直接使用AQS，而是使用ReentrantLock，简介的使用的AQS。

// CyclicBarrier的有参
public CyclicBarrier(int parties, Runnable barrierAction) {、
    // 健壮性判断！
    if (parties <= 0) throw new IllegalArgumentException();
    // parties是final修饰的，需要在重置时，使用！
    this.parties = parties;
    // count是在执行await用来计数的。
    this.count = parties;
    // 当计数count为0时 ，先执行这个Runnnable！在唤醒被阻塞的线程
    this.barrierCommand = barrierAction;
}

3.2 await源码剖析

线程执行await方法，会对count-1，再判断count是否为0，如果不为0，需要添加到AQS中的ConditionObject的Waiter队列中排队，并park当前线程。

如果为0，证明线程到齐，需要执行nextGeneration，会先将Waiter队列中的Node全部转移到AQS的队列中，并且有后继节点的，ws设置为-1。没有后继节点设置为0。然后重置count和broker标记。等到unlock执行后，每个线程都会被唤醒。

// 选手到位！！！
private int dowait(boolean timed, long nanos) throws InterruptedException, BrokenBarrierException, TimeoutException {
    // 加锁？？  因为CyclicBarrier是基于ReentrantLock-Condition的await和singalAll方法实现的。
    // 相当于synchronized中使用wait和notify
    // 别忘了，只要挂起，会释放锁资源。
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        // 里面就是boolean，默认false
        final Generation g = generation;

        // 判断之前栅栏加入线程时，是否有超时、中断等问题，如果有，设置boolean为true，其他线程再进来，直接凉凉
        if (g.broken)
            throw new BrokenBarrierException();

        if (Thread.interrupted()) {
            breakBarrier();
            throw new InterruptedException();
        }


        // 对计数器count--
        int index = --count;
        // 如果--完，是0，代表突破栅栏，干活！
        if (index == 0) {  
            // 默认false
            boolean ranAction = false;
            try {
                // 如果你用的是2个参数的有参构造，说明你传入了任务，index == 0，先执行CyclicBarrier有参的任务
                final Runnable command = barrierCommand;
                if (command != null)
                    command.run();
                // 设置为true
                ranAction = true;
                nextGeneration();
                return 0;
            } finally {
                if (!ranAction)
                    breakBarrier();
            }
        }

        // --完之后，index不是0，代表还需要等待其他线程
        for (;;) {
            try {
                // 如果没设置超时时间。  await()
                if (!timed)
                    trip.await();
                // 设置了超时时间。  await(1,SECOND)
                else if (nanos > 0L)
                    nanos = trip.awaitNanos(nanos);
            } catch (InterruptedException ie) {
                if (g == generation && ! g.broken) {
                    breakBarrier();
                    throw ie;
                } else {
                    Thread.currentThread().interrupt();
                }
            }

            if (g.broken)
                throw new BrokenBarrierException();

            if (g != generation)
                return index;

            if (timed && nanos <= 0L) {
                breakBarrier();
                throw new TimeoutException();
            }
        }
    } finally {
        lock.unlock();
    }
}



// 挂起线程
public final void await() throws InterruptedException {
    // 允许中断
    if (Thread.interrupted())
        throw new InterruptedException();
    // 添加到队列（不是AQS队列，是AQS里的ConditionObject中的队列）
    Node node = addConditionWaiter();
    int savedState = fullyRelease(node);
    int interruptMode = 0;
    while (!isOnSyncQueue(node)) {
        // 挂起当前线程
        LockSupport.park(this);
        if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
            break;
    }
}


// count到0，唤醒所有队列里的线程线程
private void nextGeneration() {
    // 这个方法就是将Waiter队列中的节点遍历都扔到AQS的队列中，真正唤醒的时机，是unlock方法
    trip.signalAll();
    // 重置计数器
    count = parties;
    // 重置异常判断
    generation = new Generation();
}

3.3 总结

其实可以看到，上面已经把我们AQS中的node节点的每一个状态都说明了，1,-1，-2，-3，其中AQS中存在两个队列，其中是一个等待队列，另外一个是在AQS内部中的 ConditionObject维护的条件队列，当ConditionObject满足唤醒条件的时候，机会将条件队列的节点移动到AQS的等待队列中等待唤醒。