AQS(AbstractQueueSynchronizer,抽象的队列式同步器),它提供了一个基于 FIFO 队列,可以用于构建锁或者其他相关同步器的基础框架,许多同步类的实现都依赖于它,比如 ReentrantLock、Semaphore、CountDownLatch
AQS 实现了 3 点基本功能:
我们来看一下 java.util.concurrent.locks 的类关系图
上图中,Lock 的实现类其实都是构建在 AQS 上的,但为何没用 UML 线表示它们之间的关系呢?这是因为每个 Lock 实现类都持有自己内部类 Sync 的实例,而这个 Sync 才是继承自 AQS 的。那为何要实现不同的 Sync 呢?这是因为不同的 Lock 的用途不同
多线程环境下,线程之间可以通过某种状态来同步,比如只有当状态满足某种条件,才能触发线程执行某种操作,能实现这个特性的可以称之为同步器
AQS 里有一个最关键的属性 private volatile int state,可以理解将它为资源数量的抽象。AQS 提供了 getState 和 setState 方法,还有一个线程安全的 compareAndSetState 方法,它利用了 Unsafe 的 CAS 操作,可以做到在并发场景下对 state 进行原子性的修改,并且可以获得修改结果。正式因为这个特性,可以使用 AQS 构建同步器
AQS 是一个抽象类,需要被子类继承并且重写其中的一些方法,官方对此做了一些说明:AQS 的子类必须重写那些会更改 state 的 protected 方法,以及定义 state 的何种状态意味着需要阻塞。比如,如果要实现一个锁,则 state 可以定义为:0 表示未锁定,1 表示锁定。如果要实现信号量,state 可以表示资源甚于
为此,AQS 提供了以下方法:
AQS 将阻塞队列线程封装到了一个内部类 Node 里,并维护了一个 CLH Node FIFO 队列。CLH 队列是一个非阻塞的 FIFO 队列,也就是说,往里面插入或移除一个节点的时候,在并发条件下不会阻塞,而是通过自旋锁和 CAS 保证原子性,实现无锁且快速的修改操作
以 ReentrantLock 的加锁过程为例:
读写锁定义为:一个资源能够被多个读线程访问,或者被一个写线程访问,并且读写线程不能同时访问
在读多写少的场景下,读写锁具有较高的性能体现
package com.sisyphus.ReentrantReadWriteLock;import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;class MyResource{Map map = new HashMap<>();Lock lock = new ReentrantLock();ReadWriteLock rwLock = new ReentrantReadWriteLock();public void write(String key, String value){rwLock.writeLock().lock();try{System.out.println(Thread.currentThread().getName() + "\t" + "正在写入");map.put(key, value);try{TimeUnit.MILLISECONDS.sleep(500);}catch (InterruptedException e){e.printStackTrace();}System.out.println(Thread.currentThread().getName() + "\t" + "完成写入" + value);}finally {rwLock.writeLock().unlock();}}public void read(String key){rwLock.readLock().lock();try{System.out.println(Thread.currentThread().getName() + "\t" + "正在读取");String result = map.get(key);try{TimeUnit.MILLISECONDS.sleep(200);}catch (InterruptedException e){e.printStackTrace();}System.out.println(Thread.currentThread().getName() + "\t" + "完成读取" + result);}finally {rwLock.readLock().unlock();}}
}public class ReentrantReadWriteLockDemo {public static void main(String[] args) {MyResource myResource = new MyResource();for(int i = 0; i < 10; i++){int finalI = i;new Thread(()->{myResource.write(finalI + "", finalI + "");},String.valueOf(i)).start();}for(int i = 0; i < 10; i++){int finalI = i;new Thread(()->{myResource.read(finalI + "");},String.valueOf(i)).start();}for(int i = 0; i < 3; i++){int finalI = i;new Thread(()->{myResource.write(finalI + "", finalI + "");},"新写线程" + String.valueOf(i)).start();}}
}
写线程饥饿
当读锁被获取时,写锁无法获取,必须等待读锁释放。如果读线程太多,那么写线程将一直被阻塞,使用“公平锁”可以一定程度上缓解这个问题,但是“公平锁”是以牺牲系统吞吐量为代价的
锁降级
锁的颗粒度减小叫锁升级,锁的颗粒度增大叫锁降级。遵循获取写锁、获取读锁再释放写锁的次序,写锁能够降级为读锁。但是无法从读锁升级为写锁,这也是会造成写线程饥饿的原因
StampedLock 是 JDK1.8 中新增的一个读写锁,是对 JDK1.5 中的 ReentrantReadWriteLock 的优化
StampedLock 有一个重要属性 long stamp,代表了锁的状态。所有获取锁的方法,都会返回一个 stamp,stamp 为 0表示失败;所有释放锁的方法,都需要传入一个 stamp,并且这个 stamp 必须是和成功获取锁得到的 stamp 一致
StampLock 是不可重入的,如果一个线程已经持有了写锁,再去获取写锁将会导致死锁
StampLock 有三种访问模式:
package com.sisyphus.StampedLockDemo;import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.StampedLock;public class StampedLockDemo {static int number = 1;static StampedLock stampedLock = new StampedLock();public void write(){long stamp = stampedLock.writeLock();System.out.println(Thread.currentThread().getName() + "\t" + "写线程准备修改");try{number = number + 1;}finally {stampedLock.unlock(stamp);}System.out.println(Thread.currentThread().getName() + "\t" + "写线程结束修改");}//悲观读public void read(){long stamp = stampedLock.readLock();System.out.println(Thread.currentThread().getName() + "\t" + "come in readLock codeBlock lock, 4 seconds continue ...");for (int i = 0; i < 4; i++){try{TimeUnit.SECONDS.sleep(1);}catch (InterruptedException e){e.printStackTrace();}System.out.println(Thread.currentThread().getName() + "\t" + "正在读取中……");}try{int result = number;System.out.println(Thread.currentThread().getName() + "\t" + "获得成员变量值result:" + result);System.out.println("写线程没有修改成功,读锁未释放,写锁无法接入,传统的读写互斥");}finally {stampedLock.unlockRead(stamp);}}//乐观读public void tryOptimisticRead(){long stamp = stampedLock.tryOptimisticRead();int result = number;//故意间隔 4 秒钟,乐观认为读取过程中没有其他线程修改过 number 值System.out.println("4 秒前 stampedLock.validate 方法值(true无修改,false有修改)" + "\t" + stampedLock.validate(stamp));for(int i = 0; i < 4; i++){try{TimeUnit.SECONDS.sleep(1);}catch (InterruptedException e){e.printStackTrace();}System.out.println(Thread.currentThread().getName() + "\t" + "正在读取..." + i + " 秒后 stampedLock.validate 方法值(true无修改,false有修改)" + stampedLock.validate(stamp));}if(!stampedLock.validate(stamp)){System.out.println("有人修改过-----有写操作");stamp = stampedLock.readLock();try{System.out.println("从乐观读升级为悲观读");result = number;System.out.println("重新悲观读后 result:" + result);}finally {stampedLock.unlockRead(stamp);}}System.out.println(Thread.currentThread().getName() + "\t" + " finally value:" + result);}public static void main(String[] args) {StampedLockDemo resource = new StampedLockDemo();/*传统版new Thread(()->{resource.read();},"readThread").start();try{TimeUnit.SECONDS.sleep(1);}catch (InterruptedException e){e.printStackTrace();}new Thread(()->{System.out.println(Thread.currentThread().getName() + "\t" + "-----come in");resource.write();},"writeThread").start();try{TimeUnit.SECONDS.sleep(1);}catch (InterruptedException e){e.printStackTrace();}System.out.println(Thread.currentThread().getName() + "\t" + "number:" + number);*/new Thread(()->{resource.tryOptimisticRead();},"readThread").start();//暂停 2 秒钟,演示读过程可以介入写操作try{TimeUnit.SECONDS.sleep(2);}catch (InterruptedException e){e.printStackTrace();}new Thread(()->{System.out.println(Thread.currentThread().getName() + "\t" + "-----come in");resource.write();},"writeThread").start();}
}