标签:
JAVA AQS的全称为(AbstractQueuedSynchronizer),用于JAVA多线程的开发,从名称我们也可以看出,它实现了同步的队列,而这个队列是指线程队列。AQS类在java.util.concurrent.locks下面。
AQS和CAS作为JAVA5之后非常重要的特性,能在并发应用中提高程序性能,具体要就实际情况使用,因为JVM也在一直优化synchronized关键字,在JAVA7之后其性能也趋于稳定,不会随着线程数增加而导致性能骤降(具体可以取网上搜索对比数据)。
总之,一般情况下还是建议用synchronized
CAS(CompareAndSet)是最小粒度的操作,保证了原子性,通过硬件指令集实现。简单来说,CAS有3个操作数,内存值V,旧的预期值A,要修改的新值B。当且仅当预期值A和内存值V相同时,将内存值V修改为B,否则返回V。
基于此,我们才能完成非阻塞同步操作(当然还有一些其他原子命令,例如FAI,LL/SC等),目的就是用乐观锁来换取性能的提升。
为什么要说CAS呢?因为AQS也是基于CAS实现的,下面进入正题,我们通过源码来具体分析下AQS的实现:
-----------------------------------------------------------------------------华丽的分隔线---------------------------------------------------------------------------------------
AQS的包结构如下
继承的子类有
我们从ReentrantLock来分析AQS的实现原理:
ReentrantLock是一个可重入的排他锁。可重入指当前拥有锁的对象可重复进入同步区域,防止重复操作,例如网页登陆时重复点击按钮。排他锁就是指该锁(Lock)是互斥锁,只允许一个对象拥有锁。
先看看ReentrantLock的类结构
可以看到有3个内部类Sync,NonfairSync和FairSync。其中NonfairSync和FairSync均继承Sync,而Sync继承了AQS。NonfairSync和FairSync的区别在于当一个线程释放了锁的时候,队列里的其他线程是否按照FIFO的规则去获取锁的。
换句话说,FairSync能够保证先到的线程先拿到锁(有一个特殊情况,就是队列里的线程在unpark到获取锁的过程中有新的还未加入到队列中的线程获取到锁,不过这种情况发生的概率很小,基本不用考虑),而NonfairSync不保证
下面我们看看ReentrantLock是如何实现上锁的,这是lock函数:
public void lock() { sync.lock(); }
sync的lock函数为抽象的,由子类实现,这里只给出FairSync的实现
final void lock() { acquire(1); }
acquire就是AQS提供的接口
public final void acquire(int arg) { if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) selfInterrupt(); }
这里涉及到4个新方法,让我们慢慢来分析:
首先,acquire有一个参数arg,是用于判断锁持有的次数,也就是重入的次数,当锁持有者需要释放锁的时候,则要将锁的state减去arg的值。在上面可以看到,调用acquire时传入的参数为1。
1. tryAcquire:AQS里tryAcquire的实现如下
protected boolean tryAcquire(int arg) { throw new UnsupportedOperationException(); }
明显是不对的,只抛出了一个异常,所以应该是子类覆盖了,那么看看FairSync的实现
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; }
getState()是从ReentrantLock获取的状态,表示锁当前是否被持有,为0时表示没有线程持有锁。此时当前线程会去争取锁的持有权。
首先判断队列中是否有排在当前线程之前的线程,有的话放弃争抢锁。hasQueuedPredecessors是AQS里的方法:
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()); }
tail记录队列里的尾节点,head就是头结点。这个函数就是判断当前线程是否是下一个可持锁的线程
然后,当当前线程满足条件时,就通过CAS设置c的值,同时通过AbstractOwnableSynchronizer类提供的setExclusiveOwnerThread接口将当前线程锁住,用于防止volatile字段被其他线程修改(这里是看注释后的个人理解)
如果c != 0,同时锁持有者为当前线程,那么这个请求就是重入请求了,将c+=acquire。如果上诉两个条件都不满足,返回false。
2. addWaiter:
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; }
根据注释可以知道,首先快速插入队列,其实就是判断队列是否为空,否则就要通过enq(node)将当前节点添加至队列中,为什么这样会慢些?我们看看enq函数
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; } } } }
这个方法可以好好看看:用for无限循环,考虑下多线程同时加入队列,通过CAS保证设置当前节点到尾节点。这里可能队列里的线程已经执行完任务释放锁了,所以还需要重新判断队列是否为空,因此其执行效率当然会有影响。
否则,设置前置节点,加入队列。用图例来帮助理解:
由于本身没有锁,可以有多个线程进来,如果有多个线程并发进入这个if判定区域,可能就会同时存在多个这样的数据结构,在各自形成数据结构后,多个线程都会去做compareAndSetHead(h)的动作,也就是尝试将这个临时h节点设置为head,
显然并发时只有一个线程会成功,因此成功的那个线程会执行tail = node的操作,整个AQS的链表就成为:
3. 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)) { setHead(node); p.next = null; // help GC failed = false; return interrupted; } if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) interrupted = true; } } finally { if (failed) cancelAcquire(node); } }
同样是for无限循环,判断当前节点之前是否没有线程节点了,如果是,就去争抢锁。用上面给出的子类tryAcquire函数,成功的话设置相关参数,这里也解释了释放指针,帮助垃圾回收(GC)。
如果争抢失败,判断是否需要阻塞当前线程
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; }
waitState参数有以下几个:
后面两个没怎么看,大致就是需要满足多个条件,PROPAGATE用于shared锁队列。
回到上面的方法:因为AQS支持中断等待,所以如果线程中断了争抢锁(CANCELLED),那么就不需要阻塞,直接返回。acquireQueued方法没有返回,而是设置一个interrupt参数为true而已,线程争抢锁失败的话继续休眠等待,而AQS里的doAcquireInterruptibly()发现争抢失败的话就直接throw new InterruptedException()。在ReentrantLock里需要调用ReentrantLock.lockInterruptibly()就会实现中断返回。否则AQS会尝试将当前线程状态设置成SIGNAL,失败就循环继续尝试
下面是parkAndCheckInterrupt
private final boolean parkAndCheckInterrupt() { LockSupport.park(this); return Thread.interrupted(); }
LockSupport.park通过Unsafe.park实现阻塞,它会设置一个AQS的blocker,让队列里的线程阻塞在一个地方,然后返回线程中断的判断
4. selfInterrupt
private static void selfInterrupt() { Thread.currentThread().interrupt(); }
acquireQueued会返回boolean值表示线程是否中断,如果未中断,就调用Thread.interrupt()中断线程
---------------------------------------------------------------------------------------------------------------------
以上是锁的实现原理,当tryAcquire()成功之后,线程获取锁,执行任务,执行完毕之后,会调用AQS的release方法:
public final boolean release(int arg) { if (tryRelease(arg)) { Node h = head; if (h != null && h.waitStatus != 0) unparkSuccessor(h); return true; } return false; }
tryRelease是子类Sync的实现:
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; }
其他就不多说了,比较简单明晰。就看看unparkSuccessor怎么做的
private void unparkSuccessor(Node node) { /* * If status is negative (i.e., possibly needing signal) try * to clear in anticipation of signalling. It is OK if this * fails or if status is changed by waiting thread. */ int ws = node.waitStatus; if (ws < 0) compareAndSetWaitStatus(node, ws, 0); /* * Thread to unpark is held in successor, which is normally * just the next node. But if cancelled or apparently null, * traverse backwards from tail to find the actual * non-cancelled successor. */ 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) LockSupport.unpark(s.thread); }
通过CAS修改节点的waitStatus,然后将后续节点去除,这里会去遍历后续节点,判断其是否状态为CANCELLED,将所有非CANCELLED的节点唤醒。
标签:
原文地址:http://www.cnblogs.com/pfan8/p/5010526.html