浅谈AbstractQueuedSynchronizer及ReentrantLock

本文主要是介绍浅谈AbstractQueuedSynchronizer及ReentrantLock，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

1、AbstractQueuedSynchronizer 简要描述

是java.util.concurrent.locks包下的抽象类

jdk原始注释：

Provides a framework for implementing blocking locks and related synchronizers (semaphores, events, etc) that rely on first-in-first-out (FIFO) wait queues.This class is designed to be a useful basis for most kinds of synchronizers that rely on a single atomic {@code int} value to represent state

该抽象类提供了一个实现依赖于FIFO等待队列以及一个原子的阻塞锁和相关同步器（semaphores，events等等）的框架。该类设计上依赖于一个原子的int值来表示同步状态，为众多同步器提供了一个有效的基础。

AbstractQueuedSynchronizer中有一个内部类Node，队列有一个个Node组成。Node包含以下属性

waitStatus，当前节点的等待状态；

prev，前一个节点，

next，下一个节点，

thread， 处于当前节点的线程，

head，队列的头节点、

tail，队列的尾部节点，

state，同步的状态（锁的状态）；

每个同步器包含父类AbstractOwnableSynchronizer的exclusiveOwnerThread，记录获得该锁使用权的线程。

2、用AbstractQueuedSynchronizer实现锁的基本思路：多个线程进程竞争锁（CAS的方式），竞争成功，则修改同步器中锁的状态（如由0为1），exclusiveOwnerThread=当前线程，当前线程继续执行任务；竞争失败的线程则开始自旋竞争进入队列尾部，并在检查和更新状态后挂起（如果前驱节点pred状态是需要唤醒的SIGNAL）。当当前线程释放锁，所得状态恢复为0，exclusiveOwnerThread=null, 公平锁则是从队列中取出离头部最近的挂起线程，unpark后执行该线程任务，而非公平锁，则是队列中取出离头部最近的挂起线程与新来的线程进行竞争，成功后执行，否则入队列。

 

3、代码分析

 acquire方法，负责获取锁

    public final void acquire(int arg) {if (!tryAcquire(arg) &&     //tryAcquire 描述了竞争的行为，由子类实现，同步非同步的精髓acquireQueued(addWaiter(Node.EXCLUSIVE), arg))//addWaiter 加入队列 selfInterrupt();                              //acquireQueued负责获取队列中的线程}

addWaiter 将当前线程插入队列尾部

1. 如果队列不为空，待插入节点的pred指向最后一个节点，并通过CAS的方法竞争插入队列尾部。竞争成功，则待插入节点成为tail，原来的tail节点的next指向待插入节点；
2. 如果队列为空或竞争失败，调用enq方法。

    private Node addWaiter(Node mode) {Node node = new Node(Thread.currentThread(), mode);// Try the fast path of enq; backup to full enq on failureNode pred = tail;if (pred != null) {         //队列不为空，竞争插入队列node.prev = pred;if (compareAndSetTail(pred, node)) {pred.next = node;return node;}}enq(node);                 //队列为空，或竞争失败，调用enqreturn node;}

***********************************************************************************************************************************************

enq方法，通过自旋的方法，循环尝试插入队尾 且保持head->…->tail的队列形式，使得每个线程都有pred节点，（pred节点的状态是决定当前线程是否会被唤醒的重要标志）

1.当队列为空，新建一个节点作为队列头部，然后开始自旋竞争插入队尾（死循环下compareAndSetTail）直至成功插入；

               Head   TAIL

               Null ->  thread1;

2.队列不为空，开始自旋竞争插入队尾（死循环下compareAndSetTail）直至成功插入；

    /*** Inserts node into queue, initializing if necessary. See picture above.* @param node the node to insert* @return node's predecessor*/private Node enq(final Node node) {for (;;) {                                 //自旋Node t = tail;if (t == null) { // Must initializeif (compareAndSetHead(new Node()))  //队列为空，新增节点当 headtail = head;} else {node.prev = t;if (compareAndSetTail(t, node)) {   //竞争插入队尾t.next = node;return t;}}}}

***********************************************************************************************************************************************

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)) {   //前一个节点为head，且竞争成功，获得锁setHead(node);p.next = null; // help GC  failed = false;return interrupted;}if (shouldParkAfterFailedAcquire(p, node) &&  //首先检查更新pred节点的状态parkAndCheckInterrupt())                  //挂起操作interrupted = true;}} finally {if (failed)cancelAcquire(node);                         }}

也是自旋操作。

1.如果该节点的前驱节点为队列头部，且获取锁成功，该节点设置为head，原head.next为空，出队列。返回false。

2.执行shouldParkAfterFailedAcquire方法（直译应该是获取锁失败将线程挂起，但事实上并不是方法挂起的，这个方法只是检查并更新前驱节点的状态）共有三种情况：

a.当前驱节点状态为Node.SIGNAL时，返回true，即需要挂起。执行parkAndCheckInterrupt()，挂起该线程;

b.执行shouldParkAfterFailedAcquire方法，若前驱节点状态值>0，将该节点忽略node.prev = pred = pred.prev; pred.next = node; 返回false,继续自旋执行shouldParkAfterFailedAcquire，a情况，执行parkAndCheckInterrupt()，挂起该线程；

c.若前驱节点为0，则用cas置为Node.SIGNAL，return false, 继续自旋执行shouldParkAfterFailedAcquire，a情况，执行parkAndCheckInterrupt()，挂起该线程；

    /*** Checks and updates status for a node that failed to acquire.* Returns true if thread should block. This is the main signal* control in all acquire loops.  Requires that pred == node.prev.** @param pred node's predecessor holding status* @param node the node* @return {@code true} if thread should block*/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;}

    /*** Convenience method to park and then check if interrupted** @return {@code true} if interrupted*/private final boolean parkAndCheckInterrupt() {LockSupport.park(this);                        //LockSupport.park挂起当前线程return Thread.interrupted();}

***********************************************************************************************************************************************

release  解锁

tryRelease释放锁的行为（锁状态减1，为0时将exclusiveOwnerThread=null）return true, head !=null，且状态为下一个节点需要唤醒，将在队列中查找首个状态<=0的节点（未被取消，即shouldParkAfterFailedAcquire未来得及更新状态及从队列中去掉的），unpark该节点中的线程。由于该线程之前是在acquireQueued中受到阻塞，会继续自旋，同样状态>0的无效节点也会进行自旋，然后cancel掉。使得刚unpark的pred为head，进而tryAcquire（竞争获取锁），失败后，检查并更新状态继续等待。成功了则成为head。并执行线程剩下的任务。

    /*** Releases in exclusive mode.  Implemented by unblocking one or* more threads if {@link #tryRelease} returns true.* This method can be used to implement method {@link Lock#unlock}.** @param arg the release argument.  This value is conveyed to*        {@link #tryRelease} but is otherwise uninterpreted and*        can represent anything you like.* @return the value returned from {@link #tryRelease}*/public final boolean release(int arg) {if (tryRelease(arg)) {Node h = head;if (h != null && h.waitStatus != 0)unparkSuccessor(h);return true;}return false;}

    /*** Wakes up node's successor, if one exists.** @param node the node*/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);}

这篇关于浅谈AbstractQueuedSynchronizer及ReentrantLock的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！

---