网页设计制作一个餐饮网站,网站开发使用天气api,wordpress页码颜色变更,微信网站开发软件Java JUC#xff08;三#xff09; AQS与同步工具详解
一. ReentrantLock 概述
ReentrantLock 是 java.util.concurrent.locks 包下的一个同步工具类#xff0c;它实现了 Lock 接口#xff0c;提供了一种相比synchronized关键字更灵活的锁机制。ReentrantLock 是一种独占…Java JUC三 AQS与同步工具详解
一. ReentrantLock 概述
ReentrantLock 是 java.util.concurrent.locks 包下的一个同步工具类它实现了 Lock 接口提供了一种相比synchronized关键字更灵活的锁机制。ReentrantLock 是一种独占式且可重入的锁并且支持可中断、公平锁/非公平锁、超时等待、条件变量等高级特性。其特点如下
独占式 一把锁在同一时间只能被一个线程所获取可重入 可重入意味着同一个线程如果已持有某个锁则可以继续多次获得该锁注意释放同样次之后才算完全释放成功可中断 在线程获取锁的等待过程中可以中断获取放弃等待而转去执行其他逻辑公平性 ReentrantLock支持公平锁和非公平锁默认两种模式。其中公平锁会按照线程请求锁的顺序来分配锁降低性能而非公平锁允许线程抢占已等待的线程的锁可能存在饥饿现象条件变量 通过 Condition接口的实现允许线程在某些条件下等待或唤醒即可以实现选择性通知
TypeMethodDescription/ReentrantLock()无参构造方法默认为非公平锁/ReentrantLock(boolean fair)带参构造方法其中fair表示锁的公平性策略- true 公平锁- false: 非公平锁voidlock()不可中断式获取锁。若当前锁已被其他线程持有则阻塞等待**注意**该获锁过程不可被中断voidlockInterruptibly() throws InterruptedException可中断式获取锁。若当前锁已被其他线程持有则阻塞等待**注意**该获锁等待过程可被中断抛出InterruptedException并清除当前线程的中断状态booleantryLock()尝试获取锁该方法会立即返回。若获锁成功则返回true否则将返回false**注意**该方法会破坏公平锁配置即在公平锁策略下该方法也会立即尝试获取可用锁booleantryLock(long timeout,TimeUnit unit) throws InterruptedException在给定时间timeout内尝试获取锁。若获锁成功则返回true否则将阻塞等待直到timeout过期返回false注意 - 获锁等待过程可被中断抛出InterruptedException并清除当前线程的中断状态- 遵循公平锁配置策略即在公平锁策略下该方法会按顺序等待获取锁voidunlock()当前线程尝试释放该锁。若当前线程未持有该锁则抛出IllegalMonitorStateException异常ConditionnewCondition()返回一个与当前Lock实例绑定的条件变量集合对象默认返回AQS内部实现类ConditionObject用于实现线程的条件等待/唤醒详见后文
1. 锁的基本使用
相比synchronized关键字来说ReentrantLock属于显式锁其锁机制都是针对Lock实例对象本身进行加锁并且在使用过程中需要手动释放即锁的获取与释放是成对出现的除此之外ReentrantLock属于JDK API层面实现的互斥锁其通过方法调用实现锁功能可以跨方法从而更加灵活。为了避免出现死锁问题官方建议的开发方式如下
// new lock object
ReentrantLock lock new ReentrantLock();
lock.lock(); // block until condition holds
try {// ... method body
} finally {lock.unlock(); // release
}问题背景 假设当前有一个卖票系统一共有100张票有4个窗口同时售卖请模拟该卖票过程注意保证出票的正确性。
public class Test_01 {public static void main(String[] args) {ReentrantLock lock new ReentrantLock();for (int i 0; i 4; i){new Thread(new TicketSeller(lock), Thread_ i).start();}}
}
// 售票类实现
class TicketSeller implements Runnable{private static int ticketCount 100;private ReentrantLock windowLock;public TicketSeller(ReentrantLock lock) {this.windowLock lock;}Overridepublic void run() {while (true) {try {Thread.sleep(10);}catch(InterruptedException e){e.printStackTrace();}windowLock.lock();try{if(ticketCount 0){System.out.println(Thread.currentThread().getName() : sale and left --ticketCount);}else{System.out.println(Thread.currentThread().getName() : sold out...);break;}}finally {windowLock.unlock();}}}
}2. 条件变量机制
在synchronized关键字中我们可以通过waitnotify实现线程的等待与唤醒但在此场景下存在虚假唤醒问题根本原因就是其等待/唤醒机制只支持单条件等待线程未作区分只能全部唤醒。相比之下ReentrantLock基于Condition接口也实现了相同的机制并提供了更细的粒度和更高级的功能每个Condition实例都对应一个条件队列用于维护在该条件场景下等待通知的线程并且ReentrantLock支持多条件变量即一个ReentrantLock可以关联多个Condition实例。其常用方法如下
TypeMethodDescriptionvoidawait() throws InterruptedException使当前线程阻塞等待进入该条件队列并释放与此条件变量所关联的锁。注意 若在等待期间被中断则抛出InterruptedException并清除当前线程中断状态booleanawait(long time,TimeUnit unit) throws InterruptedException使当前线程阻塞等待并释放与此条件变量所关联的锁直到被唤醒、被中断或等待time时间过期。其返回值表示- true: 在等待时间之内条件被唤醒- false: 等待时间过期条件未被唤醒注意 若在等待期间被中断则抛出InterruptedException并清除当前线程中断状态voidsignal()唤醒一个等待在Condition上的线程。如果有多个线程在此条件变量下等待则选择任意一个线程唤醒注意 从等待方法返回前必须重新获得Condition相关联的锁voidsignalAll()唤醒所有等待在Condition上的线程。如果有多个线程在此条件变量下等待则全部唤醒 注意 从等待方法返回前必须重新获得Condition相关联的锁
在使用时需要注意
调用await相关方法前需要先获得对应条件变量所关联的锁否则会抛出IllegalMonitorStateException异常调用signal相关方法前需要先获得对应条件变量所关联的锁否则会抛出IllegalMonitorStateException异常await线程被唤醒或等待时间过期、被中断后会重新参与锁的竞争若成功拿到锁则将从await处恢复继续向下执行
/*** 场景模拟奶茶店和咖啡店共用一个窗口window出餐等待顾客点单...* - 奶茶店teaWithMilk顾客需要奶茶则奶茶店开始工作* - 咖啡店(coffee)顾客需要咖啡则咖啡店开始工作*/
public class Test {// 窗口锁(ReentrantLock实现)static final ReentrantLock window new ReentrantLock();// 奶茶点单条件变量static Condition teaWithMilk window.newCondition();// 咖啡点单条件变量static Condition coffee window.newCondition();public static void main(String[] args) throws InterruptedException {// 奶茶店监控线程new Thread(new Runnable() {Overridepublic void run() {while (true) {window.lock();try {System.out.println([奶茶店] 等待接单...);try {teaWithMilk.await();} catch (InterruptedException e) {e.printStackTrace();}System.out.println([奶茶店] 接到订单...);} finally {window.unlock();}System.out.println([奶茶店] 开始工作...);try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}System.out.println([奶茶店] 工作完成...);}}}).start();Thread.sleep(1000);// 咖啡店监控线程new Thread(new Runnable() {Overridepublic void run() {while (true) {window.lock();try {System.out.println([咖啡店] 等待接单...);try {coffee.await();} catch (InterruptedException e) {e.printStackTrace();}System.out.println([咖啡店] 接到订单...);} finally {window.unlock();}System.out.println([咖啡店] 开始工作...);try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}System.out.println([咖啡店] 工作完成...);}}}).start();Thread.sleep(1000);// 顾客点单线程: mainwindow.lock();try{System.out.println([顾客] 点了咖啡!!);coffee.signal(); // 唤醒咖啡条件等待线程}finally {window.unlock();}}
}二. 从 ReentrantLock 分析 AQS 的原理
1. AQS 框架
AQS 全称为 AbstractQueuedSynchronizer 即抽象队列同步器AQS 是 java.util.concurrent.locks 包下的一个抽象类其为构建锁和同步器提供了一系列通用模板与框架的实现大部分JUC包下的并发工具都是基于AQS来构建的比如ReentrantLock、Semaphore、CountDownLatch等。其核心源码如下
public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable {// 同步队列的节点static final class Node {...}// 指向同步队列头部private transient volatile Node head;// 指向同步队列尾部private transient volatile Node tail;// 同步状态private volatile int state;// 提供一系列并发、同步队列的基本操作方法// 比如: 挂起、取消、节点插入、节点替换等...// 交由子类实现的模板方法钩子方法: 自定义同步器的核心实现目标protected boolean tryAcquire(int arg) {throw new UnsupportedOperationException();}protected boolean tryRelease(int arg) {throw new UnsupportedOperationException();}protected int tryAcquireShared(int arg) {throw new UnsupportedOperationException();}protected boolean tryReleaseShared(int arg) {throw new UnsupportedOperationException();}protected boolean isHeldExclusively() {throw new UnsupportedOperationException();}// 条件变量 Condition 接口的内部实现类public class ConditionObject implements Condition {...}}由上可知AQS内部实现了一个核心内部类Node该内部类表示对等待获取锁的线程的封装节点在AQS中基于Node维护了一个双向链表模拟同步队列其中head节点指向同步队列的头部而tail节点指向同步队列的尾部。Node类的核心源码如下 static final class Node {// 共享模式共享锁、比如Semaphorestatic final Node SHARED new Node();// 独占模式独占锁、比如ReentrantLockstatic final Node EXCLUSIVE null;// 标识节点线程获取锁的请求已取消、已结束static final int CANCELLED 1;// 标识节点线程已准备就绪等待被唤醒获取资源static final int SIGNAL -1;// 标识节点线程在条件变量Condition中等待static final int CONDITION -2;// 在共享模式下启用: 标识获得的同步状态会被传播static final int PROPAGATE -3;/*** waitStatus 标识节点线程在同步队列中的状态共存在以下几种情况:* 1SIGNAL: 被标记为SIGNAL的节点处于等待唤醒获取资源的状态只要前驱节点释放锁就会通知该状态的后续节点线程执行* 2CANCELLED: 在同步队列中等待超时、被中断的线程会进入取消状态不再响应并会在遍历过程中被移除* 3CONDITION: 标识当前节点线程在Condition下等待被唤醒后将重新从等待队列转移到同步队列* 4PROPAGATE: 与共享模式有关 * 50: 默认初始值状态代表节点初始化*/volatile int waitStatus;// 同步队列中的前驱节点volatile Node prev;// 同步队列中的后继节点volatile Node next;// 等待获取锁资源的线程volatile Thread thread;// Condition 等待队列中的后继节点单向链表Node nextWaiter;// 判断是否为共享模式final boolean isShared() {return nextWaiter SHARED;}// 获取当前节点在同步队列中的前驱节点final Node predecessor() throws NullPointerException {Node p prev;if (p null)throw new NullPointerException();elsereturn p;}// 省略构造方法...}在了解Node的基本数据结构与状态之后AQS还有一个核心状态变量即state该全局变量用于表示同步锁的状态其具体含义一般在子类实现中进行定义和维护独占锁和共享锁不一样。但不管独占模式还是共享模式简单来说都是使用一个volatile的全局变量来表示资源同步状态state并通过CAS完成对state值的修改修改成功则表示获锁成功当持有锁的线程数量超过当前模式独占模式一般限制为1时则通过内置的FIFO同步队列来完成资源获取线程的排队工作通过LockSupport park/unpark方法实现挂起与唤醒。其核心原理图如下 综上AQS采用了模板方法模式来构建同步框架并提供了一系列并发操作的公共基础方法支持共享模式和独占模式两种实现但AQS并不负责对外提供具体的加锁/解锁逻辑因为锁是千变万化的AQS只关注基础组件、顶层模板这些总的概念具体的锁逻辑将通过”钩子“的方式下放给子类实现。也就是说独占模式只需要实现tryAcquire-tryRelease方法、共享模式只需要实现tryAcquireShared-tryReleaseShared方法搭配AQS提供的框架和基础组件就能轻松实现自定义的同步工具。
2. ReentrantLock 源码分析 由ReentrantLock的类结构图可以看出ReentrantLock实现了Lock接口其内部包含一个内部类Sync该内部类继承了AQSAbstractQueuedSynchronizerReentrantLock大部分的锁操作都是通过Sync实现的。除此之外ReentrantLock有公平锁和非公平锁两种模式分别对应Sync的FairSync和NonfairSync两个子类实现。
public class ReentrantLock implements Lock, java.io.Serializable {private final Sync sync;// 默认构造函数: 默认创建非公平锁public ReentrantLock() {sync new NonfairSync();}// 带参构造函数: true公平锁/false非公平锁public ReentrantLock(boolean fair) {sync fair ? new FairSync() : new NonfairSync();}// 加锁操作public void lock() {sync.lock();}//...
}接下来本节将首先以ReentrantLock的非公平锁为例进行分析然后再介绍公平锁与非公平锁的主要区别。
2.1 非公平锁lock加锁原理
非公平锁NonfairSync的源码如下 static final class NonfairSync extends Sync {// 加锁操作final void lock() {// CAS修改state状态以获取锁资源if (compareAndSetState(0, 1))// 成功则将独占锁线程设置为当前线程setExclusiveOwnerThread(Thread.currentThread());else// 否则再次请求同步状态AQS的模板方法acquire(1);}// AQS 获锁的钩子方法实现protected final boolean tryAcquire(int acquires) {return nonfairTryAcquire(acquires);}}在调用lock方法获取锁的过程中当前线程会先通过CAS操作尝试修改state从0表示无锁到1表示占有锁若修改成功则将AQS中保存的独占线程exclusiveOwnerThread修改为当前线程若失败则执行acquire(1)方法该方法是AQS中的一个模板方法其源码如下 public final void acquire(int arg) {// tryAcquire - addWaiter - acquireQueuedif (!tryAcquire(arg) acquireQueued(addWaiter(Node.EXCLUSIVE), arg))selfInterrupt();}可以看到该方法首先调用了钩子方法tryAcquire该方法是交由子类NonfairSync实现的在上面的源码中我们已经给出了tryAcquire的实现代码其直接调用了父类Sync中的nonfairTryAcquire方法其源码如下
abstract static class Sync extends AbstractQueuedSynchronizer {abstract void lock();// nonfairTryAcquire 方法实现final boolean nonfairTryAcquire(int acquires) {// 获取当前线程及同步队列状态statefinal Thread current Thread.currentThread();int c getState();// 若状态为0表示锁已释放: 重新尝试获取锁if (c 0) {// CAS尝试修改state的值if (compareAndSetState(0, acquires)) {// 若成功则设置独占线程为当前线程setExclusiveOwnerThread(current);return true;}}// 若独占线程即当前线程则属于重入锁else if (current getExclusiveOwnerThread()) {// 修改state为重入值int nextc c acquires;if (nextc 0) // overflowthrow new Error(Maximum lock count exceeded);setState(nextc);return true;}return false;}//省略代码...
}由上述代码可知该方法首先再次判断锁是否已释放这是为了避免之前持锁的线程在这段时间内又重新释放了锁若state0则会尝试再次CAS修改同步状态以获取锁资源否则的话则判断当前线程是否是锁重入的情况若两个判断都不满足则返回false到目前为止我们回过头来想一下非公平锁的非公平性是在哪体现的
很明显在上述代码分析中当有任何线程尝试获取锁时调用lock方法不论当前同步队列中是否已有线程排队等待NonfairSync的lock()方法以及Sync的nonfairTryAcquire()方法都没有对同步队列中的等待情况进行判断而是直接通过CAS尝试修改state的值来为当前线程直接占有锁这就是非公平性的体现抢占线程可以直接与等待线程竞争锁资源而不用按照顺序加入队列。
分析完这部分之后我们再回到acquire(1)方法若tryAcquire(arg)方法返回false即获取不到锁时会继续向下执行到addWaiter(Node.EXCLUSIVE)方法该方法用于封装线程入队其源码如下 private Node addWaiter(Node mode) {// 将请求占锁失败的线程封装为Node节点Node node new Node(Thread.currentThread(), mode);Node pred tail;// 若同步队列不为空则尝试CAS在尾部插入当前节点FIFOif (pred ! null) {node.prev pred;if (compareAndSetTail(pred, node)) {pred.next node;return node;}}// 若队列为空或CAS插入节点失败则执行enq()方法处理入队enq(node);return node;}接着我们继续分析enq(node)方法的实现 private Node enq(final Node node) {// 开启自旋循环for (;;) {Node t tail;// 若队列为空if (t null) { // Must initialize// 则尝试CAS创建头节点不存储数据// 原因: 队列为空可能是因为其他线程非公平占有了锁当前线程试过没抢到因此这里需要先斩后奏即再次创建头节点表示已占锁线程的占位来维护同步队列if (compareAndSetHead(new Node()))tail head;} else {// 否则尝试CAS添加尾节点node.prev t;if (compareAndSetTail(t, node)) {t.next node;return t;}}}// 若CAS失败可能有多线程并发操作则不断自旋重试直到插入成功}可以看到头节点head其实是不存储数据的它只表示一个线程占位占位了锁资源因为位于头节点的线程肯定已经获取到了锁头节点只存储后继节点指向用于当前线程释放锁资源时唤醒后继节点那么到此这个方法也就分析完成了在节点入队成功之后会返回当前节点node然后会继续执行到acquireQueued(addWaiter(Node.EXCLUSIVE),arg)方法其源码如下 final boolean acquireQueued(final Node node, int arg) {boolean failed true;try {boolean interrupted false; //阻塞挂起标志// 开启自旋循环for (;;) {// 获取当前节点的前驱节点pfinal Node p node.predecessor();// 若p是头节点则当前节点尝试获取锁资源if (p head tryAcquire(arg)) {// 占锁成功则设置当前节点node为头节点: 头节点状态保持SIGNAL状态setHead(node);p.next null; // help GCfailed false;return interrupted;}// 若p不是头节点或获取锁资源失败则判断是否阻塞挂起线程来等待if (shouldParkAfterFailedAcquire(p, node) parkAndCheckInterrupt())interrupted true;}} finally {// 若最终无法获取锁则取消该线程的请求if (failed)cancelAcquire(node);}}// 将传递的节点设置为同步队列的头节点private void setHead(Node node) {head node;// 清空当前节点存储的线程信息node.thread null;node.prev null;}在acquireQueued方法中线程会开启自旋若发现或被唤醒后发现当前节点的前驱节点变为头节点则说明当前节点能够尝试获取锁资源并尝试通过tryAcquire方法获取同步状态需要注意的是head头节点表示当前占有锁的线程节点只有当head节点对应的线程释放锁资源并唤醒后继节点时后继节点线程才会自旋去尝试占有锁资源因此在同步队列中只有前驱节点变为头节点时当前节点才有资格尝试获取锁资源其他时候都将被挂起等待避免空转CPU。
除此之外若在自旋过程中当前节点的前驱节点不是头节点或者节点尝试tryAcquire获取锁资源失败则会执行shouldParkAfterFailedAcquire(p,node)parkAndCheckInterrupt()逻辑需要注意的是在前驱节点是头节点但尝试获取锁资源失败这种特殊情况发生时比如非公平锁模式下被新到来的请求线程抢占头节点head此时可能有两种状态 waitStatus0 处于该状态一种情况是初始同步队列为空时默认头节点状态初始化为0另一种情况是锁释放时见后文被unparkSuccessor重置头节点状态 waitStatusSIGNAL 处于该状态一种情况是waitStatus0时更新状态后又自旋回来但仍未获取到锁可能释放锁后被非公平抢占另一种情况是释放锁时unparkSuccessor重置失败
其源码如下 // 判断节点线程是否挂起等待private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {// 获取前驱节点的等待状态int ws pred.waitStatus;// 若是SIGNAL状态则说明前驱节点就绪当前节点正常需要继续等待即返回trueif (ws Node.SIGNAL)/** This node has already set status asking a release* to signal it, so it can safely park.*/return true;// 若等待状态0则说明前驱节点是结束状态需要遍历前驱节点直到找到非结束状态的有效节点if (ws 0) {/** Predecessor was cancelled. Skip over predecessors and* indicate retry.*/do {node.prev pred pred.prev;} while (pred.waitStatus 0);pred.next node;} else {// 若等待状态0且非SIGNAL则尝试将前驱节点设置为SIGNAL/** waitStatus must be 0 or PROPAGATE. Indicate that we* need a signal, but dont park yet. Caller will need to* retry to make sure it cannot acquire before parking.*/compareAndSetWaitStatus(pred, ws, Node.SIGNAL);}// false 则返回去继续自旋return false;}// 执行挂起阻塞操作 LockSupport.parkprivate final boolean parkAndCheckInterrupt() {LockSupport.park(this);return Thread.interrupted();}shouldParkAfterFailedAcquire方法的执行逻辑是判断前驱节点的等待状态waitStatus用于挂起当前节点线程等待结合上述分析包括以下几种情况
前驱节点状态waitStatusSIGNAL 若前驱节点是头节点则说明占锁线程还未执行结束当前节点线程仍需挂起等待若前驱节点不是头节点则说明前驱节点就绪/拥有更高的优先级下次执行还轮不到当前节点所以也可以安全挂起直接返回true前驱节点状态waitStatus0 说明前驱节点已处于结束/取消状态应该从同步队列中移除并遍历所有前驱节点直到找到非结束状态的有效节点作为前驱前驱节点状态waitStatus0且非SIGNAL 前驱节点刚从Condition的条件等待队列被唤醒从而转移到同步队列需要转换为SIGNAL状态等待前驱节点状态waitStatus0若前驱节点是头节点则说明同步队列刚初始化0或锁刚被释放重置锁资源可能未被其他线程持有需判断能否占有锁不管当前线程能否占有该锁一定会被占有都需要转换状态为SIGNAL若前驱节点不是头节点则说明该线程节点刚初始化并被插入队列需要转换为SIGNAL状态
综上当shouldParkAfterFailedAcquire()方法返回true时会调用parkAndCheckInterrupt方法挂起线程等待被唤醒返回false时则会继续自旋判断至此ReetrantLock内部间接依靠AQS的FIFO同步队列就完成了lock()加锁操作。
2.2 公平锁lock加锁原理
公平锁FairSync的源码如下 static final class FairSync extends Sync {// 加锁操作final void lock() {acquire(1);}// AQS 获锁的钩子方法实现protected final boolean tryAcquire(int acquires) {// 获取当前线程final Thread current Thread.currentThread();// 获取同步状态int c getState();// 若当前没有线程持有锁资源if (c 0) {// 首先判断同步队列是否存在等待节点if (!hasQueuedPredecessors() 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;}}与非公平锁唯一不同的是公平锁的tryAcquire实现中在尝试修改state之前会先调用hasQueuedPredecessors()判断AQS内部同步队列中是否已存在等待节点。如果存在则说明在此之前已经有线程提交了获取锁的请求那么当前线程就会直接被封装成Node节点追加到队尾等待。
2.3 释放锁原理
ReetrantLock显式锁需要手动释放锁资源其unlock()方法直接调用了Sync中的release(1)方法而该方法又是在其父类AQS中直接实现的其源码如下 public final boolean release(int arg) {// 尝试释放锁if (tryRelease(arg)) {// 进入该代码块则说明锁已完全释放state0// 获取头节点Node h head;if (h ! null h.waitStatus ! 0)// 唤醒head头节点的后继节点线程unparkSuccessor(h);return true;}return false;}// 唤醒node的后继节点线程private void unparkSuccessor(Node node) {// 获取节点状态int ws node.waitStatus;if (ws 0) // 重置节点状态允许失败compareAndSetWaitStatus(node, ws, 0);// 获取后继节点Node s node.next;// 若后继节点为空或已结束if (s null || s.waitStatus 0) {s null;// 寻找后继可被唤醒的有效等待节点for (Node t tail; t ! null t ! node; t t.prev)if (t.waitStatus 0)s t;}// 执行线程唤醒继续去自旋 LockSupport.unparkif (s ! null)LockSupport.unpark(s.thread);}release方法能否释放锁并唤醒后继节点线程依赖于tryRelease钩子方法而该方法又下放到了Sync中实现其源码如下 // ReentrantLock - Sync - tryRelease(1)protected final boolean tryRelease(int releases) {// 计算释放锁后的同步更新状态int c getState() - releases;// 如果当前释放锁的线程不为持有锁的线程则抛出异常if (Thread.currentThread() ! getExclusiveOwnerThread())throw new IllegalMonitorStateException();boolean free false;// 判断更新状态是否为0如果是则说明已完全释放锁if (c 0) {free true;// 完全释放锁才清空当前占有线程setExclusiveOwnerThread(null);}// 更新state值setState(c);// 完全释放锁才返回truereturn free;}注意tryRelease方法执行完成返回true之后就说明当前线程持有的锁已被释放非公平锁中就已经可以被抢占了后续unparkSuccessor方法只是进行一些善后工作其中重置头节点状态的目的是表示逻辑上从持锁到无锁的转换锁资源目前可能并没有线程持有因此在后续线程唤醒后执行acquireQueued自旋时waitStatus0状态会再一次判断并尝试获取锁而修改为SIGNAL就表示占锁线程正在执行其他线程需要挂起等待。至此整个流程可以结合起来理解s节点的线程被唤醒后会继续执行acquireQueued()方法中的自旋判断if (p head tryAcquire(arg))代码是否成立从而执行判断操作。
三. 其他同步工具类
1. Semaphore
1.1 基本概述
Semaphore是java.util.concurrent包下的一种计数信号量它同样也是基于AQS实现的同步工具类。相比ReentrantLock来说它应该属于共享锁即允许多个线程同时访问某个共享资源但会限制同时访问特定资源的线程数量Semaphore同样也支持公平模式和非公平模式两种方式其构造方法如下
public Semaphore(int permits) {sync new NonfairSync(permits);
}public Semaphore(int permits, boolean fair) {sync fair ? new FairSync(permits) : new NonfairSync(permits);
}Semaphore默认为非公平模式抢占式在构造信号量对象时都必须提供permits参数permits可以理解为许可证数量只有拿到许可证的线程才能执行该参数限制了能同时获取或访问到共享资源的线程数量其他超出线程都将阻塞等待。基于此Semaphore通常用于实现资源有明确访问数量限制的场景比如限流、池化等。Semaphore的常用方法介绍如下
TypeMethodDescriptionvoidacquire() throws InterruptedException当前线程请求获取该信号量的一个许可证。若有可用许可证则获得许可证并返回执行同步代码同时可用许可证数量将减少一个若没有可用许可证则阻塞等待直到有许可被释放或线程中断。注意 若当前线程在等待过程中被中断则会抛出InterruptedException并清除当前线程的中断状态。voidacquire(int permits) throws InterruptedException当前线程请求获取该信号量的permits个许可证。若有可用数量的许可证则获得许可证并返回执行同步代码同时可用许可证数量将减少permits个若没有可用数量的许可证则阻塞等待直到可用许可达到指定数量或线程中断。**注意 **若当前线程在等待过程中被中断则会抛出InterruptedException并清除当前线程的中断状态。voidrelease()释放该信号量的一个许可证并使可用许可证数量增加一个。注意 没有通过acquire()获取许可的线程甚至也可以直接调用release()来为信号量增加许可证数量并且可用许可有可能会超出构造时限制的permits值因此信号量的正确使用必须是通过应用程序中的编程约束来建立。voidrelease(int permits)释放该信号量的permits个许可证并使可用许可证数量增加permits个。注意 同release()方法信号量的正确使用必须是通过应用程序中的编程约束来建立。booleantryAcquire()尝试获取该信号量的一个许可证但该方法会立即返回。若有可用许可证则获得许可证并返回true同时可用许可证数量将减少一个若没有可用许可证则返回false。注意 该方法会破坏公平策略对该方法的调用会进行抢占式获取不管是否有线程在等待。booleantryAcquire(long timeout, TimeUnit unit) throws InterruptedException尝试在指定时间timeout内获取该信号量的一个许可证遵循公平策略。若有可用许可证则获得许可证并返回true同时可用许可证数量将减少一个若没有可用许可证则阻塞等待直到timeout过期等待时间过期则返回false。注意 若当前线程在等待过程中被中断则会抛出InterruptedException并清除当前线程的中断状态。intavailablePermits()获取当前信号量中的可用许可证数量。
可以看出许可证是Semaphore的核心概念Semaphore信号量对许可证的获取是强限制但对许可证的释放是弱限制的即请求线程在执行时必须获取acquire到指定数量的许可证但在释放release时并不会对先前是否获取进行检查因此可用许可有时可能会超出构造时限制的permits值。换句话说构造时传入的permits参数只表示信号量的初始许可数量并且许可证只决定了线程执行的门槛但并不会对线程作全程限制当前线程一旦获取到指定数量的许可便开始执行即使中途释放许可也不会影响后续执行过程这也就是为什么说信号量的正确使用必须是通过应用程序中的编程约束来建立。举例如下
public class test {public static void main(String[] args) {// 初始化许可数量 5Semaphore semaphore new Semaphore(5);new Thread(() - {try {System.out.println(Thread.currentThread().getName() : 等待许可...);semaphore.acquire(3); // acquire permits 3System.out.println(Thread.currentThread().getName() : 拿到许可,剩余许可 semaphore.availablePermits());Thread.sleep(3000);semaphore.release(); // release 1System.out.println(Thread.currentThread().getName() : 释放许可...);} catch (InterruptedException e) {e.printStackTrace();}},thread-1).start();try {Thread.sleep(10);} catch (InterruptedException e) {e.printStackTrace();}new Thread(() - {try {System.out.println(Thread.currentThread().getName() : 等待许可...);semaphore.acquire(3);// acquire permits 3System.out.println(Thread.currentThread().getName() : 拿到许可,剩余许可 semaphore.availablePermits());Thread.sleep(3000);semaphore.release();// release 1System.out.println(Thread.currentThread().getName() : 释放许可,剩余许可 semaphore.availablePermits());} catch (InterruptedException e) {e.printStackTrace();}},thread-2).start();}
}thread-1: 等待许可...
thread-1: 拿到许可,剩余许可 2
thread-2: 等待许可...
thread-1: 释放许可...
thread-2: 拿到许可,剩余许可 0
thread-2: 释放许可,剩余许可 1前文说过Semaphore通常用于实现资源有明确访问数量限制的场景比如限流、池化等此处通过Semaphore模拟一个请求限流的场景其中限制最大并发数为3实现代码如下
public class test {public static void main(String[] args) {// 自定义线程池ThreadPoolExecutor threadPool new ThreadPoolExecutor(2*2, 8,60, TimeUnit.SECONDS,new LinkedBlockingDequeRunnable(1024),new ThreadPoolExecutor.AbortPolicy());// 流量控制: 限制最大并发数为3final Semaphore semaphore new Semaphore(3);// 模拟10个客户端任务请求for(int index 0;index 10;index){final int serial index;threadPool.execute(() - {try {// 请求获取许可semaphore.acquire();System.out.println(Thread.currentThread().getName() : 请求成功!访问编号 serial);// 模拟IO操作Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();} finally {// 释放许可semaphore.release();}});}// 等待线程池执行结束关闭: 不再接受新任务提交对已经提交了的任务不会产生影响threadPool.shutdown();}
}1.2 原理分析
Semaphore的大部分方法都是基于内部类Sync实现的而该类又继承了 AbstractQueuedSynchronizer即AQS并且Sync对应的还有两个子类 NonfairSync非公平模式实现 和 FairSync公平模式实现。在Semaphore中AQS的 state 被定义为 permits许可证数量对象创建时传入的参数permits实际是在对AQS内部的state进行初始化初始化完成后state代表着当前信号量对象的可用许可数state0。
以非公平模式为例当线程调用Semaphore.acquire(arg)请求获取许可时会首先判断remaining getState() - arg是否大于0如果是则代表还有满足可用的许可数并尝试对state进行CAS操作使stateremaining若CAS成功则代表获取许可成功否则线程需要封装成Node节点并加入同步队列阻塞等待直到许可释放被唤醒。
// Semaphore类 - acquire()方法
public void acquire() throws InterruptedException {// Sync类继承AQS此处直接调用AQS内部的acquireSharedInterruptibly()方法sync.acquireSharedInterruptibly(1);}// AbstractQueuedSynchronizer类 - acquireSharedInterruptibly()方法
public final void acquireSharedInterruptibly(int arg)throws InterruptedException {// 判断是否出现线程中断信号标志if (Thread.interrupted())throw new InterruptedException();// 如果tryAcquireShared(arg)执行结果不小于0则线程获取同步状态成功if (tryAcquireShared(arg) 0)// 未获取成功加入同步队列阻塞等待doAcquireSharedInterruptibly(arg);
}// Semaphore类 - NofairSync内部类 - tryAcquireShared()方法
protected int tryAcquireShared(int acquires) {// 调用了父类Sync中的实现方法return nonfairTryAcquireShared(acquires);
}// Syn类 - nonfairTryAcquireShared()方法
abstract static class Sync extends AbstractQueuedSynchronizer {final int nonfairTryAcquireShared(int acquires) {// 开启自旋死循环for (;;) {int available getState();int remaining available - acquires;// 判断信号量中可用许可数是否已0或者CAS执行是否成功if (remaining 0 ||compareAndSetState(available, remaining))return remaining;}}
}释放逻辑对比获取许可的逻辑相对来说要简单许多只需要更新state值增加后调用doReleaseShared()方法唤醒后继节点线程即可需要注意的是而在共享模式中可能会存在多条线程同时释放许可/锁资源所以在此处使用了CAS自旋的方式保证线程安全问题。
2. CountDownLatch
2.1 基本概述
CountDownLatch同样是java.util.concurrent包下的基于AQS实现的同步工具类。类似于SemaphoreCountDownLatch在初始化时也会传入一个参数count来间接赋值给AQS的state用于表示一个线程计数值不过CountDownLatch并没有构建公平模式和非公平模式内部Sync没有子类实现其构造方法如下
public CountDownLatch(int count) {if (count 0) throw new IllegalArgumentException(count 0);this.sync new Sync(count);
}CountDownLatch的主要作用是等待计数值count归零后唤醒所有的等待线程。基于该特性CountDownLatch常被用于控制多线程之间的等待与协作多线程条件唤醒相比join来说CountDownLatch更加灵活且粒度更细join是以线程执行结束为条件而CountDownLatch是以方法的主动调用为条件。其常用方法如下
TypeMethodDescriptionvoidawait() throws InterruptedException使当前线程阻塞等待直到计数器count归零或线程被中断。若当前计数已为零则此方法立即返回。注意 若在等待过程中被中断则会抛出InterruptedException并清除当前线程的中断状态。voidcountDown()使当前计数器count递减。如果新计数归零则唤醒所有await()等待线程注意 若当前计数已为零则无事发生。longgetCount()获取当前计数器count的值。
需要注意的是CountDownLatch 是一次性的即计数器的值count只能在构造方法中初始化此外再没有任何设置值的方法当 CountDownLatch 使用完毕后计数归零将不能重复被使用若需要重置计数的版本可以考虑使用CyclicBarrier。CountDownLatch 的常用方法有两种
多等一初始化count1多条线程await()阻塞等待一条线程调用countDown()唤醒所有线程。比如模拟并发安全、死锁等一等多初始化countN一条线程await()阻塞等待N条线程调用countDown()归零后唤醒。比如多接口调用的数据合并、多操作完成后的数据检查、主服务启动后等待多个组件加载完毕等注意线程间的通信与数据传递需结合Future实现
public class test {public static void main(String[] args) {// 模拟10人拼团活动final CountDownLatch countDownLatch new CountDownLatch(10);// 固定数量线程池ExecutorService threadPool Executors.newFixedThreadPool(50);// 拼团人员ID集合ListString ids new ArrayList();// 模拟30人开始抢单拼团for (int i 0; i 30; i) {threadPool.execute(() - {boolean orderSucess false;System.out.println(Thread.currentThread().getName() : 请求拼团...);if (countDownLatch.getCount() 0) {synchronized (ids) {if (countDownLatch.getCount() 0) {ids.add(Thread.currentThread().getName());System.out.println(Thread.currentThread().getName() : 拼团成功);countDownLatch.countDown();orderSucess true;}}}if (!orderSucess) {System.out.println(Thread.currentThread().getName() : 拼团失败已无名额...);}});}// 订单生成线程new Thread(() - {try {countDownLatch.await();System.out.println(Thread.currentThread().getName() : 拼团结束, 订单已生成...);System.out.println(Thread.currentThread().getName() : 拼团人员id ids);} catch (InterruptedException e) {e.printStackTrace();}}, 拼团).start();// 释放线程池threadPool.shutdown();}
}2.2 原理分析
CountDownLatch的底层实现原理也非常简单当线程调用 await() 的时候如果 state 不为 0 则证明任务还没有执行结束await() 就会进入阻塞等待其源码如下
// CountDownLatch - await()
public void await() throws InterruptedException {// 调用内部类sync的acquireSharedInterruptibly方法sync.acquireSharedInterruptibly(1);
}
// CountDownLatch - Sync - acquireSharedInterruptibly
public final void acquireSharedInterruptibly(int arg) throws InterruptedException {// 被中断抛出异常if (Thread.interrupted())throw new InterruptedException();// tryAcquireShared - 判断是否阻塞等待if (tryAcquireShared(arg) 0)// 自旋阻塞(AQS实现)doAcquireSharedInterruptibly(arg);
}
// CountDownLatch - Sync - tryAcquireShared
protected int tryAcquireShared(int acquires) {// 判断当前state是否归零return (getState() 0) ? 1 : -1;
}当线程调用 countDown() 时其实最终是调用了Sync中重写的tryReleaseShared方法该方法以 CAS 的操作来减少 state若更新后state 归零则表示所有的计数任务线程都执行完毕那么在 CountDownLatch 上等待的线程就会被AQS的doReleaseShared方法唤醒并继续向下执行。
// CountDownLatch - countDown()
public void countDown() {sync.releaseShared(1);
}
// CountDownLatch - Sync - AQS - releaseShared
public final boolean releaseShared(int arg) {// 判断递减后计数器是否归零if (tryReleaseShared(arg)) {// 唤醒所有等待线程(AQS实现)doReleaseShared();return true;}return false;
}
// CountDownLatch - Sync - tryReleaseShared
protected boolean tryReleaseShared(int releases) {// Decrement count; signal when transition to zerofor (;;) {// 获取当前stateint c getState();// 若计数器归零则返回false其他什么也不做if (c 0)return false;// CAS更新state递减int nextc c-1;if (compareAndSetState(c, nextc))// 若更新成功则判断新计数值是否归零return nextc 0;}
}3. CyclicBarrier
//
