中国航天科工集团有限公司,seo企业优化顾问,做网站充值微信必须是企业,实时网站制作1、什么是锁#xff0c;为什么需要锁#xff1f;
并发环境下#xff0c;会存在多个线程对同一个资源进行争抢的情况#xff0c;假设线程A对资源正在进行修改#xff0c;此时线程B又对同一资源进行了修改#xff0c;就会导致数据不一致的问题。为了解决这个问题#xff…1、什么是锁为什么需要锁
并发环境下会存在多个线程对同一个资源进行争抢的情况假设线程A对资源正在进行修改此时线程B又对同一资源进行了修改就会导致数据不一致的问题。为了解决这个问题很多编程语言引入了锁机制。
通过一种抽象的“锁”来对资源进行锁定当一个线程持有“锁”时其他线程必须等待 ------ 在临界资源上对线程进行一种串行化处理。 2、悲观锁 VS 乐观锁
乐观锁和悲观锁是一种广义上的概念体现了看待线程并发时的角度。
悲观锁认为自己在使用数据的时候一定会有别的线程修改数据因此获取数据时会先加锁确保不会被别的线程修改乐观锁认为自己使用数据的时候不会有别的线程修改数据所以不会去加锁只是在更新数据的时候去判断有没有别的线程更新了这个数据如果没有被更新则自己执行相关操作如果被其他线程更新了则执行对应的操作重试或者报错 3、介绍一下悲观锁。
java中悲观锁的实现是基于object的 ------ 也就是说每个对象都拥有一把锁这个锁存放在对象头中记录了当前对象被哪个线程所占用持有。
3.1、对象结构对象头结构
首先说一下对象的结构
对象头实例数据填充字节为了满足java对象大小是8字节的整数倍没有实际意义
对象头结构
mark word --- 见下图class pointer --- 是一个指针指向当前对象类型所在方法区中的class信息
相比于实例数据对象头其实是一种额外的开销因此设计的很小32bit或者64bit 从上图可以看到“锁”的信息就存放在mark word中根据后两个标志位锁的状态又可以分为 无锁偏向锁轻量级锁重量级锁。--- java中启用对象锁的方式就是 synchronized 关键字。下面会详细介绍一下synchronized背后的原理。 3.2 synchronized关键字
在java中synchronized关键字可以用来加锁而synchronized被编译后会生成两个字节码指令monitorenter 和 monitorexit 这两个字节码指令底层是依赖于操作系统的 mutex lock 来实现的这个时候java线程实际就是操作系统线程的映射每当唤醒或者挂起一个线程的时候都需要切换到操作系统的内核态这个操作是重量级的换句话说就是比较耗费时间甚至切换的时间就超出了线程执行任务的时间。因此使用synchronized其实是会对程序的性能产生影响。 刚刚提到了monitor又叫监视器、管程等我们可以将它理解为只能容纳一个客人的饭店客人就是想要获取对象锁的线程一旦有一个线程进入了monitor那么其他线程只能等待只有当这个线程退出去其他线程才有机会进入。 3.3 介绍一下对象锁的四种状态
1、无锁
顾名思义就是没有对系统资源进行锁定1比如某种资源不会出现在多线程环境下或者即便在多线程环境下面也不会有竞争情况那么就不需要加锁。2)虽然会被竞争但是采用了其他机制来控制而不是使用操作系统对资源锁定CAS后面会介绍。 2、偏向锁
假如一个对象被加锁了但是实际运行过程中只有一个线程会获取这个对象锁前提最理想的方式是只在用户态就把锁交出去而不是通过操作系统来切换线程状态。 ---- 如果对象能够认识这个唯一的线程只要这个线程过来就直接把锁交出去就行了。对象偏爱这个线程称为偏向锁
偏向锁是怎么实现的呢 回到mark word 中当锁标志位为01的时候判断倒数第三个bit, 如果是1则表示当前对象锁的状态为偏向锁此时去读mark word的前23bit这记录的就是线程ID可以通过线程id来判断这个线程是不是被偏爱的线程。 3、轻量级锁
某一个时间情况发生了变化不止一个线程想要获取这个锁了两个线程这个时候偏向锁就会升级成为轻量级锁。
轻量级锁是如何实现的 --- Markword 后两位变成00的时候就意味着当前对象锁的状态是轻量级锁了此时不在用线程id了前30位变成了指向虚拟机栈中锁记录的指针。
当一个线程想要获取某个对象锁发现Markword 后两位是00此时会在自己的虚拟机栈中开辟一块成为 lock record 的空间这个lock record存放的是Markword 的副本以及一个owner指针
线程通过cas去尝试获取锁一旦获得就会复制这个对象的Markword到自身虚拟机栈的lock record中并且将lock record中的owner指针指向该对象锁。同时对象的mark word的前30位生成一个指针指向持有改对象锁的线程虚拟机栈中的lock record这样就完成了线程和对象锁的绑定双方都知道各自的存在了。
此时获得了对象锁的线程就可以执行对应的任务了那么没有获得锁的线程应该怎么办呢--- 没有获得锁的线程会自旋等待自旋就是一种轮询不断判断锁有没有被释放如果释放了就获取锁如果没有释放就下一轮循环。 --- 这种自旋区别于被操作系统挂起阻塞因为如果对象很快被释放的话自旋获取锁就可以在用户态解决而不用切换到内核态效率比较高。
但是自旋其实就是CPU在空转长时间的自旋会浪费CPU资源于是出现了一种叫做“适应性自旋”的优化。简单来说就是自旋时间不在固定了而是又上一次在同一个锁上的自旋时间以及锁状态来决定比如在同一个锁上当前自旋等待的线程刚刚成功获得过锁但是此时锁被其他线程持有虚拟机就会认为下次自旋获取锁的概率很大进而运行更长时间的自旋。 4、重量级锁
一旦自旋等待的线程超过一个即有三个及以上的线程想获取同一个锁这个时候就会升级成为重量级锁这个时候就是通过monitor来对线程进行控制了一旦进入了这个状态就会调用内核空间产生极大的开销。 上述描述了对象锁的四种状态需要注意的是锁只能升级不能降级。 4、介绍一下乐观锁
在讲述悲观锁的时候提到了“无锁”这个概念其中有一种是共享资源会出现被竞争的情况但是不是适用操作系统同步原语进行保护而是使用CAS这种方式进行线程同步尽量将获取锁释放锁的操作在用户空间内完成减少用户态和内核态之间的切换次数提升程序的性能。 --------- 可以看到其实乐观锁并不是锁而是一种无锁的实现。
CAS就是实现乐观锁的一种经典巧妙的算法。 compare and swap简单的翻译为比较然后交换。
4.1 如何理解CAS
举例比如说厕所的坑位里面没人的时候你才可以进去有人就只能在外面等着设定开门状态是0关门状态是1, 某一时间两个人都想上厕所A先冲了过去并把门关上这个时候B才过来但是发现门已经关了但是B也不能放弃啊就不断回来看看门打开了没。
上述例子中AB两个人就是代表线程坑位就是共享资源这样就应该比较容易理解CAS了当一个共享资源状态值为0的一瞬间AB线程读到了此时两个线程都认为当前这个共享资源没有被占用于是他们会各自生成两个值
old value代表之前读到的共享资源对象的状态值 -- 上述例子中都为0new value代表想要将共享资源对象状态值更新的值 -- 上述例子中都为1
此时AB线程都去争抢修改对象的状态值然后占用对象。假设A运气比较好 A将old value 和 资源对象的状态值进行compare 比较发现是一致的于是将对象的状态值 swap 为new value B落后一步compare的时候发现自己的old value 和对象的状态值不一样只能放弃swap操作一般是自旋操作同时配置自旋次数防止死循环默认值10 。 上述是一个CAS函数但是其实是有问题的因为这个函数本身没有任何同步措施还是存在不安全的问题因此想要通过 CAS实现乐观锁有一个必须的前提CAS操作本身必须是原子性的。
那么CAS是如何实现原子性的呢 --- 不同架构的CPU都提供了指令级的CAS原子操作比如X86架构下的cmpxchg指令ARM架构下的LL/SC指令。也就是说CPU已经原生的支持了CAS那么上层直接调用即可。
// 乐观锁的实现举例private static AtomicInteger num new AtomicInteger();
public void testCAS() {for (int i 0; i 3; i) {Thread thread new Thread(new Runnable() {Overridepublic void run() {while (num.get() 1000) {System.out.println(Thread.currentThread().getName() num.getAndIncrement());}}});}
}
通过AtomicIteger的源码发现是使用Unsafe类来实现CAS操作的这个cas方法是一个native函数说明是本地方法和具体的平台实现相关。 5、AQS 机制是什么
通过上述介绍CAS我们知道了java通过unsafe 类封装了CAS方法支持了对CAS原语的调用但是针对上层业务开发怎么能够无感知的调用并且业务场景中我们最常竞争的资源往往是通过对象进行封装的而CAS只能原始的修改内存上的一个值。如何进一步对CAS进行抽象呢
下面就首先介绍一下JUC中经典的同步框架AQS AbstractQueuedSynchronizer 5.1 、AQS的成员属性
state是用于判断共享资源是否正在占用的标记为volatile关键字保证了线程之间的可见性。至于为什么用int类型而不是布尔类型是因为AQS中有独占锁和共享锁的区别共享模式下state可以表示占用锁的线程的数量。AQS中还存在一个队列用于管理等待获取锁的线程。FIFO双向链表head和tail定义了头和尾 // 队列的头 尾部private transient volatile Node head;private transient volatile Node tail;// 判断共享资源的标志位private volatile int state;
5.2、AQS的内部类
AQS维护了一个FIFO队列因此定义了一个Node节点里面存储了线程对象 thread节点在队列中的等待状态 waitstatus前后指针等信息。
waitStatusAQS工作的时候必然伴随着Node 节点状态值的各种变化这里的waitStatus是一个枚举值 0节点初始化默认值或者节点已经释放锁1取消获取锁-1节点的后续节点需要被唤醒-2条件模式相关-3共享模式相关传递共享模式下锁的释放状态predecessor() 方法获取当前节点的前置Node // 简化的删除了很多东西static final class Node {static final int CANCELLED 1;static final int SIGNAL -1;static final int CONDITION -2;static final int PROPAGATE -3;volatile int waitStatus;volatile Node prev;volatile Node next;volatile Thread thread;final Node predecessor() throws NullPointerException {Node p prev;if (p null)throw new NullPointerException();elsereturn p;}} 5.3、AQS中的核心方法
一般业务可以分为两种场景使用锁
尝试获取锁不管有没有获取到立即返回必须获取锁没有获取到则进行等待
恰好AQS中针对上述两种场景提供了两个方法tryAcquire 和 acquire 两个方法。
5.3.1 tryAcquire -- 尝试获取锁
这个方法是参数是一个int 类型的值代表对 state的增加操作返回值是boolean代表是否成功获取锁。该方法只抛出了一个异常目的就是为了让子类进行重写在子类中定义相关的业务逻辑比如没有获取到锁是等待还是别的处理。 protected boolean tryAcquire(int arg) {throw new UnsupportedOperationException();} 5.3.2 acquire -- 必须获取锁
acquire方法被final修饰无法重写想要等待并获取锁直接调用这个方法就可以。 public final void acquire(int arg) {if (!tryAcquire(arg) acquireQueued(addWaiter(Node.EXCLUSIVE), arg))selfInterrupt();}
if条件包含了两个判断条件tryAcquire已经说过了现在讲一下acquireQueue方法。
【addWaiter】
首先看一下 addWaiter发方法 --- 这个方法作用就是将当前线程封装成一个node然后加入等待队列中。 private Node addWaiter(Node mode) {// 封装当前线程为一个NODE节点Node node new Node(Thread.currentThread(), mode);// 获取当前尾节点Node pred tail;if (pred ! null) {// 先将当前节点前指针指向尾节点node.prev pred;// CAS判断当前尾节点是否还是尾节点如果是就把当前节点变为尾节点然后返回出去if (compareAndSetTail(pred, node)) {pred.next node;return node;}}// 如果当前队列为空或者CAS失败就进入这个方法enq(node);return node;}private Node enq(final Node node) {// 自旋 如果队列没有初始化那么就初始化如果尾节点插入失败就不断重试直到插入为止for (;;) {Node t tail;if (t null) { // Must initializeif (compareAndSetHead(new Node()))tail head;} else {node.prev t;if (compareAndSetTail(t, node)) {t.next node;return t;}}}}
既然将node加入队列成功了后面肯定还会从队列中取出节点一般这种FIFO队列都会使用 “生产者-消费者”模式但是AQS却不是这么使用的我们接着往下看。 【acquireQueued】方法
首先定义了一个failed变量默认是true如果当前线程正常获取到了锁这个值就改为falsefinally语句块里面的方法只是为了解决异常情况下取消当前线程获取锁的行为在AQS中头节点head是个虚节点即队列中的第一个节点的前一个节点是头节点 首先只有队列中的第一个节点才有权限尝试获取锁如果获取到锁进入if中然后返回结果如果没有获取到锁就进入下一个判断。第二个if看判断条件从名字上来看首先判断当前线程是否需要挂起如果需要挂起就执行挂起操作如果不需要就继续自旋获取锁。【shouldParkAfterFailedAcquire】判断获取锁失败后是否需要挂起 final boolean acquireQueued(final Node node, int arg) {boolean failed true;try {boolean interrupted false;for (;;) {final Node p node.predecessor();// 如果是队列中的第一个节点并且获取到了锁就进入这个if中if (p head tryAcquire(arg)) {setHead(node);p.next null; // help GCfailed false;return interrupted;}// 判断是否需要挂起如果需要挂起就执行挂起操作否则下一次for循环if (shouldParkAfterFailedAcquire(p, node) parkAndCheckInterrupt())interrupted true;}} finally {if (failed)// 取消当前线程获取锁的行为。处理try中的异常情况cancelAcquire(node);}}private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {int ws pred.waitStatus;if (ws Node.SIGNAL)// ws为-1当前节点也在等待拿锁因此可以挂起休息return true;if (ws 0) {// 表示获取锁请求被取消了就从当前队列中删除do {node.prev pred pred.prev;} while (pred.waitStatus 0);pred.next node;} else {// 将当前节点状态改为-1然后外层重试compareAndSetWaitStatus(pred, ws, Node.SIGNAL);}return false;}private final boolean parkAndCheckInterrupt() {// 这一行就是执行了挂起操作。是通过unsafe的native方法操作系统原语LockSupport.park(this);// 被唤醒之后返回当前线程有没有被中断。return Thread.interrupted();} 通过对acquireQueued 方法的分析可以说这个方法就是将当前队列中的线程节点都挂起避免不必要的自旋浪费CPU资源。
既然有线程被挂起那么就需要将这些挂起的线程唤醒当持有锁的线程释放了锁那么就会尝试唤醒后续的节点AQS中提供了release方法用来唤醒挂起的线程。 5.3.3 tryRelease方法
和tryAcquire方法一样tryRelease方法也是开放给上层实现的。 protected boolean tryRelease(int arg) {throw new UnsupportedOperationException();}
5.3.4 release 方法
release方法中假设尝试释放锁成功下一步就要唤醒等待队列中的其他节点unparkSuccessor方法中传入的是头节点。
【unparkSuccessor】方法
设置头节点的状态为0表示已经释放了锁获取队列中最靠前的一个准备获取锁的节点状态不能是canceled取消获取锁将这个节点唤醒去争抢锁。 public final boolean release(int arg) {if (tryRelease(arg)) {Node h head;if (h ! null h.waitStatus ! 0)unparkSuccessor(h);return true;}return false;}private void unparkSuccessor(Node node) {// 获取头节点的状态如果不是0就修改为0表示锁已经释放了int ws node.waitStatus;if (ws 0)compareAndSetWaitStatus(node, ws, 0);// 获取头节点的后续节点Node s node.next;// 如果为空或者处于canceled状态那么就从后往前搜索找到除head外最靠前的nodeif (s null || s.waitStatus 0) {s null;for (Node t tail; t ! null t ! node; t t.prev)if (t.waitStatus 0)s t;}// 唤醒这个node让他起来尝试拿锁if (s ! null)LockSupport.unpark(s.thread);}
【注意】 unparkSuccessor中搜索节点时是从后往前搜的为什么这样操作呢
我们前面介绍了addWaiter方法后节点的pre指针先指向前节点前节点的next指针才会指向后节点这两个步骤不是原子性操作因此如果从前往后搜索可能前节点的next还没有建立好那么搜索可能就中断了。 6、ReentrantLock介绍
ReentrantLock被称为可重入锁是JUC包中并发锁的实现之一底层调用了AQS这个锁还包含了公平锁和非公平锁的特性
6.1 ReentrantLock的属性
ReentrantLock中只有一个sync属性sync属性被final修饰意味着一旦ReentrantLock被初始化sync属性就不可修改了。
ReentrantLock提供了两个构造器通过名字可以看出来我们可以通过传参的形式将ReentrantLock实例化为公平锁 或者是非公平锁。 private final Sync sync;// 默认构造器初始化为非公平锁public ReentrantLock() {sync new NonfairSync();}// 入参为true的时候是公平锁否则是非公平锁public ReentrantLock(boolean fair) {sync fair ? new FairSync() : new NonfairSync();}
6.2 ReentrantLock的内部类 - Sync
上面说了ReentrantLock存在一个Sync类型的构造器那么一起看看这个内部类吧这个内部类没有属性除了lock和readObject方法其余方法都用final修饰了不希望被外部破坏。
首先Sync内部类继承了AbstractQueuedSynchronizer说明AQS中机制Sync都可以借用了Sync被abstrat修饰说明需要通过子类来进行实例化NonfairSync / FairSync 后续会做相关介绍。lock() 方法加锁的抽象方法需要子类来实现后续介绍nonfairTryAcquire() 方法从名字来看是获取非公平锁在父类中定义是因为外层有调用tryRelease方法返回当前锁是否完全释放其余的一些方法简单了解即可 abstract static class Sync extends AbstractQueuedSynchronizer {private static final long serialVersionUID -5179523762034025860L;abstract void lock();// 尝试获取非公平锁 final boolean nonfairTryAcquire(int acquires) {final Thread current Thread.currentThread();// 获取AQS的state属性int c getState();if (c 0) {// state属性为0说明锁空闲通过cas来更改state// 如果成功则获取到了锁返回true否则在下面返回falseif (compareAndSetState(0, acquires)) {setExclusiveOwnerThread(current);return true;}}// state不为0判断当前线程是否是独占线程---- 这就是可重入锁的实现// 是独占锁则将state1并赋值回去因此AQS中的state数量其实就是当前线程重入次数else if (current getExclusiveOwnerThread()) {int nextc c acquires;if (nextc 0) // overflow// state是int类型的16位小于0就代表溢出了 -- 可重入的最大次数throw new Error(Maximum lock count exceeded);setState(nextc);return true;}return false;}// 释放锁 -- 注意返回值是“是否完全释放”protected final boolean tryRelease(int releases) {int c getState() - releases;if (Thread.currentThread() ! getExclusiveOwnerThread())throw new IllegalMonitorStateException();boolean free false;if (c 0) {free true;setExclusiveOwnerThread(null);}setState(c);return free;}// 判断当前线程是否获取到锁protected final boolean isHeldExclusively() {return getExclusiveOwnerThread() Thread.currentThread();}final ConditionObject newCondition() {return new ConditionObject();}// 获取占用锁的线程对象final Thread getOwner() {return getState() 0 ? null : getExclusiveOwnerThread();}// 返回state的值final int getHoldCount() {return isHeldExclusively() ? getState() : 0;}// 判断锁是否空闲final boolean isLocked() {return getState() ! 0;}// 反序列化可以不关注private void readObject(java.io.ObjectInputStream s)throws java.io.IOException, ClassNotFoundException {s.defaultReadObject();setState(0); // reset to unlocked state}} 6.3 ReentrantLock的内部类 - FairSync
公平锁锁的分配会按照请求锁的顺序比如按照AQS中的FIFO队列来排队获取锁就是公平锁的实现。公平锁能够保证只要你排队了就一定可以拿到锁。 static final class FairSync extends Sync {private static final long serialVersionUID -3000897897090466540L;// 调用AQS获取锁的逻辑final void lock() {acquire(1);}// 重写AQS的tryAcquire方法protected final boolean tryAcquire(int acquires) {final Thread current Thread.currentThread();int c getState();// 锁空闲并且AQS队列中没有其他节点那么就尝试取锁 --- 体现了公平性if (c 0) {if (!hasQueuedPredecessors() compareAndSetState(0, acquires)) {setExclusiveOwnerThread(current);return true;}}else if (current getExclusiveOwnerThread()) {// 如果获取锁的是当前线程就state1 -- 可重入int nextc c acquires;if (nextc 0)throw new Error(Maximum lock count exceeded);setState(nextc);return true;}return false;}}
6.4 ReentrantLock的内部类 - NonFairSync
非公平锁获取锁不会按照锁请求顺序比如抢占式非公平锁不会保证排队的线程一定会获取锁某个线程可能一直处于阻塞状态称为“饥饿”为什么设计非公平锁呢 --- 某些时候唤醒已经挂起的线程这个线程的状态切换会产生短暂的延时而这个延时时间可能就够进行一次业务处理因此非公平锁可利用这段时间完成操作提高效率。 static final class NonfairSync extends Sync {private static final long serialVersionUID 7316153563782823691L;// 加锁先尝试进行一次CAS失败则进入队列排队// 因为AQS的acquire方法中调用了tryAcquire下面又重写了tryAcquire方法// 结合nonfairTryAcquire方法代码可以看到非公平锁是先两次尝试获取锁失败之后在排队拿锁final void lock() {if (compareAndSetState(0, 1))setExclusiveOwnerThread(Thread.currentThread());elseacquire(1);}// 直接调用父类的nonfairTryAcquire方法。protected final boolean tryAcquire(int acquires) {return nonfairTryAcquire(acquires);}} 6.5 ReentrantLock的公共方法 - lock / tryLock / unlock
lock方法 : 加锁的方法很简单就是通过sync属性去调用sync的lock方法在上述内部类中都已经介绍了。-- 多态tryLock方法无论sync的实现是否是公平锁tryLock的实现都是非公平的 --- nonfairTryAcquire方法写在Sync中的原因unlock 释放锁操作每次对于每次执行state-1 public void lock() {sync.lock();}// 无论sync的实现是否是公平锁tryLock的实现都是非公平的public boolean tryLock() {return sync.nonfairTryAcquire(1);}// 含参的给了个超时时间public boolean tryLock(long timeout, TimeUnit unit)throws InterruptedException {return sync.tryAcquireNanos(1, unit.toNanos(timeout));}public void unlock() {sync.release(1);}
6.6 ReentrantLock 类的使用介绍
6.6.1 使用ReentrantLock进行普通加锁
功能类似于synchronized关键字获取到锁的线程释放锁之后其他线程才可以获取到锁。
public class TestReentrantLock {private static final Lock lock new ReentrantLock();public void testLock() {try {lock.lock();System.out.println(Thread name: Thread.currentThread().getName() lock);Thread.sleep(2000);lock.unlock();System.out.println(Thread name: Thread.currentThread().getName() unlock);} catch (InterruptedException e) {throw new RuntimeException(e);}}public static void main(String[] args) {TestReentrantLock testLock new TestReentrantLock();// 起两个线程第一个线程释放锁之后第二个线程才可以获取到锁new Thread(new Runnable() {Overridepublic void run() {testLock.testMethod();}}).start();new Thread(new Runnable() {Overridepublic void run() {testLock.testMethod();}}).start();}
}------print------------
Thread name: Thread-0 lock
Thread name: Thread-0 un lock
Thread name: Thread-1 lock
Thread name: Thread-1 un lock 6.6.2 使用ReentrantLock实现锁的可入
main方法中先加锁然后在解锁前调用testLock方法因为是在一个线程中所以不需要释放锁也可以获取到锁。
public class TestReentrantLock {private static final Lock lock new ReentrantLock();public void testLock() {try {lock.lock();System.out.println(Thread name: Thread.currentThread().getName() 加锁);Thread.sleep(2000);lock.unlock();System.out.println(Thread name: Thread.currentThread().getName() 解锁);} catch (InterruptedException e) {throw new RuntimeException(e);}}public static void main(String[] args) {TestReentrantLock testLock new TestReentrantLock();// main线程先加锁lock.lock();System.out.println(Thread.currentThread().getName() 线程获得了锁);testLock.testLock();lock.unlock();System.out.println(Thread.currentThread().getName() 线程释放了锁);}
}------print-----
main 线程获得了锁
Thread name: main 加锁
Thread name: main 解锁
main 线程释放了锁 7、CountDownLatch 介绍
CountDownLatch是JUC工具包中很重要的一个同步工具。CountDownLatch基于AQS, 他的作用是让某一个线程等待多个线程的操作完成之后再执行自己的操作。
CountDownLatch定义了一个计数器和一个阻塞队列当计数器的值递减为0之前阻塞队列里面的线程 7.1、CountDownLatch 的使用。
public class DemoCountDownLatch {private final static Random random new Random();static class SearchTask implements Runnable {private int id;private CountDownLatch latch;public SearchTask(int id, CountDownLatch latch) {this.id id;this.latch latch;}Overridepublic void run() {System.out.println(开始寻找 id 号龙珠);int seconds random.nextInt(10);try {Thread.sleep(seconds * 1000);} catch (InterruptedException e) {e.printStackTrace();}System.out.println(找到 id 号龙珠 花费了 seconds 秒时间);/*** 调用countDown() 方法计数器会减一*/latch.countDown();}}public static void main(String[] args) throws InterruptedException {ListInteger idList Arrays.asList(1, 2, 3, 4, 5, 6, 7);/*** CountDownLatch初始化的时候给定一个整数计数器不可变*/CountDownLatch latch new CountDownLatch(idList.size());for (Integer id : idList) {Thread thread new Thread(new SearchTask(id, latch));thread.start(); // 注意start 是开启新线程多线程执行run方法是串行的}/*** 调用 await() 方法时 如果计数器大于0当前线程阻塞直到计数器被countDown方法减到0时线程才会继续执行*/latch.await();/*** 调用 await方法时设置超时参数* 如果计数器大于0当前线程阻塞直到计数器被countDown方法减到0时线程才会继续执行*/// latch.await(3, TimeUnit.SECONDS);System.out.println(所有龙珠找到召唤神龙);}
}-----await() print ------------
----- 子线程都执行完了主线程才开始执行 ----------
开始寻找1号龙珠
开始寻找3号龙珠
开始寻找2号龙珠
开始寻找5号龙珠
开始寻找6号龙珠
开始寻找4号龙珠
开始寻找7号龙珠
找到3号龙珠花费了3秒时间
找到1号龙珠花费了3秒时间
找到7号龙珠花费了5秒时间
找到6号龙珠花费了5秒时间
找到2号龙珠花费了6秒时间
找到4号龙珠花费了6秒时间
找到5号龙珠花费了8秒时间
所有龙珠找到召唤神龙-----await(3, TimeUnit.SECONDS) print ------------
--- 设置了三秒超时三秒之后无论子线程有没有全部结束都执行主线程 ------
开始寻找1号龙珠
开始寻找3号龙珠
开始寻找2号龙珠
开始寻找4号龙珠
开始寻找5号龙珠
开始寻找7号龙珠
开始寻找6号龙珠
找到5号龙珠花费了3秒时间
所有龙珠找到召唤神龙
找到6号龙珠花费了5秒时间
找到7号龙珠花费了8秒时间
找到2号龙珠花费了8秒时间
找到1号龙珠花费了8秒时间
找到4号龙珠花费了9秒时间
找到3号龙珠花费了9秒时间-----不执行await print ------------
--- 主线程直接执行不等待子线程 ------
开始寻找3号龙珠
所有龙珠找到召唤神龙
开始寻找4号龙珠
开始寻找5号龙珠
开始寻找6号龙珠
开始寻找2号龙珠
开始寻找7号龙珠
开始寻找1号龙珠
找到1号龙珠花费了2秒时间
找到3号龙珠花费了3秒时间
找到4号龙珠花费了4秒时间
找到2号龙珠花费了5秒时间
找到5号龙珠花费了6秒时间
找到6号龙珠花费了9秒时间
找到7号龙珠花费了9秒时间 7.2、CountDownLatch设计思路
CountDownLatch我们就简单的描述为主线程等待子线程处理完任务之后在继续执行自己的任务。既然主线程在等待根据前面学习的AQS此时主线程应该放入等待队列中那么什么时候唤醒主线程呢当然是子任务都执行结束之后那么AQS中的state就可以派上用场了state表示子线程的数量也就是主线程需要等待的线程数目每当一个任务完成了state就减去1当state值为0 的时候就唤醒正在等待的主线程。 CountDownLatch 的设计思路大致就是上面介绍的下面会根据源码来进行剖析。
7.2.1 sync内部类
CountDownLatch内部也定义了一个sync内部类并继承了AQS 【tryAcquireShared】方法这个方法是尝试获取共享锁是对AQS方法的一个重写。 这个方法很简单获取state的值如果等于0就返回1否则返回-1 子类对父类方法的重写也是要按照约定去重写的我们在看看AQS中对tryAcquireShared方法的定义 返回负值获取锁失败返回0共享模式下获取锁成功不唤醒后续节点返回整数获取锁成功并唤醒后续节点 从父类方法定义来看如果tryAcquireShared方法返回整数是需要获取锁的但是子类实现的方法并没有获取锁的操作这个是为什么呢 ---- 实际上我们需要从CountDownLatch这个组件需要解决的问题来看待当CountDownLatch被初始化时必须传入一个count这个值会赋给aqs的state每当一个子任务完成state值就会减一一直到state为0然后主任务继续操作。这其实就是一个子任务不断释放锁主任务不断检查锁有没有完全释放的过程所有的操作不涉及到加锁的情况虽然主任务在state为0的时候也可以加锁但是完全没有必要。 【tryReleaseShared】方法就是一个释放锁的操作 锁完全释放返回true否则返回false private static final class Sync extends AbstractQueuedSynchronizer {private static final long serialVersionUID 4982264981922014374L;// 构造方法需要传入一个count值Sync(int count) {setState(count);}// 获取当前count值int getCount() {return getState();}// 对AQS的重写从名字看方法被shared修饰应该是用到了共享模式// 共享模式下state可以被多个线程同时修改加1代表获取共享锁减1代表释放共享锁// 这个方法就是如果state为0代表子任务全部完成否则就是还有没完成的protected int tryAcquireShared(int acquires) {return (getState() 0) ? 1 : -1;}// 释放锁的操作一个自旋的过程// 不需要释放锁或者没有完全释放锁返回false// 锁完全释放了返回trueprotected 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;}}}
7.2.2 内部属性和构造器
内部属性只有一个sync只有一个有参数构造器必须传递一个大于0的整数主线程需要等待子线程的数量。 private final Sync sync;public CountDownLatch(int count) {if (count 0) throw new IllegalArgumentException(count 0);this.sync new Sync(count);}
7.2.3 await 方法
这个方法就是主线程等待子线程执行完的逻辑。 public void await() throws InterruptedException {sync.acquireSharedInterruptibly(1);}// AQS的方法public final void acquireSharedInterruptibly(int arg)throws InterruptedException {if (Thread.interrupted())throw new InterruptedException();// 锁没有完全释放的情况也就是主线程等待子线程的场景if (tryAcquireShared(arg) 0)doAcquireSharedInterruptibly(arg);}private void doAcquireSharedInterruptibly(int arg)throws InterruptedException {// 设置共享模式的nodefinal Node node addWaiter(Node.SHARED);boolean failed true;try {for (;;) {final Node p node.predecessor();// 当前节点的前置节点为head说明当前节点可以被唤醒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);}}private void setHeadAndPropagate(Node node, int propagate) {Node h head; // Record old head for check belowsetHead(node);// 传播行为大于0唤醒后续节点// 被唤醒的节点会在doAcquireSharedInterruptibly方法中的for循环继续执行// 不断的唤醒队列中处于等待状态的且共享模式的线程。if (propagate 0 || h null || h.waitStatus 0 ||(h head) null || h.waitStatus 0) {Node s node.next;if (s null || s.isShared())doReleaseShared();}}
7.2.4 countDown方法
就是为了保证state的自减操作调用此方法state减1 public void countDown() {sync.releaseShared(1);}public final boolean releaseShared(int arg) {if (tryReleaseShared(arg)) {doReleaseShared();return true;}return false;}public final boolean releaseShared(int arg) {if (tryReleaseShared(arg)) {doReleaseShared();return true;}return false;}
