juc-并发工具类源码解析

CountDownLatch

应用

countDownLatch就是使一个线程在其他线程都执行完之后再执行
CountDownLatch提供了一个构造函数，入参是一个int类型的变量；构造函数中，完成的事情是：把入参的值调用setState(int i);方法

public class  CountDownLatchTest {public static void main(String[] args) throws Exception {CountDownLatch countDownLatch = new CountDownLatch(5);for (int i = 0; i < 16; i++) {new Thread(() -> {try {TimeUnit.SECONDS.sleep(2);} catch (InterruptedException e) {e.printStackTrace();}System.out.println(Thread.currentThread().getName()+"\t"+"离开了");//countDown()表示减1countDownLatch.countDown();},String.valueOf(i)).start();}//await会阻塞，只有countDownLatch变为0时，才会执行下面的方法countDownLatch.await();System.out.println(Thread.currentThread().getName()+"\t"+"可以关灯");}
}/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/bin/java -javaagent:/Applications/IntelliJ IDEA.app/Contents/lib/idea_rt.jar=63442:/Applications/IntelliJ IDEA.app/Contents/bin -Dfile.encoding=UTF-8 -classpath /Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/jre/lib/charsets.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/jre/lib/deploy.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/jre/lib/ext/cldrdata.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/jre/lib/ext/dnsns.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/jre/lib/ext/jaccess.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/jre/lib/ext/jfxrt.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/jre/lib/ext/localedata.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/jre/lib/ext/nashorn.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/jre/lib/ext/sunec.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/jre/lib/ext/sunjce_provider.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/jre/lib/ext/sunpkcs11.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/jre/lib/ext/zipfs.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/jre/lib/javaws.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/jre/lib/jce.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/jre/lib/jfr.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/jre/lib/jfxswt.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/jre/lib/jsse.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/jre/lib/management-agent.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/jre/lib/plugin.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/jre/lib/resources.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/jre/lib/rt.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/lib/ant-javafx.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/lib/dt.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/lib/javafx-mx.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/lib/jconsole.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/lib/packager.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/lib/sa-jdi.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_221.jdk/Contents/Home/lib/tools.jar:/Volumes/mpy/studyWorkspace/JUC/target/classes:/Volumes/mpy/software/mavenRepository/junit/junit/4.12/junit-4.12.jar:/Volumes/mpy/software/mavenRepository/org/hamcrest/hamcrest-core/1.3/hamcrest-core-1.3.jar:/Volumes/mpy/software/mavenRepository/org/openjdk/jol/jol-core/0.9/jol-core-0.9.jar com.juc.tool.CountDownLatchTest
5   离开了
7   离开了
0   离开了
3   离开了
9   离开了
main    可以关灯
10  离开了
1   离开了
4   离开了
6   离开了
2   离开了
8   离开了
11  离开了
12  离开了
15  离开了
14  离开了
13  离开了

await()

public final void acquireSharedInterruptibly(int arg)throws InterruptedException {if (Thread.interrupted())throw new InterruptedException();// await()方法，会先调用tryAcquireShared方法，这里面的判断逻辑也简单  // return (getState() == 0) ? 1 : -1;也就是说，如果state不为0，就调用下面的方法进行排队if (tryAcquireShared(arg) < 0)doAcquireSharedInterruptibly(arg);}//这个排队的逻辑就不细说了，大致的意思就是:private void doAcquireSharedInterruptibly(int arg)throws InterruptedException {//1.如果当前线程是AQS队列中第一个排队的线程，就new一个空节点，然后把当前节点加入到空节点后面final Node node = addWaiter(Node.SHARED);boolean failed = true;try {for (;;) {final Node p = node.predecessor();//如果当前节点的上一个节点是head,表示当前线程是第一个排队的线程，就尝试加锁；// 这里所谓的加锁，就是执行这行代码 return (getState() == 0) ? 1 : -1;if (p == head) {int r = tryAcquireShared(arg);//如果r>0;表示当前state为0，可以执行await()方法对应的线程中的其他业务代码；// 就会把自己设置为头节点(thread设置为null，prev设置为null)if (r >= 0) {setHeadAndPropagate(node, r);p.next = null; // help GCfailed = false;return;}}//如果加锁失败，就执行这里，将上一个节点的waitStatus设置为-1，然后调用park()方法休眠if (shouldParkAfterFailedAcquire(p, node) &&parkAndCheckInterrupt())throw new InterruptedException();}} finally {if (failed)cancelAcquire(node);}

countDown()

countDown()方法，整体上，我们可以理解为是来将state-1，然后知道该值为0的时候，将阻塞在队列中的线程唤醒，继续执行主线程的逻辑

//就是一个类似于释放锁的过程，只不过，在tryReleaseShared(arg);方法中，是将当前的state -1;
// 然后通过cas，回写到state变量中；这个方法返回的是一个boolean值，boolean值的判断条件是：
//当前state是否为0；如果为0，就返回true；
//如果这里返回true，就唤醒AQS队列中的等待节点(这时候，理论上是countDownLatch.await();
//方法对应的线程在队列中)；也就是说，只有在state变为0的时候，才会执行countDownLatch.await()对应的线程中的方法public final boolean releaseShared(int arg) {if (tryReleaseShared(arg)) {doReleaseShared();return true;}return false;}/**
* 在调用countDown()方法的时候，会调用到这里：
*  将当前AQS的state值 - 1
*  如果未减1之前，state已经是0了，就直接返回false(表示这时候已经被其他线程唤醒了)，阻塞
*  如果 -1之后的值等于0，就返回true，就尝试唤醒阻塞队列中的线程
* @param releases
* @return
*/
protected boolean tryReleaseShared(int releases) {// Decrement count; signal when transition to zerofor (;;) {int c = getState();if (c == 0)return false;int nextc = c-1;if (compareAndSetState(c, nextc))return nextc == 0;}
}

doReleaseShared()方法我就不贴代码了，直接截图吧

CyclicBarrier

CyclicBarrier是做减法，在初始化的时候，指定count的阈值和达到阈值之后要执行的代码
每次调用await方法的时候，会将count-1；
如果减1之后的count不为0，就将线程阻塞(休眠)；如果调用await()指定了休眠时间(等待时间)，就将线程await指定的时间
如果count为0，就执行初始化时，指定的command，然后将所有等待的线程唤醒，重新对count进行赋值，再开始一轮
如果await方法指定了等待时间，那就休眠指定的时间，时间到了之后，线程依旧还在阻塞（count还没有变为0），那就会唤醒所有线程，然后抛出超时的异常

应用

public class CyclicBarrierTest {public static void main(String[] args) {CyclicBarrier cyclicBarrier = new CyclicBarrier(6,()->{System.out.println("可以执行该语句");});for (int i = 0; i < 15; i++) {final int temp = i;new Thread(() -> {System.out.println(Thread.currentThread().getName()+"执行第"+temp+"次");try {try {//                        cyclicBarrier.await(2L, TimeUnit.SECONDS);cyclicBarrier.await();} catch (Exception e) {e.printStackTrace();}} catch (Exception e) {e.printStackTrace();}},String.valueOf(i)).start();}}
}0执行第0次
1执行第1次
2执行第2次
3执行第3次
4执行第4次
5执行第5次
可以执行该语句
6执行第6次
7执行第7次
8执行第8次
9执行第9次
10执行第10次
11执行第11次
可以执行该语句
12执行第12次
13执行第13次
14执行第14次

初始化

/* parties:只有达到这个值，才会执行barrierAction的代码
* 这里的count是来做计数的，在每次await的时候，是修改count的值，来判断什么时候，执行runnable,再count变为0的时候，就执行，
* 执行完之后，再对count的值进行重新赋值，赋值为parties
*/
public CyclicBarrier(int parties, Runnable barrierAction) {if (parties <= 0) throw new IllegalArgumentException();this.parties = parties;this.count = parties;this.barrierCommand = barrierAction;
}

阻塞：await()

在调用await的时候，无论是否指定了等待时间，都会调用到该方法
如果指定了等待时间，这里入参的timed是true，且nanos就是指定的等待时间
如果没有指定等待时间，这里入参的timed是false，nanos是null

/**
* Main barrier code, covering the various policies.
*
* 这里的timed如果调用await()方法的时候，默认是false
* 如果调用了await的时候，指定了超时时间，timed就是true
*
* 调用await指定超时时效和不指定的区别是：
* 如果指定了：在超过这个时效之后，还是没有被唤醒，就会尝试唤醒所有线程，并且抛出异常
*/
private int dowait(boolean timed, long nanos)throws InterruptedException, BrokenBarrierException,
TimeoutException {final ReentrantLock lock = this.lock;lock.lock();try {final Generation g = generation;/*** 如果 当前generation被中止 ，就抛出异常*/if (g.broken)throw new BrokenBarrierException();/*** 如果线程被中断：* generation.broken 将该属性设置为true* 唤醒所有的等待线程* 并抛出异常*/if (Thread.interrupted()) {breakBarrier();throw new InterruptedException();}/*** 将计数器 - 1*/int index = --count;/*** 如果await调用的次数达到了设置的阈值，就执行下面的run()方法*/if (index == 0) {  // trippedboolean ranAction = false;try {final Runnable command = barrierCommand;if (command != null)command.run();/*** 执行完毕之后，会唤醒所有等待的线程：nextGeneration()* 并且会new 一个generation对象* 并将count计数重新设置为最初设置的阈值** ranAction为true，表示正常执行完毕， 如果执行业务逻辑的时候，报错，这时候  ranAction应该还是false*/ranAction = true;nextGeneration();return 0;} finally {/*** 如果ranAction为false，表示未未正常执行完逻辑* 这时候：就尝试唤醒所有的线程，确保在任务未执行成功时，将所有线程唤醒*/if (!ranAction)breakBarrier();}}// loop until tripped, broken, interrupted, or timed out/*** 如果-1之后的state不为0，就会进入到这里的for循环*/for (;;) {try {/*** 如果调用await()时，指定了时间，这里time的就是true;* 如果没有指定超时时间，这里就是false，就会一直等待，直到阈值达到指定的数量时，唤醒所有阻塞的线程** 如果指定了超时时间，那就等待响应的时间*/if (!timed)trip.await();else if (nanos > 0L)nanos = trip.awaitNanos(nanos);} catch (InterruptedException ie) {/*** 如果在等待的过程中发生了异常* 那就尝试唤醒所有的线程*/if (g == generation && ! g.broken) {breakBarrier();throw ie;} else {// We're about to finish waiting even if we had not// been interrupted, so this interrupt is deemed to// "belong" to subsequent execution.Thread.currentThread().interrupt();}}if (g.broken)throw new BrokenBarrierException();/*** 如果generation已经创建了一个新的，就返回index*/if (g != generation)return index;/*** 如果设置了超时时间并且达到了超时时效，就停止CyclicBarrier，并且唤醒所有等待的线程* 并抛出异常*/if (timed && nanos <= 0L) {breakBarrier();throw new TimeoutException();}}} finally {lock.unlock();}
}

Semaphore

semaphore是信号量，表示控制线程数量的，可以简单这样理解
类似于停车场抢占车位，停车场有三个车位，10辆车来了，前三辆会先进行占用，后面7辆会阻塞排队，等前三辆，出来一辆，后面排队的就会进入一个
Semaphore是基于AQS的共享锁实现的，初始化的时候，指定入参的数值，表示同一时间可以访问共享资源的线程数，内部是通过AQS的state变量来控制的

应用

public class SemaphoreTest {public static void main(String[] args) {Semaphore semaphore = new Semaphore(3);for (int i = 0; i < 7; i++) {new Thread(() -> {try {semaphore.acquire();System.out.println(Thread.currentThread().getName() + "\t" + "获取到了资源");TimeUnit.SECONDS.sleep(3);} catch (InterruptedException e) {e.printStackTrace();} finally {semaphore.release();}},String.valueOf(i)).start();}}
}0  获取到了资源
2   获取到了资源
1   获取到了资源3 获取到了资源
5   获取到了资源
4   获取到了资源6 获取到了资源

初始化

首先，Semaphore是基于AQS的共享锁实现的，初始化的时候，指定入参的数值，表示同一时间可以访问共享资源的线程数，内部是通过AQS的state变量来控制的
内部也分为公平锁和非公平锁，两者的区别和reentrantLocK的公平锁、非公平锁一样，在抢占资源的时候，少了一个判断：是否需要排队；非公平锁在抢占资源的时候，无需判断是否需要排队，直接尝试cas，抢占成功，就无须排队，否则就去排队

Semaphore semaphore = new Semaphore(3);
这行代码会初始化一个公平锁或者非公平锁，入参指定的变量是锁对应的state，也就是最多可以有多少个线程来加锁；构造函数最终会调用这里，将state设置为3

Sync(int permits) {setState(permits);
}

抢占资源：acquire()

这里不指定入参，默认每次抢占一个资源，可以指定一个线程抢占的资源数；对于底层来讲，没有区别，无非就是在尝试加锁的时候，将state-指定的数量即可，如果不指定，默认是-1

public void acquire() throws InterruptedException {sync.acquireSharedInterruptibly(1);
}public final void acquireSharedInterruptibly(int arg)throws InterruptedException {if (Thread.interrupted())throw new InterruptedException();// 这里返回值小于0，表示需要去排队if (tryAcquireShared(arg) < 0)doAcquireSharedInterruptibly(arg);
}protected int tryAcquireShared(int acquires) {for (;;) {/*** 这里是尝试判断是否需要进行排队，如果返回false，表示可以抢占资源*/if (hasQueuedPredecessors())return -1;int available = getState();int remaining = available - acquires;/*** 这里在将state减去指定的数量之后，如果remaining小于0，就表示资源不够本次线程来加锁，就会去排队* 如果remaining还大于等于0，就表示当前剩余的资源，可以满足本次线程的加锁，就进行cas*/if (remaining < 0 ||compareAndSetState(available, remaining))return remaining;}
}

排队

/**
* semaphore在抢占资源失败，会来这里，进行排队
* 1.生成node节点
* 2.判断当前node节点是否是第二个节点(也即第一个排队的节点)，如果是，尝试抢占资源，抢占成功，就无须排队，return
* 3.抢占失败，或者不是第一个排队的节点，就将前一个节点的waitStatus设置为-1，然后调用park()方法
* @param arg
* @throws InterruptedException
*/
private void doAcquireSharedInterruptibly(int arg)throws InterruptedException {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 GCfailed = false;return;}}if (shouldParkAfterFailedAcquire(p, node) &&parkAndCheckInterrupt())throw new InterruptedException();}} finally {if (failed)cancelAcquire(node);}
}

释放资源：semaphore.release();

1、将state变量 + 1；如果这里release()不指定数量，默认是1；也可以指定一次要释放多少个资源
2、通过cas将state变量更新成功之后，尝试唤醒排队队列中的线程
这里的这个方法是aqs中的，都是来释放资源或者解锁的
tryReleaseShared：是尝试释放资源或者解锁，由于每个工具类解锁或者释放资源逻辑不同，所以这个方法放到子类中进行实现
doReleaseShared：这是尝试唤醒队列中排队的线程，这个方法是共用的，就可以直接在父类中实现
这就是模板设计模式，aqs中大量用到了该设计模式

public final boolean releaseShared(int arg) {// 尝试释放资源，释放资源成功，就唤醒排队线程if (tryReleaseShared(arg)) {doReleaseShared();return true;}return false;
}/**
* semaphore尝试释放占用的资源，将state + release
* @param releases
* @return
*/
protected final boolean tryReleaseShared(int releases) {for (;;) {int current = getState();int next = current + releases;if (next < current) // overflowthrow new Error("Maximum permit count exceeded");if (compareAndSetState(current, next))return true;}
}

doReleaseShared()方法，在前面countDownLatch中有调用，我就不贴源码了

所以：semaphore的流程是这样的：
1.在创建semaphore的时候，指定同一时刻，允许的最大信号量
2.多个线程会尝试占有资源，占有资源的过程是：将state-1，判断state是否小于0，然后通过cas，将-1后的state写到内存中
3.如果线程抢占资源时，返回的state-1 < 0;表示当前信号量资源已经被占完，需要进行排队
4.在排队的时候，会先创建AQS中排队的node节点，如果当前是第一个排队的线程，就创建一个空节点，然后把刚创建的node节点，添加到空节点的后面；否则，就直接在AQS队列中，添加node节点即可；然后接下来，会区分两种情况
4.1 如果当前排队的是AQS队列中第一个排队的线程，在插入队列之后，会进行一次尝试加锁的过程，因为有可能这个线程在处理排队的过程中，已经有线程释放了资源；
如果抢占资源成功，就把当前线程对应的node节点设置为头结点(thread设置为null；node.prev设置为null)；然后唤醒下一个waitStatus为-1的节点(正常情况下，就是下一个节点)
如果抢占失败，就执行4.2的逻辑
4.2 如果抢占失败，或者是非第一个排队线程，就执行以下逻辑：判断上一个节点的waitStatus是否为-1，如果不为-1，就将上一个node节点的waitStatus设置为-1，然后unpark()

5.在释放资源的时候，会获取到当前的state，然后+1，通过cas，写到state变量中；然后就唤醒下一个waitStatus为-1的节点；如果head == tail；表示当前队列中，已经没有排队的线程，就无需唤醒，直接return即可

6.在使用Semapore的时候，创建公平锁和非公平锁的唯一区别是：
在尝试占有资源的时候，非公平锁，会直接cas；而公平锁会先判断是否需要排队，如果需要排队，就返回-1，然后进行排队，不需要排队，就cas自旋