面试官：谈一谈java中基于AQS的并发锁原理
我：java中的AQS是指AbstractQueuedSynchronizer类，java中并发锁控制逻辑都是基于这个类来实现的。
面试官：能说一下你用过的基于AQS的并发类有哪些吗？
我：首先是基于AQS在内部类实现了公平锁和非公平锁，具体有3个类：ReentrantLock、ReentrantReadWriteLock、Semaphore，UML类图如下：
还有2个Latch类基于AQS实现了并发控制，他们是CountDownLatch和LimitLatch，UML类图如下：
面试官：谈一下AQS是怎么获取锁的？
我：首先，AbstractQueuedSynchronizer是一个基于FIFO的队列实现的并发控制，队列中的元素通过操作共享资源state来获取和释放锁，state是一个volatile修饰的int类型变量。我以ReentrantLock中独占锁为例，如果有一个线程来获取锁，这时如果队列中没有元素，那就把这个线程加入队列，同时线程申请的数量加入到state变量。如果队列中已经有元素，这个线程入队尾，之后线程中断等待队列前一个元素释放锁后唤醒。
下面的流程是一个获取锁的流程，如果下面的流程返回false，则把当前线程加入到等待队列。
面试官：读过这部分的源代码吗，能不能讲一下？
我：看一下流程中的源码
final void lock() { acquire(1);//获取1个资源 } public final void acquire(int arg) { if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg))//获取锁失败，入队 selfInterrupt(); } 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; } //当前队列中是否有等待时间更长的元素 public final boolean hasQueuedPredecessors() { Node t = tail; Node h = head; Node s; return h != t && ((s = h.next) == null || s.thread != Thread.currentThread()); } 下面的代码是加入到等待队列的过程 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; // help GC failed = false; return interrupted; } 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; if (ws > 0) {//注意：如果前置节点不通知，继续往前查找，找到一个可以通知的节点作为当前节点的node节点 do { node.prev = pred = pred.prev; } while (pred.waitStatus > 0); pred.next = node; } else {//把前置节点的状态设置为通知状态-1 compareAndSetWaitStatus(pred, ws, Node.SIGNAL); } return false; } 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; } } enq(node); return node; } private Node enq(final Node node) { for (;;) { Node t = tail; if (t == null) { //队列为空，初始化队列，head=tail=node if (compareAndSetHead(new Node())) tail = head; } else {//队列非空，放到队尾 node.prev = t; if (compareAndSetTail(t, node)) { t.next = node; return t; } } } }
这儿要注意：队列上的元素是Node对象，node中定义了waitStatus变量，有4个状态，
static final int CANCELLED = 1;//当前线程已经取消锁等待 static final int SIGNAL = -1;//下一个节点需要被唤醒 static final int CONDITION = -2;//当前线程正在等待condition，这个状态需要跟condition配合使用 static final int PROPAGATE = -3;//释放锁的时候使用，这个状态只给头节点，并且要不断传播下去
面试官：再说一下释放锁的过程？
我：锁的释放过程比较简单，还是以ReentrantLock为例。首先尝试释放锁(state变量中减去1)，把当前锁的拥有者置空，通知队列中下一个节点。整个流程入下：
面试官：这部分的源代码能不能讲一下？
我：以ReentrantLock为例，源码入下：
public final boolean release(int arg) { 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; } protected final void setExclusiveOwnerThread(Thread thread) { exclusiveOwnerThread = thread; } private void unparkSuccessor(Node node) { //把当前节点的等待状态置为0 int ws = node.waitStatus; if (ws < 0) compareAndSetWaitStatus(node, ws, 0); //如果队列中下一个节点为空或者下一个节点等待状态是取消状态(1)，则从队尾开始查找，知道找到一个非空并且等待状态小于0的节点 Node s = node.next; if (s == null || s.waitStatus > 0) { s = null; for (Node t = tail; t != null && t != node; t = t.prev)//为什么从队尾开始查找，这样遍历整个队列啊？但是如果当前节点的下一个是null，那就始终找不到下下个节点了，必须从队尾找。 if (t.waitStatus <= 0) s = t; } if (s != null)//唤醒等待线程 LockSupport.unpark(s.thread); }
面试官：上面你讲的是独占锁，那AQS中的共享锁怎么用的？
我:首先共享锁是指多个线程可以同时使用这个锁，AQS中的使用是只要不超过共享锁允许的总数，都可以获取到。在获取读锁时，首先尝试获取共享锁，如果获取失败，入队后等待获取。以ReentrantReadWriteLock为例，获取共享锁流程入下：
从上面这个流程可以看到，如果获取不到锁，就会进入fullTryAcquireShared，这个方式是在死循环中不断尝试获取到锁，直到成功。
面试官：这部分的源代码能介绍一下吗？
我：源码入下
public final void acquireShared(int arg) { if (tryAcquireShared(arg) < 0)//先尝试获取，获取失败，则进入队列等待获取 doAcquireShared(arg); } protected final int tryAcquireShared(int unused) { Thread current = Thread.currentThread(); int c = getState(); if (exclusiveCount(c) != 0 && getExclusiveOwnerThread() != current)//其他线程已经获取到排它锁 return -1; int r = sharedCount(c); if (!readerShouldBlock() && //是否需要阻塞，在公平锁中，如果队列中已经有元素了那就返回true；在非公平锁中，队列中第二个元素不是共享节点，返回true r < MAX_COUNT && compareAndSetState(c, c + SHARED_UNIT)) {//cas失败 if (r == 0) { firstReader = current; firstReaderHoldCount = 1; } else if (firstReader == current) { firstReaderHoldCount++; } else { HoldCounter rh = cachedHoldCounter; if (rh == null || rh.tid != getThreadId(current)) cachedHoldCounter = rh = readHolds.get(); else if (rh.count == 0) readHolds.set(rh); rh.count++; } return 1; } return fullTryAcquireShared(current); } final int fullTryAcquireShared(Thread current) {//下面的方法前面都讲过了 HoldCounter rh = null; for (;;) { int c = getState(); if (exclusiveCount(c) != 0) { if (getExclusiveOwnerThread() != current) return -1; // else we hold the exclusive lock; blocking here // would cause deadlock. } else if (readerShouldBlock()) { // Make sure we're not acquiring read lock reentrantly if (firstReader == current) { // assert firstReaderHoldCount > 0; } else { if (rh == null) { rh = cachedHoldCounter; if (rh == null || rh.tid != getThreadId(current)) { rh = readHolds.get(); if (rh.count == 0) readHolds.remove(); } } if (rh.count == 0) return -1; } } if (sharedCount(c) == MAX_COUNT) throw new Error("Maximum lock count exceeded"); if (compareAndSetState(c, c + SHARED_UNIT)) { if (sharedCount(c) == 0) { firstReader = current; firstReaderHoldCount = 1; } else if (firstReader == current) { firstReaderHoldCount++; } else { if (rh == null) rh = cachedHoldCounter; if (rh == null || rh.tid != getThreadId(current)) rh = readHolds.get(); else if (rh.count == 0) readHolds.set(rh); rh.count++; cachedHoldCounter = rh; // cache for release } return 1; } } } private void doAcquireShared(int arg) { final Node node = addWaiter(Node.SHARED);//共享模式入队 boolean failed = true; try { boolean interrupted = false; for (;;) { final Node p = node.predecessor(); if (p == head) { int r = tryAcquireShared(arg); if (r >= 0) {//获取成功 setHeadAndPropagate(node, r); p.next = null; // help GC if (interrupted) selfInterrupt(); failed = false; return; } } if (shouldParkAfterFailedAcquire(p, node) && //失败后等待前置节点通知 parkAndCheckInterrupt()) //睡眠后等待唤醒 interrupted = true; } } finally { if (failed) cancelAcquire(node);//不再获取 } } private void setHeadAndPropagate(Node node, int propagate) { Node h = head; // Record old head for check below setHead(node);//自己成功头结点从而唤醒 if (propagate > 0 || h == null || h.waitStatus < 0 || (h = head) == null || h.waitStatus < 0) { Node s = node.next; if (s == null || s.isShared()) doReleaseShared();//唤醒下一个节点 } }
面试官：好的，共享锁的释放流程是怎样的？
我：跟共享锁的获取流程一样，先尝试释放(state变量中减去1)，成功后唤醒队列中下一个等待线程
这部分代码如下：
public final boolean releaseShared(int arg) { if (tryReleaseShared(arg)) { doReleaseShared(); return true; } return false; } protected final boolean tryReleaseShared(int unused) { Thread current = Thread.currentThread(); if (firstReader == current) {//当前线程是第一个获取到锁的线程 // assert firstReaderHoldCount > 0; if (firstReaderHoldCount == 1) firstReader = null; else firstReaderHoldCount--; } else { HoldCounter rh = cachedHoldCounter; if (rh == null || rh.tid != getThreadId(current)) rh = readHolds.get(); int count = rh.count; if (count <= 1) { readHolds.remove(); if (count <= 0) throw unmatchedUnlockException(); } --rh.count;//释放一个count数减1 } for (;;) { int c = getState(); int nextc = c - SHARED_UNIT; if (compareAndSetState(c, nextc))//CAS修改state值 // Releasing the read lock has no effect on readers, // but it may allow waiting writers to proceed if // both read and write locks are now free. return nextc == 0; } } private void doReleaseShared() { for (;;) { Node h = head; if (h != null && h != tail) { int ws = h.waitStatus; if (ws == Node.SIGNAL) { if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))//当前节点waiteStatus置为0 continue; // loop to recheck cases unparkSuccessor(h);//唤醒下一个节点 } else if (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))//头节点waiteStatus置为-3 continue; // loop on failed CAS } if (h == head) // loop if head changed break; } }
面试官:前面你提到了公平锁和非公平锁，这2个有什么区别呢？
我：一般情况下，公平锁是指当前线程在释放锁的时候，会通知一个等待队列中等待时间最长的线程，而非公平锁，当前线程释放锁的时候，会随机通知一个线程。非公平锁代码如下：
final void lock() { if (compareAndSetState(0, 1)) setExclusiveOwnerThread(Thread.currentThread()); else acquire(1); }
在ReentrantLock中，公平锁和非公平锁的不同就是非公平锁会直接尝试compareAndSetState(0, 1)，失败后才走获取锁流程。而公平锁直接走获取锁流程。
在ReentrantReadWriteLock中，公平锁只要队列中有其他线程占用锁，读写锁就需要阻塞，而非公平锁中写锁不阻塞，读锁只有队列中第一个排队等待线程使用独占锁时才阻塞，代码如下
static final class NonfairSync extends Sync { final boolean writerShouldBlock() { return false; // writers can always barge } final boolean readerShouldBlock() { return apparentlyFirstQueuedIsExclusive();//队列中第一个等待线程以独占方式等待锁 } } final boolean apparentlyFirstQueuedIsExclusive() { Node h, s; return (h = head) != null && (s = h.next) != null && !s.isShared() && s.thread != null; } static final class FairSync extends Sync { final boolean writerShouldBlock() { return hasQueuedPredecessors();//队列中有元素正在使用锁或者第一个等待的线程不是当前线程 } final boolean readerShouldBlock() { return hasQueuedPredecessors(); } }
面试官:好的，再谈一谈CountDownLatch的使用和原理？
我：上面的原理理解了之后，CountDownLatch的使用就非常简单了。示例代码如下：
public class Test { public static void main(String[] args) throws InterruptedException { CountDownLatch latch = new CountDownLatch(3);//将state置为3 for (int i = 0; i < 3; i++){ LocalThreadPool.run(new WorkerThread(latch)); } System.out.println("latch test begin"); latch.await();//获取到共享锁 System.out.println("latch test end"); } static class WorkerThread implements Runnable { private CountDownLatch latch; public WorkerThread(CountDownLatch latch) { this.latch = latch; } @Override public void run() { System.out.println(Thread.currentThread().getName() + " finished"); latch.countDown();//将state值减1，最后一个线程将state减为0后，释放上面await方法获取到的共享锁。 } } }
上面输出结果如下：
latch test begin pool-1-thread-3 finished pool-1-thread-2 finished pool-1-thread-1 finished latch test end
初始化的时候，批量将state值设置为线程数量，然后通过await获取共享锁。每个线程执行完成后调用释放锁的方法将state减1，最后一个线程将state减为0后返回true，这时CountDownLatch就会释放掉共享锁。看下面源代码
public final boolean releaseShared(int arg) { if (tryReleaseShared(arg)) {//state减为0才会执行这个逻辑 doReleaseShared();//释放await()方法获取到的共享锁 return true; } return false; } protected boolean tryReleaseShared(int releases) { // Decrement count; signal when transition to zero for (;;) { int c = getState(); if (c == 0) return false; int nextc = c - 1; if (compareAndSetState(c, nextc)) return nextc == 0;//减为0时返回true } }
面试官：那CountDownLatch在await方法中加入参数是怎么实现等待超时的呢？
我：这个就是在一个死循环中不断获取锁，直到超时，这个是AQS本身就支持的逻辑，代码如下：
private boolean doAcquireSharedNanos(int arg, long nanosTimeout) throws InterruptedException { if (nanosTimeout <= 0L) return false; final long deadline = System.nanoTime() + nanosTimeout;//设置申请锁失败时间 final Node node = addWaiter(Node.SHARED); try { for (;;) { final Node p = node.predecessor(); if (p == head) { int r = tryAcquireShared(arg); if (r >= 0) { setHeadAndPropagate(node, r); p.next = null; // help GC return true; } } nanosTimeout = deadline - System.nanoTime(); if (nanosTimeout <= 0L) {//超时，取消获取锁 cancelAcquire(node); return false; } if (shouldParkAfterFailedAcquire(p, node) && nanosTimeout > SPIN_FOR_TIMEOUT_THRESHOLD) LockSupport.parkNanos(this, nanosTimeout); if (Thread.interrupted()) throw new InterruptedException(); } } catch (Throwable t) { cancelAcquire(node); throw t; } }
面试官：那CountDownLatch设置等待超时时间，有什么好处呢？
我：这个主要作用是设置主线程等待时间，以免长期阻塞主线程。从上面源代码看出这个并不影响任务线程的执行，不过如果等待任务执行线程执行完成后再做一些日志或者通知，就会失败，因为超时后直接就会调用这些日志或通知，不一定真的所有任务都完成了。
比如下面的代码：
public class Test { public static void main(String[] args) throws InterruptedException { CountDownLatch latch = new CountDownLatch(3); for (int i = 0; i < 3; i++){ LocalThreadPool.run(new WorkerThread(latch)); } System.out.println("latch test begin"); latch.await(1, TimeUnit.SECONDS); System.out.println("latch test end"); } static class WorkerThread implements Runnable { private CountDownLatch latch; public WorkerThread(CountDownLatch latch) { this.latch = latch; } @Override public void run() { try { Thread.sleep(3000); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println(Thread.currentThread().getName() + " finished"); latch.countDown(); } } }
输出结果是
latch test begin latch test end pool-1-thread-2 finished pool-1-thread-1 finished pool-1-thread-3 finished
面试官：再聊一聊Semaphore的使用？
我：Semaphore的使用也是基于AQS的，它更像一个限流器，初始化是给state赋值，每次执行任务时Semaphore获取共享锁并且将state值减1。如果state值小于0则入队等待。任务执行完成后，Semaphore释放锁，首先state值加1，如果state小于0，则通知队列中下一个等待线程。我们可以用Semaphore实现一个数据库连接池，代码如下：
class ConnectionPool { final List<Connection> pool = new ArrayList<>(10);//这儿没有考虑线程安全 final Semaphore sem = new Semaphore(10);//将state设置为10 //构造函数初始化连接池 ConnectionPool(int size) { for (int i = 0; i < size; i++) { pool.add(new Connection() { //省略实现代码 }); } } //执行任务过程 void doExecution() throws InterruptedException { Connection connection = null; sem.acquire();//获取可中断共享锁，如果state小于1，则进入锁等待队列 try { connection = pool.remove(0);//从连接池取连接 connection.commit();//执行事务 } catch (SQLException e) { e.printStackTrace(); } finally { pool.add(connection);//用完后归还连接池 sem.release();//释放锁,将state值加1，如果state小于0，去等待队列唤醒下一个等待线程 } } }
面试官：再聊一聊Condition的使用？
我：Condition类的作用是可以实现wait-notify机制，比synchronized灵活，可以在一个Lock上创建多个Condition，线程选择注册不同的Condition进项调度。假设我们现在有一个需求，2个线程轮流打印出1到10的数字，这个时候我们可以用信号量来做，代码如下：
public class ConditionTest { private ReentrantLock lock = new ReentrantLock(); public Condition conditionA = lock.newCondition(); public Condition conditionB = lock.newCondition(); private int count = 0; public String printThreadA() throws InterruptedException { while (count < 10){ lock.lock(); System.out.println(Thread.currentThread().getName() + ":" + (++count)); conditionB.signal();//打印后唤醒B线程打印 conditionA.await();//打印后等待B线程唤醒 lock.unlock(); } return null; } public String printThreadB() throws InterruptedException { while (count < 10){ lock.lock(); System.out.println(Thread.currentThread().getName() + ":" + (++count)); conditionA.signal();//打印后唤醒A线程打印 conditionB.await();//打印后等待A线程唤醒 lock.unlock(); } return null; } public void printTenNumber(){ LocalThreadPool.call(() -> printThreadA()); LocalThreadPool.call(() -> printThreadB()); } public static void main(String[] args){ ConditionTest conditionTest = new ConditionTest(); conditionTest.printTenNumber(); } }
上面signal()方法具体实现为将当前线程从condition等待队列转入锁等待队列队尾等待前一个节点唤醒，await()方法的具体实现为在condition等待队列队尾新加一个等待者，释放锁并且唤醒下一个等待线程。所以，condition的await和signal本质上是当前线程在condition等待队列和锁等待队列直接的转移。
代码实现如下：
public final void signal() { if (!isHeldExclusively()) throw new IllegalMonitorStateException(); Node first = firstWaiter; if (first != null) doSignal(first); } private void doSignal(Node first) { do { if ( (firstWaiter = first.nextWaiter) == null) lastWaiter = null; first.nextWaiter = null; } while (!transferForSignal(first) && (first = firstWaiter) != null); } final boolean transferForSignal(Node node) { //waitStatus置为初始化 if (!node.compareAndSetWaitStatus(Node.CONDITION, 0)) return false; //放入锁等待队列队尾并且把前置节点置为SIGNAL，以通知当前线程 Node p = enq(node); int ws = p.waitStatus; if (ws > 0 || !p.compareAndSetWaitStatus(ws, Node.SIGNAL)) LockSupport.unpark(node.thread); return true; } public final void await() throws InterruptedException { if (Thread.interrupted()) throw new InterruptedException(); Node node = addConditionWaiter();//等待队列队尾新加一个等待者 int savedState = fullyRelease(node);//释放锁并且唤醒下一个等待线程 //删除部分代码 }
面试官：最后一个问题，如果我们自己实现一个AQS的类，需要实现哪些方法呢？
我：AQS类中以下5个方法没有实现，都是throw new UnsupportedOperationException()，这些方法留给自定义子类来实现。
protected boolean tryAcquire(int arg)：独占方式获取锁 protected boolean tryRelease(int arg)：独占方式释放锁 protected int tryAcquireShared(int arg)：共享方式获取锁 protected boolean tryReleaseShared(int arg)：共享方式释放锁 isHeldExclusively()：判断当前线程是否正在独占锁
面试官：恭喜你，通过了。
