AQS须要解决下面几个问题:
1.锁状态,怎样保证并发情况下可以安全的更新?
2.当前线程不能获取锁时,放在哪里? AQS是放在一个队列其中
3.怎样提高效率?
AQS的主要职责是当获取不到锁时。将线程放入CLH队列(一种变体)。而且当有线程释放当前拥有的锁时。找出header结点下一个结点(代表一个堵塞线程),并
将它唤醒。
因为AQS放等待堵塞线程的队列用的是CLH变体队列,先大致了解下队列。
AQS的队列有一个header结点,这里AQS仅仅有在第一次有线程尝试获取锁但获取不到时。它才会进行创建。
队列中的每一个结点代表了一个等待锁的线程,用一个Node进行封装。
Node中有一个next指针批向下一结点。这个next指针主要用于之后唤醒后面线程结点。
一个prev指向指向上一结点。
Node结点有一个waitStatus状态,有signal。cancelled,CONDITION。PROPAGATE。0(初始状态)。
当中主要关注signal状态,这个状态用于暗示当前设置了waitStatus为signal的结点线程。告诉它在释放锁时记得唤醒它后面的结点。
prev指针的作用则是用于获取前结点,这里包含设置前结点的waitStatus。
注:当第一次有线程获取不到锁时。要入队列时,会创建一个虚拟结点,header会指向这个结点,然后tail指向这个要获取锁的线程结点A。
之后,仅仅有header后一个结点,也就是刚刚入队的等待锁的线程结点A能够尝试获取锁。
获取到锁后,将header结点指向这个等待锁结点A,释放之前虚拟结点。
拥有锁的线程A。之后调用unlock释放锁,释放锁会将state减1。
当state减到0时。会唤醒header(即A结点)后一个等待锁线程结点。
除了第一次获取锁时。header是指向虚拟结点,之后在线程获取到锁后,header线程指向这个获取到锁的线程结点。
全部能够获取锁的线程结点条件为:header结点的下一个结点。
AQS的设计环绕着在适度的情况尽量不让线程挂起,进行CAS获取锁。
AQS本身有两个核心实现方法acquire及:
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
public final boolean release(int arg) {
if (tryRelease(arg)) {
Node h = head;
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);
return true;
}
return false;
}
AQS获取锁与释放锁调用的是acquire与release方法。
acquire方法的职责是:
假设当前线程不能获取到锁。则将其封装成一个Node放入队列其中,并决定何时将当前线程真正堵塞。
否则获取到锁时,直接返回。
release的职责是:
尝试释放锁(假设当前线程没有拥有对象锁,不能进行兴许操作,即仅仅有拥有锁的线程才干对锁进行release),
成功,则从队列中找出header下一个结点(存储了堵塞了的线程),调用LockSupport的unpark方法将它唤醒。
aquire及release中的tryAcquire与tryRelease方法交由子类去完毕,子类在获取锁及释放锁时添加一些特性,如进行公平锁与非公平锁,超时等特性。
这里全部的堵塞与唤醒操作使用LockSupport的 park(Object blocker)及unpark(Thread thread)方法。
AQS定义了四个方法供子类个性化实现假设可中断,定时获取锁的形式,例如以下:
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();
}
ReentrantLock就是用AQS作为基础框架。内部使用Sync,这个类继承了AQS。
ReentrantLock类结构例如以下:
能够看到ReentrantLock有一个sync实例。然后拥有lock。lockINterruptibly等方法。
其实lock这些方法是给客户调用的,但终于ReentrantLock会将对应的调用交由sync对应方法去处理。
ReentrantLock使用sync,相当于策略模式一样。ReentrantLock为了支持非公平及公平锁,因此在内部同一时候创建了两个sync的子类,分别相应于公平锁与非公平锁实现。
Sync继承图:
接下去看下一个线程请求一个ReentrantLock锁时发生的大致流程:
能够看到线程想要获取锁。在操作1时,这里详细的获取操作是:
查看当前锁state是否为0(说明当前锁未被占有),不是则说明已被其他线程获取。进入can not分支。
假设当前锁state为0。则尝试用cas操作将锁的state改为1。cas成功后再将exclusiveOwnerThread改为当前线程。
ReentrantLock的Sync类的lock代码例如以下:
final void lock() { if (compareAndSetState(0, 1)) setExclusiveOwnerThread(Thread.currentThread()); else acquire(1); }
这里假设当前线程不能马上获取锁时用调用AQS的acquire方法。
先看下AQS的acquire方法,代码例如以下:
public final void acquire(int arg) { if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) selfInterrupt(); }
acquirey方法做了下面几件事情:
调用tryAcquire(子类ReentrantLock去实现)尝试获取锁。假设获取到则退出acquire方法。
假设tryAcquire获取不到锁,这时先通过addWaiter方法将当前线程封装成一个CLH的Node放入CLH队列。而且调用acquireQueued方法决定何时将当前线程堵塞。
接下去分别对acquire方法中的tryAcquire,addWaiter及acquireQueued进行分析。
tryAcquire是AQS提供给子类用于实现自己个性化的获取锁机制。
ReentrantLock中分别有公平锁(FairSync)及非公平锁(NonFairSync)实现。
这里看非公平锁NonFairSync的实现。代码例如以下:
protected final boolean tryAcquire(int acquires) { return nonfairTryAcquire(acquires); }
能够看到这里最后调用的是nonFairTryAcquire方法。它的实如今Sync类主体上。代码例如以下:
/** * Performs non-fair tryLock. tryAcquire is * implemented in subclasses, but both need nonfair * try for trylock method. */ 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()) { int nextc = c + acquires; if (nextc < 0) // overflow throw new Error("Maximum lock count exceeded"); setState(nextc); return true; } return false; }
nonFairTryAcquire的之所以是非公平的,是由于不论什么线程进入这种方法获取锁时都有机会先获取锁,而无论在CLH队列中堵塞(或在进行CAS操作)的线程。
这里会先推断:
假设当前锁对象state为0进入if(c==0)分支,这里立即尝试CAS操作试图将state状态改为被锁状态。假设成功继续将锁对象上的exclusiveOwnerThread改为当前线程。
假设不能CAS成功这里就返回false,表明不能获取到锁。
假设当前锁对象state不为0,进入else if分支,这时依据锁对象上的exclusiveOwnerThread来推断是否是当前线程拥有了这个锁对象,假设是则将锁占有次数(放在state状态上)加1。而且返回true。
在nonFairTryAcquire方法运行完后,返回到AQS的acquire方法:
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
这里假设tryAcquire(arg)返回的是true则表明当前线程已经获取到锁直接返回,否则进行 &&后面的操作。
这里会先运行addWaiter方法,这种方法会先当前线程放入到队列。
代码例如以下:
/** * Creates and enqueues node for current thread and given mode. * * @param mode Node.EXCLUSIVE for exclusive, Node.SHARED for shared * @return the new node */ private Node addWaiter(Node mode) { Node node = new Node(Thread.currentThread(), mode); // Try the fast path of enq; backup to full enq on failure Node pred = tail; if (pred != null) { node.prev = pred; if (compareAndSetTail(pred, node)) { pred.next = node; return node; } } enq(node); return node; }
能够看到先将当前线程封装成一个Node对象,然后尝试将新结点放入到队尾,这里会推断队尾是否存在.
假设存在,则尝试将它放入队尾。
不存在调用enque方法。创建一个新的header结点使header指针指向这个header结点。
tail指针指向这个当前线程结点。
代码例如以下:
private Node enq(final Node node) { for (;;) { Node t = tail; if (t == null) { // Must initialize if (compareAndSetHead(new Node())) tail = head; } else { node.prev = t; if (compareAndSetTail(t, node)) { t.next = node; return t; } } } }
能够看到这种方法终于是创建一个header结点,将header指针指向了header结点。并将header结点的next指针指向当前线程结点。
而tail指针指向这个当前线程结点。
如图:
然后返回到acquire方法:
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
运行acquireQueued方法,这种方法会观察当前在队列中的结点的上一结点看它的状态waitState是否为signal,假设不是则进行一次CAS尝试获取锁,还是不成功的情况下则调用LookSurport的park方法堵塞该线程。
代码例如以下:
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)) { setHead(node); p.next = null; // help GC failed = false; return interrupted; } if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) interrupted = true; } } finally { if (failed) cancelAcquire(node); } }
能够看到对于这部分来讲,仅仅有头结点后面的一个结点线程才有机会用CAS获取锁。即运行if (p == head && tryAcquire(arg)) 。
这里假设当前结点线程不是头结点下一结点或者说CAS获取锁还是失败,则进入shouldParkAfterFailedAcquire方法,看是否该返回true表明该堵塞该线程了。
则进入shouldParkAfterFailedAcquire方法代码例如以下:
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) { int ws = pred.waitStatus; if (ws == Node.SIGNAL) /* * This node has already set status asking a release * to signal it, so it can safely park. */ return true; 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 { /* * waitStatus must be 0 or PROPAGATE. Indicate that we * need a signal, but don't park yet. Caller will need to * retry to make sure it cannot acquire before parking. */ compareAndSetWaitStatus(pred, ws, Node.SIGNAL); } return false; }
能够看到这里假设当前线程结点上一结点的waitStatus不为signal或者大于1时(表明CANCELLED),这时会先设置前一结点的waitStatus为signal。
然后返回false表明先不堵塞当前线程,返回acquireQueued方法会再走一次for循环。
第二次走acquireQueued的for循环时,发现还是不能获取锁,这时shouldParkAfterFailedAcquire会返回true,然后回到acquireQueued方法就调用parkAndCheckInterrupt方法真正堵塞线程了。
这样完毕了线程获取锁的整个流程解析。接下去须要解析释放锁这部分逻辑。相对于线程获取锁简单一点。