1.初识AQS:
1.1 简单介绍:
aqs(同步器)是用来构建锁和其他同步工具的基础框架, 它的实现主要依赖于一个int成员变量来标识同步状态, 以及通过一个FIFO队列来构成等待队列。锁(ReentrantLock)或者同步工具(CountDownLatch…)都有一个内部类继承自AbstractQueuedSynchronizer, 并重写几个改变同步状态的方法tryAcquire(), tryRelease()… 这些方法实质上调用的也是getState(), setState(), compareAndSetState()这三个方法。 aqs类的其他方法主要实现了排队和阻塞机制。
AQS采用的是模板方法的设计模式, 将一些方法开放给子类去重写, 而同步器给同步组件提供的模板方法又会去调用被子类重写的方法。
在AQS中, 该方法是开放给子类去重写的。
ReentrantLock中NonfairSync（继承AQS）会重写该方法
AQS中的模板方法acquire()会调用tryAcquire方法。在ReentrantLock中其实调的是NonfairSync中重写的方法。
1.2 AQS中可重写方法 & 模板方法
可重写:
模板方法:
1.3 自定义锁(同ReentrantLock):
class Mutex implements Lock, Serializable {// Our internal helper class// 继承AQS的静态内存类// 重写方法private static class Sync extends AbstractQueuedSynchronizer {// Reports whether in locked stateprotected boolean isHeldExclusively() {return getState() == 1;}// Acquires the lock if state is zeropublic boolean tryAcquire(int acquires) {assert acquires == 1; // Otherwise unusedif (compareAndSetState(0, 1)) {setExclusiveOwnerThread(Thread.currentThread());return true;}return false;}// Releases the lock by setting state to zeroprotected boolean tryRelease(int releases) {assert releases == 1; // Otherwise unusedif (getState() == 0) throw new IllegalMonitorStateException();setExclusiveOwnerThread(null);setState(0);return true;}// Provides a ConditionCondition newCondition() {return new ConditionObject();}}// The sync object does all the hard work. We just forward to it.private final Sync sync = new Sync();//使用同步器的模板方法实现自己的同步语义public void lock() {sync.acquire(1);}public boolean tryLock() {return sync.tryAcquire(1);}public void unlock() {sync.release(1);}public Condition newCondition() {return sync.newCondition();}public boolean isLocked() {return sync.isHeldExclusively();}public boolean hasQueuedThreads() {return sync.hasQueuedThreads();}public void lockInterruptibly() throws InterruptedException {sync.acquireInterruptibly(1);}public boolean tryLock(long timeout, TimeUnit unit)throws InterruptedException {return sync.tryAcquireNanos(1, unit.toNanos(timeout));}public static void main(String[] args) {Mutex mutex = new Mutex();for (int i = 0; i < 10; i++) {Thread thread = new Thread(() -> {mutex.lock();try {Thread.sleep(300);} catch (InterruptedException e) {e.printStackTrace();} finally {mutex.unlock();}});thread.start();}}}
上述例子是一个独占锁, 同一时刻只允许一个线程占用锁.当前线程0正在执行并且占有锁, 其他线程处于wait状态.
2. 深入理解AQS:
2.1 同步队列:
AQS的同步队列是通过链式方式(双向队列)实现的, AQS持有头尾指针管理同步队列, 在AQS内部有一个静态内部类Node.
static final Node SHARED = new Node();//指示节点正在共享模式下等待的标记
static final Node EXCLUSIVE = null;//指示节点正在独占模式下等待的标记
volatile int waitStatus; //节点状态
volatile Node prev; //当前节点/线程的前驱节点
volatile Node next; //当前节点/线程的后继节点
volatile Thread thread;//加入同步队列的线程引用
每一个节点都有如下状态:
static final int CANCELLED = 1//此节点持有的线程被中断, 节点从同步队列中取消
static final int SIGNAL = -1//节点处于等待状态，如果当前节点释放同步状态会通知后继节点，使得后继节点的线程能够运行；
static final int CONDITION = -2//当前节点进入等待队列中
static final int PROPAGATE = -3//表示下一次共享式同步状态获取将会无条件传播下去
static final int INITIAL = 0;//初始状态
2.1.1 ReentrantLock举例:
2.2 源码解读:
2.2.1 独占锁获取:
在AQS中, 独占锁获取同步状态是使用如下方法:
对于tryAcquire方法, 会被子类重写, 我们分析ReentrantLock中的tryAcquire方法。
protected final boolean tryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
if (c == 0) {
// 信号量为0, 可以去获取锁
if (!hasQueuedPredecessors() &&
compareAndSetState(0, acquires)) {
// hasQueuedPredecessors 询问是否有线程在排队
// 如果没有, 自旋的方式获取锁。保证了互斥性。
// 在获取到锁的时候, 修改exclusiveOwnerThread字段为当前线程
setExclusiveOwnerThread(current);
return true;
}
}
else if (current == getExclusiveOwnerThread()) {
// 若不为0, 且exclusiveOwnerThread指向的线程是当前线程
// 还可以获取锁(可重入锁)
int nextc = c + acquires;
if (nextc < 0)
throw new Error("Maximum lock count exceeded");
setState(nextc);
return true;
}
return false;
}
当线程获取独占式锁失败后就会将当前线程加入同步队列，(分析如下方法: addWaiter()和acquireQueued()).
private Node addWaiter(Node mode) {
Node node = new Node(Thread.currentThread(), mode);
Node pred = tail;
if (pred != null) {
node.prev = pred;
if (compareAndSetTail(pred, node)) {
// 自旋给尾部设置节点
pred.next = node;
return node;
}
}
// 尾节点为null时进行入队操作(第一次进队列)
// CAS尾插入节点失败后负责自旋进行尝试
enq(node);
return node;
}
private Node enq(final Node node) {
for (;;) {
Node t = tail;
if (t == null) {
// 在当前线程是第一个加入同步队列时，
//调用compareAndSetHead(new Node())方法，完成链式队列的头结点的初始化
if (compareAndSetHead(new Node()))
tail = head;
} else {
// 自旋不断尝试CAS尾插入节点直至成功为止
node.prev = t;
if (compareAndSetTail(t, node)) {
t.next = node;
return t;
}
}
}
}
当线程进入到同步队列后, 会都做一些尝试去使得自己获取独占锁。 详见acquireQueued()方法。
final boolean acquireQueued(final Node node, int arg) {
boolean failed = true;
try {
boolean interrupted = false;
for (;;) {
final Node p = node.predecessor(); // 当前节点的前驱节点
// 当前节点的前驱节点是头结点并且成功获取了同步状态, 就可以获取独占锁。
if (p == head && tryAcquire(arg)) {
// head节点执行当前节点
setHead(node);
// 释放前驱节点
p.next = null;
failed = false;
return interrupted;
}
// 获取锁失败，线程进入等待状态等待获取独占式锁
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}
/*
当compareAndSetWaitStatus设置失败则说明shouldParkAfterFailedAcquire方法返回false，
然后会在acquireQueued()方法中for (;;)死循环中会继续重试，
直至compareAndSetWaitStatus设置节点状态位为SIGNAL时shouldParkAfterFailedAcquire返回true
才会执行方法parkAndCheckInterrupt()方法
*/
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
int ws = pred.waitStatus;
if (ws == Node.SIGNAL)
// 后继节点属于阻塞状态
return true;
if (ws > 0) {
do {
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
pred.next = node;
} else {
// 使用CAS将节点状态由INITIAL设置成SIGNAL，表示当前线程阻塞
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}
2.2.2 独占锁释放:
首先调用release方法去修改信号量,释放同步状态
protected final boolean tryRelease(int releases) {
// 信号量 -1
int c = getState() - releases;
if (Thread.currentThread() != getExclusiveOwnerThread())
throw new IllegalMonitorStateException();
boolean free = false;
if (c == 0) {
// 所有锁都释放, 修改exclusiveOwnerThread为null
free = true;
setExclusiveOwnerThread(null);
}
setState(c);
return free;
}
如果同步状态释放成功, 且head指向的头结点不为null，并且该节点的状态值不为0的话, 会执行unparkSuccessor()方法。
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;
}
if (s != null)
//后继节点不为null时唤醒该线程
LockSupport.unpark(s.thread);
}
2.2.3 可中断独占锁 & 超时等待独占锁:
可响应中断式锁调用方法lock.lockInterruptibly()。 底层调用的是AQS的acquireInterruptibly方法, 大体逻辑和独占锁获取是一致的, 都是在获取锁失败后, 将节点插入到同步队列中.唯一不同的是, 唯一的区别是当parkAndCheckInterrupt返回true时，即线程阻塞时该线程被中断，代码抛出被中断异常。
超时等待式独占锁通过调用lock.tryLock(timeout,TimeUnit)方式达到超时等待获取锁的效果。底层调用的是AQS的tryAcquireNanos方法。大体逻辑是: a.如果在超时时间内获取锁, 则返回结果. b. 当前线程在超时时间内被打断, 返回结果. c. 超时结束, 任未返回结果则返回false。
private boolean doAcquireNanos(int arg, long nanosTimeout)
throws InterruptedException {
if (nanosTimeout <= 0L)
return false;
// 根据超时时间和当前时间计算出截止时间
final long deadline = System.nanoTime() + nanosTimeout;
final Node node = addWaiter(Node.EXCLUSIVE);
boolean failed = true;
try {
for (;;) {
final Node p = node.predecessor();
if (p == head && tryAcquire(arg)) {
setHead(node);
p.next = null; // help GC
failed = false;
return true;
}
// 重新计算超时时间
nanosTimeout = deadline - System.nanoTime();
if (nanosTimeout <= 0L)
// 已经超时返回false
return false;
if (shouldParkAfterFailedAcquire(p, node) &&
nanosTimeout > spinForTimeoutThreshold)
// 线程阻塞等待
LockSupport.parkNanos(this, nanosTimeout);
// 线程被中断抛出被中断异常
if (Thread.interrupted())
throw new InterruptedException();
}
} finally {
if (failed)
cancelAcquire(node);
}
}
上述逻辑如图所示:
2.2.4 共享锁获取 & 共享锁释放:
2.2.5 可中断共享锁 & 超时等待共享锁:
3. ReentrantLock介绍:
java除了使用关键字synchronized外，还可以使用ReentrantLock实现独占锁的功能。而且ReentrantLock相比synchronized而言功能更加丰富，使用起来更为灵活，也更适合复杂的并发场景。
3.1 与synchronized区别
3.1.1 使用方式:
public void testWait() {
synchronized (this) {
...
}
}
public synchronized void testWait() {
...
}
--------------------------------
private ReentrantLock lock = new ReentrantLock();
public void increment() throws Exception {
try {
lock.lock();
。。。
} finally {
lock.unlock();
}
}
synchronized 作用在方法体或者代码块上, 只需要用关键字包起来即可, 不需要关心锁的释放。
synchronized在代码执行完之后,会自动让线程释放对锁的占用。
ReentrantLock 是new一个对象, 然后显示的去调用上锁和关锁的方法。如果没有手动释放锁, 会导致死锁问题。
3.1.2 底层实现:
synchronized是jvm层的锁, 是java关键字. 借助monitor对象来完成。在同步方法或者同步代码块的头和尾分别加(monitorenter和monitorexit)。而ReentrantLock是一个api(java.util.concurrent.locks.Lock).
synchronized 的实现涉及到锁的升级，具体为无锁、偏向锁、自旋锁、向OS申请重量级锁，ReentrantLock实现则是通过利用AQS(同步队列)+ CAS（CompareAndSwap）自旋机制保证线程操作的原子性和volatile保证数据可见性以实现锁的功能。
3.1.3 是否可中断:
synchronized不可响应中断，一个线程获取不到锁就一直等着；ReentrantLock可以相应中断。可以通过使用tryLock()以及lockInterruptibly()方法进行中断。
public void test3() {
try {
Thread.sleep(1000);
if (lock.tryLock(5, TimeUnit.SECONDS)) {
//lock.lock();
System.out.println("尝试获取5秒 ThreadName = " + Thread.currentThread().getName() + " 尝试获取锁 获得锁了没有呢:" + lock.isHeldByCurrentThread());
lock.unlock();
} else {
System.out.println("尝试获取5秒 ThreadName = " + Thread.currentThread().getName() + " 尝试获取锁失败");
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
使用tryLock()获取锁, 在等待5秒钟还没有返回结果, 则停止获取锁。
---------------------------------------------------------
public class LockInterruptiblyThreadTest {
private ReentrantLock lock = new ReentrantLock();
public void test() {
try {
System.out.println("ThreadName = " + Thread.currentThread().getName() + "等待获取的锁 当前时间戳:" + System.currentTimeMillis());
lock.lockInterruptibly();
System.out.println("ThreadName = " + Thread.currentThread().getName() + "获取的锁 当前时间戳:" + System.currentTimeMillis());
Thread.sleep(10000);
System.out.println("ThreadName = " + Thread.currentThread().getName() + "运行完毕 当前时间戳:" + System.currentTimeMillis());
lock.unlock();
} catch (InterruptedException e) {
System.out.println("ThreadName = " + Thread.currentThread().getName() + " error");
e.printStackTrace();
}
}
public static void main(String[] args) throws InterruptedException {
LockInterruptiblyThreadTest test = new LockInterruptiblyThreadTest();
Runnable runnable = new Runnable() {
public void run() {
test.test();
}
};
Thread thread = new Thread(runnable);
thread.start();
Thread.sleep(1000);
Thread thread1 = new Thread(runnable);
thread1.start();
Thread.sleep(5000);
System.out.println("5秒了...线程1 是否还在等待锁没有:" + test.lock.hasQueuedThread(thread1));
thread1.interrupt();
}
}
lockInterruptibly() 允许在等待时由其它线程调用等待线程的Thread.interrupt方法来中断等待线程的等待而直接返回，
这时不用获取锁，而会抛出一个InterruptedException。
ReentrantLock.lock方法不允许Thread.interrupt中断,即使检测到Thread.isInterrupted,
一样会继续尝试获取锁，失败则继续休眠。只是在最后获取锁成功后再把当前线程置为interrupted状态,然后再中断线程。
3.1.4 是否是公平锁:
synchronized为非公平锁 ReentrantLock则即可以选公平锁也可以选非公平锁，通过构造方法new ReentrantLock时传入boolean值进行选择，为空默认false非公平锁，true为公平锁。
private ReentrantLock lock = new ReentrantLock(false);
public void testMethod() {
lock.lock();
System.out.println("ThreadName= " + Thread.currentThread().getName() + "获得锁");
lock.unlock();
}
公平锁：根据加锁的顺序来分配锁，就是根据谁先执行到lock.lock这句代码来分配的
非公平：随机的，所有加锁的线程都会抢锁
3.1.5 锁是否可以绑定条件Condition:
synchronized不能绑定, 从而在唤醒线程的时候, 是通过Object类的wait()和notify()去随机唤醒一个线程或者唤醒全部线程。ReentrantLock通过绑定Condition结合await()/singal()方法实现线程的精确唤醒。调用不同的Condition对象可以实现精确唤醒。
public class MuitConditionTest {
private ReentrantLock lock = new ReentrantLock();
public Condition c1 = lock.newCondition();
public Condition c2 = lock.newCondition();
public void awaitA() {
try {
lock.lock();
System.out.println("awaitA 时间为" + System.currentTimeMillis());
c1.await();
System.out.println("awaitA ending");
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
public void awaitB() {
try {
lock.lock();
System.out.println("awaitB 时间为" + System.currentTimeMillis());
c2.await();
System.out.println("awaitB ending");
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
public void signal_A() {
try {
lock.lock();
System.out.println("signalA 时间为:" + System.currentTimeMillis());
c1.signalAll();
} finally {
lock.unlock();
}
}
public void signal_B() {
try {
lock.lock();
System.out.println("signalB 时间为:" + System.currentTimeMillis());
c1.signalAll();
} finally {
lock.unlock();
}
}
public static void main(String[] args) throws InterruptedException {
MuitConditionTest conditionTest = new MuitConditionTest();
new Thread(() -> {
conditionTest.awaitA();
}).start();
new Thread(() -> {
conditionTest.awaitB();
}).start();
Thread.sleep(1000);
conditionTest.signal_A();
System.out.println("ending");
}
}
-----------------------------
awaitA 时间为1623210929023
awaitB 时间为1623210929024
signalA 时间为:1623210930027
ending
awaitA ending
方法B还在wait中, 并没有被唤醒。
3.2 使用 & api方法
4. 读写锁
5. CountDownLatch & CyclicBarrier & Semaphore
若有收获，就点个赞吧
0 人点赞
