AbstractQueuedSynchronizer 原理分析

简介

首先各种blocking lock 和 synchronizers (semaphores, events, etc)都是依赖于FIFO队列的。而AbstractQueuedSynchronizer 通过一个整型(int)的数值表示当前队列的状态，子类中可以通过getState/setStatye/compareAndSetState方法来改变队列状态。两种模式： 共享模式 shared mode 线程共享排他模式 exclusive mode 线程独占

数据结构

Queue中的节点 - Node

Node{ int waitState; Node prev; Node next; Thread thread;//该节点加入队列时的线程 Node nextWaiter;//等待Condition的节点 }

Node 中定义了Queue的运行模式shared/exclusive(default)

static final Node SHARED = new Node(); static final Node EXCLUSIVE = null;

waitStatus（节点状态）:

CANCELLED 1 由于超时、被打断导致cancel。该线程不会再被阻塞。SIGNAL -1 这个记号表示当前node之后的节点已经或即将被阻塞，需要再release或者cancel后唤醒一个或若干个后续节点，在该节点被release/cancel的时候，需要唤醒后面的节点。CONDITION -2表示该节点处于CONDITION队列中。PROPAGATE -3 当前场景下后续的acquireShared能够得以执行。0 表示当前节点在队列中，等待获取锁。

AbstractQueuedSynchronizer 持有Queue

{ Node head; Node tail; int state; }

AbstractQueuedSynchronizer 操作Queue

getState(); setState(); Node enq(final Node node)//insert node; Node addWaiter(Node mode)// add waiter - shared/exclusive node //唤醒后面节点 private void unparkSuccessor(Node node) { int ws = node.waitStatus; if (ws < 0) compareAndSetWaitStatus(node, ws, 0);//更新为0,表示等待锁 Node s = node.next; if (s == null || s.waitStatus > 0) { s = null; //从tail开始查找non-canceled node,然后唤醒他。 for (Node t = tail; t != null && t != node; t = t.prev) if (t.waitStatus <= 0) s = t; } if (s != null) LockSupport.unpark(s.thread); }

同步器实现

获取 public final void acquire(int arg) { if (!tryAcquire(arg) && //tryAcquire（）获取arg状态,依赖子类具体实现 acquireQueued(addWaiter(Node.EXCLUSIVE), arg))//获取失败，将该节点插入tail,再次尝试获取。 selfInterrupt(); } 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); } }

从上面的代码中可以看出，acquire会忽略Interruption，为此代码提供者还实现了可终止的方法来获取状态-抛异常。

public final void acquireInterruptibly(int arg) throws InterruptedException { if (Thread.interrupted()) throw new InterruptedException(); if (!tryAcquire(arg)) doAcquireInterruptibly(arg); } //doAcquireInterruptibly 处理方式同上面的处理方式是相似的，只不过在获取失败的时候，会抛异常 throw new InterruptedException(); 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; } 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) LockSupport.unpark(s.thread);//唤醒节点 }

ReentrantLock#FairSync 中的实现

lock final void lock() { acquire(1); } //AbstractQueuedSynchronizer#acquire() public final void acquire(int arg) { if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) selfInterrupt(); } // ReentrantLock#FairSync#tryAcquire protected final boolean tryAcquire(int acquires) { final Thread current = Thread.currentThread(); int c = getState();//获取当前队列的状态 if (c == 0) {//state == 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; } release protected final boolean tryRelease(int releases) { int c = getState() - releases;//正常情况下，getState() 和releases是相等的。 if (Thread.currentThread() != getExclusiveOwnerThread()) throw new IllegalMonitorStateException(); boolean free = false; if (c == 0) { free = true; setExclusiveOwnerThread(null);// } setState(c); return free; }

疑问点：

unparkSuccessor 中 查找non-canceled node 为什么需要从tail开始查找？