JUC中提供了几个比较经常使用的并发工具类,好比CountDownLatch、CyclicBarrier、Semaphore。 其实在之前咱们 课堂的演示代码中,或多或少都有用到过这样一些api,接下来咱们会带你们去深刻研究一些经常使用的api。 CountDownLatch
countdownlatch是一个同步工具类,它容许一个或多个线程一直等待,直到其余线程的操做执行完毕再执行。从 命名能够解读到countdown是倒数的意思,相似于咱们倒计时的概念。 countdownlatch提供了两个方法,一个是countDown,一个是await, countdownlatch初始化的时候须要传入一 个整数,在这个整数倒数到0以前,调用了await方法的程序都必需要等待,而后经过countDown来倒数。node
public static void main(String[] args) throws InterruptedException { CountDownLatch countDownLatch=new CountDownLatch(3); new Thread(()->{ countDownLatch.countDown(); },"t1").start(); new Thread(()->{ countDownLatch.countDown(); },"t2").start(); new Thread(()->{ countDownLatch.countDown(); },"t3").start(); countDownLatch.await(); System.out.println("全部线程执行完毕"); }
从代码的实现来看,有点相似join的功能,可是比join更加灵活。CountDownLatch构造函数会接收一个int类型的 参数做为计数器的初始值,当调用CountDownLatch的countDown方法时,这个计数器就会减一。 经过await方法去阻塞去阻塞主流程
使用场景
1. 经过countdownlatch实现最大的并行请求,也就是可让N个线程同时执行,这个我也是在课堂上写得比较多 的
2. 好比应用程序启动以前,须要确保相应的服务已经启动,好比咱们以前在讲zookeeper的时候,经过原生api连 接的地方有用到countDownLatch 源码分析
CountDownLatch类存在一个内部类Sync,上节课咱们讲过,它是一个同步工具,必定继承了 AbstractQueuedSynchronizer。很显然,CountDownLatch其实是是使得线程阻塞了,既然涉及到阻塞,就一 定涉及到AQS队列 await
await函数会使得当前线程在countdownlatch倒计时到0以前一直等待,除非线程别中断;从源码中能够得知await 方法会转发到Sync的acquireSharedInterruptibly
方法
public void await() throws InterruptedException { sync.acquireSharedInterruptibly(1); } acquireSharedInterruptibly
这块代码主要是判断当前线程是否获取到了共享锁; 上一节课提到过,AQS有两种锁类型,一种是共享锁,一种是 独占锁,在这里用的是共享锁; 为何要用共享锁,由于CountDownLatch能够多个线程同时经过。api
public final void acquireSharedInterruptibly(int arg) throws InterruptedException { doAcquireSharedInterruptibly 获取共享锁 if (Thread.interrupted()) //判断线程是否中断 throw new InterruptedException(); if (tryAcquireShared(arg) < 0) //若是等于0则返回1,不然返回-1,返回-1表示须要阻塞 doAcquireSharedInterruptibly(arg); } 在这里,state的意义是count,若是计数器为0,表示不须要阻塞,不然,只有在知足条件的状况下才会被唤醒
doAcquireSharedInterruptibly
获取共享锁并发
private void doAcquireSharedInterruptibly(int arg) throws InterruptedException { final Node node = addWaiter(Node.SHARED); //建立一个共享模式的节点添加到队列中 boolean failed = true; try { for (;;) { //自旋等待共享锁释放,也就是等待计数器等于0。 final Node p = node.predecessor(); //得到当前节点的前一个节点 if (p == head) { int r = tryAcquireShared(arg);//就判断尝试获取锁 if (r >= 0) {//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); } }
待续。。。。。。。。。函数