ReentrantLock原理学习

ReentrantLock 实际上是基于 AbstractQueuedSynchronizer实现的可重入锁。在阅读了美团技术博客的相关文章后,对 ReentrantLock 以及 AQS 有了更深入的理解，这里梳理自己的思路，记录自己的学习经验。

AQS 简要原理

核心为由 volatile 修饰的 int 变量 state 和 CLH 等待队列

state 变量用来维护状态，一般非共享锁用0表示空闲，1表示已被占有，若已被占有时还有线程来请求，则加入 CLH 等待队列。

CLH等待队列其实是一个双向链表，存储的元素为 AQS 的内部类 Node，其中一些方法和属性(摘自美团博客)：

lock() 的流程

ReentrantLock 中 lock() 方法的源码。

public void lock() {
    sync.acquire(1);
}

sync 是 ReentrantLock 中的内部类 Sync 的对象, 其又分为公平和非公平，代表着公平锁和非公平锁。

而 acquire() 方法继承自 AQS

public final void acquire(int arg) {
    if (!tryAcquire(arg) &&
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        selfInterrupt();
}

首先判断 tryAcquire(arg) 是否成功, tryAcquire() 方法由 Sync子类重写，看一下非公平锁的实现：

protected final boolean tryAcquire(int acquires) {
    return nonfairTryAcquire(acquires);
}

@ReservedStackAccess
final boolean nonfairTryAcquire(int acquires) {
    final Thread current = Thread.currentThread();
    int c = getState();
    if (c == 0) {
        // 若state为0，尝试CAS将state置为1
        if (compareAndSetState(0, acquires)) {
            //将占有锁的线程设为当前线程，AbstractOwnableSynchronizer中成员变量
            //private transient Thread exclusiveOwnerThread;
            setExclusiveOwnerThread(current);
            return true;
        }
    }
    else if (current == getExclusiveOwnerThread()) {
        //可重入，重入一次+1
        int nextc = c + acquires;
        if (nextc < 0) // overflow
            throw new Error("Maximum lock count exceeded");
        setState(nextc);
        return true;
    }
    return false;
}

若 tryAcquire() 成功, 则成功获取锁，否则执行 acquireQueued(addWaiter(Node.EXCLUSIVE), arg)方法。

先看 addWaiter() 方法，返回一个 Node。

private Node addWaiter(Node mode) {
    //构造方法中会把Node中thread置为当前线程
    Node node = new Node(mode);

    for (;;) {
        Node oldTail = tail;
        if (oldTail != null) {
            node.setPrevRelaxed(oldTail);
            //将新节点置为tail.
            if (compareAndSetTail(oldTail, node)) {
                oldTail.next = node;
                return node;
            }
        } else {
            //队列还未初始化，则初始化队列
            initializeSyncQueue();
        }
    }
}

private final void initializeSyncQueue() {
    Node h;
    if (HEAD.compareAndSet(this, null, (h = new Node())))
        tail = h;
 	//创建一个空节点，作为虚拟头部
}

再看 acquireQueued(final Node node,int arg)方法

    final boolean acquireQueued(final Node node, int arg) {
        boolean interrupted = false;
        try {
            for (;;) {
                final Node p = node.predecessor();
                if (p == head && tryAcquire(arg)) {
                    setHead(node);
                    p.next = null; // help GC
                    return interrupted;
                }
                if (shouldParkAfterFailedAcquire(p, node))
                    interrupted |= parkAndCheckInterrupt();
            }
        } catch (Throwable t) {
            cancelAcquire(node);
            if (interrupted)
                selfInterrupt();
            throw t;
        }
    }

不断自旋尝试获取锁，为避免长期自旋，还会在循环中检测是否需要阻塞。

若当前节点前驱节点为 head (空节点作虚拟头部)，说明当前节点已为队列中最前的节点，执行 tryAcquire() 方法，成功则将当前节点设为头节点。之后会判断当前节点是否需要阻塞，源码如下，自旋重复上述操作。

检查是否需要阻塞。若前一个节点状态已经为 SIGNAL 返回 true, 若前一个节点被取消，则不断往前找到第一个非取消的节点与当前 node 互连，否则尝试通过 CAS 将前一节点等待状态设为 SIGNAL。

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.
         */
        pred.compareAndSetWaitStatus(ws, Node.SIGNAL);
    }
    return false;
}

若抛出异常则执行 cancelAcquire() 方法。

private void cancelAcquire(Node node) {
    // Ignore if node doesn't exist
    if (node == null)
        return;

    node.thread = null;

    // Skip cancelled predecessors
    Node pred = node.prev;
    while (pred.waitStatus > 0)
        node.prev = pred = pred.prev;

    // predNext is the apparent node to unsplice. CASes below will
    // fail if not, in which case, we lost race vs another cancel
    // or signal, so no further action is necessary, although with
    // a possibility that a cancelled node may transiently remain
    // reachable.
    Node predNext = pred.next;

    // Can use unconditional write instead of CAS here.
    // After this atomic step, other Nodes can skip past us.
    // Before, we are free of interference from other threads.
    node.waitStatus = Node.CANCELLED;

    // If we are the tail, remove ourselves.
    if (node == tail && compareAndSetTail(node, pred)) {
        pred.compareAndSetNext(predNext, null);
    } else {
        // If successor needs signal, try to set pred's next-link
        // so it will get one. Otherwise wake it up to propagate.
        int ws;
        if (pred != head &&
            ((ws = pred.waitStatus) == Node.SIGNAL ||
             (ws <= 0 && pred.compareAndSetWaitStatus(ws, Node.SIGNAL))) &&
            pred.thread != null) {
            Node next = node.next;
            if (next != null && next.waitStatus <= 0)
                pred.compareAndSetNext(predNext, next);
        } else {
            unparkSuccessor(node);
        }

        node.next = node; // help GC
    }
}

回到 acquire() 方法通过返回值判断线程是否被中断，是则将线程中断。

unlock() 流程

public final boolean release(int arg) {
    if (tryRelease(arg)) {
        Node h = head;
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);
        return true;
    }
    return false;
}


// Sync类重写
protected final boolean tryRelease(int releases) {
    //考虑可重入
    int c = getState() - releases;
    if (Thread.currentThread() != getExclusiveOwnerThread())
        throw new IllegalMonitorStateException();
    boolean free = false;
    //为0则释放锁
    if (c == 0) {
        free = true;
        setExclusiveOwnerThread(null);
    }
    setState(c);
    return free;
}

先尝试 tryRelease() 方法，若可重入锁完全释放，判断后继节点是否需要释放，需要则释放 head 节点的后继节点（即解除后继节点维护的线程阻塞）。

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)
        node.compareAndSetWaitStatus(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 p = tail; p != node && p != null; p = p.prev)
            if (p.waitStatus <= 0)
                s = p;
    }
    if (s != null)
        LockSupport.unpark(s.thread);
}

tryLock(long timeout, TimeUnit unit)流程

public boolean tryLock(long timeout, TimeUnit unit)
        throws InterruptedException {
    return sync.tryAcquireNanos(1, unit.toNanos(timeout));
}

public final boolean tryAcquireNanos(int arg, long nanosTimeout)
        throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    return tryAcquire(arg) ||
        doAcquireNanos(arg, nanosTimeout);
}

private boolean doAcquireNanos(int arg, long nanosTimeout)
        throws InterruptedException {
    if (nanosTimeout <= 0L)
        return false;
    final long deadline = System.nanoTime() + nanosTimeout;
    final Node node = addWaiter(Node.EXCLUSIVE);
    try {
        for (;;) {
            final Node p = node.predecessor();
            if (p == head && tryAcquire(arg)) {
                setHead(node);
                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;
    }
}