AbstractQueuedSynchronizer是基于一个FIFO双向链队列 ==CLH队列==,用于构建锁或者同步装置的类,也称为Java同步器,ReentrantLock的公平锁与非公平锁就是由该同步器构成,链队列结构图如下。
你可以理解为银行ATM机取钱,一个人先去取,获取到了锁,在这个时间内其他线程处于阻塞状态,只有等他取完钱了,他走了,释放了锁,排在它后面的人才可以获取到释放的锁并进行取钱。
该同步器利用一个int值表示状态,实现方式是==使用内部类继承该同步器的方式==实现它的tryRelease、tryAcquire等方法管理状态,管理状态使用以下三个方法:
节点包含的状态有:
节点其他信息:
Node prev | 前驱节点 |
---|---|
Node next | 后继节点 |
Node nextWaiter | 存储condition队列中的后继节点 |
Thread thread | 入队列时的当前线程 |
锁在一个时间点只能被一个线程锁占有,AQS实现的ReentrantLock,又分为公平锁和非公平锁
公平锁
保障了多线程下各线程获取锁的顺序,先到的线程优先获取锁
非公平锁
加锁时不考虑排队等待问题,直接尝试获取锁,获取不到自动到队尾等待
锁在一个时间点可以被多个线程同时获取,AQS实现的CountDownLatch、ReadWriteLock
final void lock() { if (compareAndSetState(0, 1)) setExclusiveOwnerThread(Thread.currentThread()); else acquire(1); }
根据指定状态获取,能获取到 执行compareAndSetState方法设置新状态
public final void acquire(int arg) { //tryAcquire成功的话 acquire结束; if (!tryAcquire(arg) && //AcquireQueued方法进行阻塞等待,直到获取锁为止 //addWaiter把当前线程添加到队列尾部 acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) //tryAcquire失败并且acquiredQueued成功的话把当前线程中断 selfInterrupt(); }
final boolean nonfairTryAcquire(int acquires) { final Thread current = Thread.currentThread(); //获取状态 int c = getState(); //如果该锁未被任何线程占有,该锁能被当前线程获取 if (c == 0) { if (compareAndSetState(0, acquires)) { setExclusiveOwnerThread(current); return true; } } //若被获取到 查看是否被当前线程 else if (current == getExclusiveOwnerThread()) { //是当前线程的话再次获取,计数+1 int nextc = c + acquires; if (nextc < 0) // overflow throw new Error("Maximum lock count exceeded"); setState(nextc); return true; } return false; }
在tryAcquire方法中使用了同步器提供的对state操作的方法,利用CAS原理保证只有一个线程能够对状态进行成功修改,而没有成功修改的线程将进入队列排队。
AcquireQueued方法进行阻塞等待,直到获取锁为止
//node为null,排他方式阻塞等待 final boolean acquireQueued(final Node node, int arg) { boolean failed = true; try { boolean interrupted = false; for (;;) { //p为当前节点的前一个节点 final Node p = node.predecessor(); //如果p为头结点并且获取成功就把当前线节点设置为头结点 if (p == head && tryAcquire(arg)) { setHead(node); p.next = null; // help GC failed = false; return interrupted; } //如果线程需要被阻塞 则interrputr为true if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) interrupted = true; } } finally { if (failed) cancelAcquire(node); } }
判断线程是否需要阻塞
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) { int ws = pred.waitStatus; //线程需要运行 if (ws == Node.SIGNAL) return true; //ws>0 处于CANCEL状态的线程 if (ws > 0) { //把这些线程从队列中清除 do { node.prev = pred = pred.prev; } while (pred.waitStatus > 0); pred.next = node; } else { //把等待的设置为运行状态 compareAndSetWaitStatus(pred, ws, Node.SIGNAL); } return false; }
// park方法让其他线程处于等待状态 private final boolean parkAndCheckInterrupt() { LockSupport.park(this); return Thread.interrupted(); }
调用release释放一个锁
public void unlock() { //释放一个锁 sync.release(1); }
释放锁
public final boolean release(int arg) { //调用tryRelease尝试释放一个锁 if (tryRelease(arg)) { Node h = head; //释放成功后 if (h != null && h.waitStatus != 0) unparkSuccessor(h); 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; }
锁释放后唤醒线程,一同竞争CPU资源
private void unparkSuccessor(Node node) { //获取当前节点状态 int ws = node.waitStatus; //把状态设置为等待获取锁状态 if (ws < 0) compareAndSetWaitStatus(node, ws, 0); //如果没有下一个节点或者下一个节点状态为CANCEL,则把它们清除 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中unpark方法,唤醒后一个节点 if (s != null) LockSupport.unpark(s.thread); }
调用UNSAFE的本地方法unpark唤醒线程
public static void unpark(Thread thread) { if (thread != null) UNSAFE.unpark(thread); }
AQS实现ReentrantLock公平锁与非公平锁最大的区别在下面这段代码:
源码如下:
public final boolean hasQueuedPredecessors() { // The correctness of this depends on head being initialized // before tail and on head.next being accurate if the current // thread is first in queue. Node t = tail; // Read fields in reverse initialization order Node h = head; Node s; return h != t && ((s = h.next) == null || s.thread != Thread.currentThread()); }
==判断"当前线程"是不是在CLH队列的队首,来实现公平性==。
public class MyAQSLock implements Lock { private final Sync sync; public MyAQSLock() { sync = new Sync(); } @Override public void lock() { sync.acquire(1); } @Override public void lockInterruptibly() throws InterruptedException { sync.acquireInterruptibly(1); } @Override public boolean tryLock() { return sync.tryAcquire(1); } @Override public boolean tryLock(long time, TimeUnit unit) throws InterruptedException { return sync.tryAcquireNanos(1,unit.toNanos(time)); } @Override public void unlock() { sync.release(1); } @Override public Condition newCondition() { return sync.newCondition(); } /** * 把lock、unlock实现使用AQS构建为内部类 */ private class Sync extends AbstractQueuedSynchronizer{ Condition newCondition(){ return new ConditionObject(); } @Override protected boolean tryAcquire(int arg) { //第一个线程进来拿到锁 int state = getState(); //用于重入锁判断 Thread current = Thread.currentThread(); if(state==0){ if(compareAndSetState(0,arg)){ setExclusiveOwnerThread(Thread.currentThread()); return true; } } //重入锁判断 当前线程和独占锁线程相同,则再次获取 else if(current==getExclusiveOwnerThread()){ int next = state+arg; if(next<0){ throw new RuntimeException(); } setState(next); return true; } return false; } /** * 可重入释放锁 * @param arg * @return */ @Override protected boolean tryRelease(int arg) { if(Thread.currentThread()!=getExclusiveOwnerThread()){ throw new RuntimeException(); } int state = getState()-arg; if(state==0){ setExclusiveOwnerThread(null); setState(0); return true; } setState(0); return false; } } }
public class TestAQSLock2 { MyAQSLock myLock = new MyAQSLock(); private int value; private int value2; public int a(){ myLock.lock(); try { b(); return value++; }finally { myLock.unlock(); } } public void b(){ myLock.lock(); try { System.out.println(++value2); }finally { myLock.unlock(); } } public static void main(String[] args) { TestAQSLock2 myLock = new TestAQSLock2(); for(int i=0;i<50;i++) { new Thread(() -> { System.out.println(Thread.currentThread().getName() + ":" + myLock.a()); }).start(); } } }