ReentrantLock 源码分析

ReentrantLock 怎么使用

ReentrantLock 通常作为类的成员变量，用于加锁和释放锁，下面举个例子

class X {
    private final ReentrantLock lock = new ReentrantLock();
    // ...

    public void m() {
        lock.lock();  // block until condition holds
        try {
            // ... method body
        } finally {
            lock.unlock();
        }
    }
}

ReentrantLock 一般作为成员变量使用，通过 lock 方法获取锁，如果没有获取到锁会阻塞，获取到锁之后，执行业务逻辑，在 finally 代码中释放锁，防止死锁。

ReentrantLock 构造方法

我们首先看下 ReentrantLock 类的构造方法

public class ReentrantLock implements Lock, java.io.Serializable {
    private static final long serialVersionUID = 7373984872572414699L;
    /** Synchronizer providing all implementation mechanics */
    private final Sync sync;

    // 省略部分代码

    /**
     * Creates an instance of {@code ReentrantLock}.
     * This is equivalent to using {@code ReentrantLock(false)}.
     */
    public ReentrantLock() {
        sync = new NonfairSync();
    }

    /**
     * Creates an instance of {@code ReentrantLock} with the
     * given fairness policy.
     *
     * @param fair {@code true} if this lock should use a fair ordering policy
     */
    public ReentrantLock(boolean fair) {
        sync = fair ? new FairSync() : new NonfairSync();
    }
  	
    // 省略部分代码
}

ReentrantLock 提供了两个构造方法，默认的构造方法内部实现的是一个非公平锁 NonfairSync，也可以通过参数指定要创建的是公平锁 FairSync 还是非公平锁 NonfairSync。

静态内部类 Sync

我们先看下静态内部类 Sync 的结构

public class ReentrantLock implements Lock, java.io.Serializable {
    private static final long serialVersionUID = 7373984872572414699L;
    /** Synchronizer providing all implementation mechanics */
    private final Sync sync;

    /**
     * Base of synchronization control for this lock. Subclassed
     * into fair and nonfair versions below. Uses AQS state to
     * represent the number of holds on the lock.
     */
    abstract static class Sync extends AbstractQueuedSynchronizer {
        private static final long serialVersionUID = -5179523762034025860L;

        /**
         * Performs {@link Lock#lock}. The main reason for subclassing
         * is to allow fast path for nonfair version.
         */
        abstract void lock();

        /**
         * 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;
        }

        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;
        }

        protected final boolean isHeldExclusively() {
            // While we must in general read state before owner,
            // we don't need to do so to check if current thread is owner
            return getExclusiveOwnerThread() == Thread.currentThread();
        }

        final ConditionObject newCondition() {
            return new ConditionObject();
        }

        // Methods relayed from outer class

        final Thread getOwner() {
            return getState() == 0 ? null : getExclusiveOwnerThread();
        }

        final int getHoldCount() {
            return isHeldExclusively() ? getState() : 0;
        }

        final boolean isLocked() {
            return getState() != 0;
        }

        /**
         * Reconstitutes the instance from a stream (that is, deserializes it).
         */
        private void readObject(java.io.ObjectInputStream s)
            throws java.io.IOException, ClassNotFoundException {
            s.defaultReadObject();
            setState(0); // reset to unlocked state
        }
    }

    // 省略部分代码
}

Sync 是一个抽象类，默认提供了两个实现，分别是我们上面提到的公平锁 FairSync 和非公平锁 NonfairSync，公平锁和非公平锁获取锁的方式不同，公平锁会先判断等待队列中是否有其他线程在排队，如果有，则加入等待队列的尾部，如果没有则通过 CAS 去尝试获取锁，如果获取不到锁则加入等待队列的尾部。而非公平锁获取锁的时候直接通过 CAS 尝试获取锁，不管等待队列中是否有其他线程在排队。

Sync 继承了 AbstractQueuedSynchronizer ，AbstractQueuedSynchronizer就是我们常说的 AQS 类。

AQS (AbstractQueuedSynchronizer)

内部就是一个 int 状态字段 state 和一个FIFO等待队列（双向链表），state 等于0 表示没有线程持有该锁，state 大于0表示有线程持有该锁。

ReentrantLock#lock 非公平锁的实现

下面我们通过 ReentrantLock#lock 方法一步一步的跟进代码

public class ReentrantLock implements Lock, java.io.Serializable {
    private static final long serialVersionUID = 7373984872572414699L;
    /** Synchronizer providing all implementation mechanics */
    private final Sync sync;

    // 省略部分代码
  
    /**
     * Acquires the lock.
     *
     * <p>Acquires the lock if it is not held by another thread and returns
     * immediately, setting the lock hold count to one.
     *
     * <p>If the current thread already holds the lock then the hold
     * count is incremented by one and the method returns immediately.
     *
     * <p>If the lock is held by another thread then the
     * current thread becomes disabled for thread scheduling
     * purposes and lies dormant until the lock has been acquired,
     * at which time the lock hold count is set to one.
     */
    public void lock() {
        sync.lock();
    }

    // 省略部分代码
}

我们可以看到 lock 方法内部调用了 Sync 一个子类的 lock方法，我们先看下非公平锁 NonfairSync 的实现

public class ReentrantLock implements Lock, java.io.Serializable {
    private static final long serialVersionUID = 7373984872572414699L;
    /** Synchronizer providing all implementation mechanics */
    private final Sync sync;

    // 省略部分代码

    /**
     * Sync object for non-fair locks
     */
    static final class NonfairSync extends Sync {
        private static final long serialVersionUID = 7316153563782823691L;

        /**
         * Performs lock.  Try immediate barge, backing up to normal
         * acquire on failure.
         */
        final void lock() {
            if (compareAndSetState(0, 1))
                setExclusiveOwnerThread(Thread.currentThread());
            else
                acquire(1);
        }

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

    // 省略部分代码
  
    public void lock() {
        sync.lock();
    }

    // 省略部分代码
} 

非公平锁首先通过 compareAndSetState 方法尝试将 state 的值从0改为1，如果修改成功，则将当前线程设置为持有锁的线程。

final void lock() {
  if (compareAndSetState(0, 1))
    setExclusiveOwnerThread(Thread.currentThread());
  else
    acquire(1);
}

否在执行父类 AQS 的 acquire 方法，我们继续跟进到 AbstractQueuedSynchronizer 类中的 acquire 方法

    /**
     * Acquires in exclusive mode, ignoring interrupts.  Implemented
     * by invoking at least once {@link #tryAcquire},
     * returning on success.  Otherwise the thread is queued, possibly
     * repeatedly blocking and unblocking, invoking {@link
     * #tryAcquire} until success.  This method can be used
     * to implement method {@link Lock#lock}.
     *
     * @param arg the acquire argument.  This value is conveyed to
     *        {@link #tryAcquire} but is otherwise uninterpreted and
     *        can represent anything you like.
     */
    public final void acquire(int arg) {
        if (!tryAcquire(arg) &&
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }

首先调用了 tryAcquire 方法，可以看到在 AbstractQueuedSynchronizer 类中的tryAcquire 方法是一个 protected 修饰的方法，并且内部直接抛出了异常，那么很明显这个方法需要子类去实现（模版方法模式）

    /**
     * Attempts to acquire in exclusive mode. This method should query
     * if the state of the object permits it to be acquired in the
     * exclusive mode, and if so to acquire it.
     *
     * <p>This method is always invoked by the thread performing
     * acquire.  If this method reports failure, the acquire method
     * may queue the thread, if it is not already queued, until it is
     * signalled by a release from some other thread. This can be used
     * to implement method {@link Lock#tryLock()}.
     *
     * <p>The default
     * implementation throws {@link UnsupportedOperationException}.
     *
     * @param arg the acquire argument. This value is always the one
     *        passed to an acquire method, or is the value saved on entry
     *        to a condition wait.  The value is otherwise uninterpreted
     *        and can represent anything you like.
     * @return {@code true} if successful. Upon success, this object has
     *         been acquired.
     * @throws IllegalMonitorStateException if acquiring would place this
     *         synchronizer in an illegal state. This exception must be
     *         thrown in a consistent fashion for synchronization to work
     *         correctly.
     * @throws UnsupportedOperationException if exclusive mode is not supported
     */
    protected boolean tryAcquire(int arg) {
        throw new UnsupportedOperationException();
    }

那么我们去 NonfairSync 类中看到 tryAcquire 方法调用了父类 Sync 的 nonfairTryAcquire 方法

/**
* 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)) { // 通过 CAS 去抢锁
      // 抢到锁了，抢持有锁的线程 exclusiveOwnerThread 设置为当前线程
      setExclusiveOwnerThread(current);
      return true;
    }
  }
  //  当前已经有线程获取到了锁，判断持有锁到线程是不是自己
  else if (current == getExclusiveOwnerThread()) {
    // 持有锁的线程是自己（可重入锁）
    int nextc = c + acquires;
    if (nextc < 0) // overflow
      throw new Error("Maximum lock count exceeded");
    // 将 state 值+1
    setState(nextc);
    return true;
  }
  // 没有获取到锁，返回 false
  return false;
}

至此，AQS 类中的 tryAcquire 实现已经分析完了，如果没有获取到锁，会继续执行 if 中的条件判断，我们继续分析 AQS 类 addWaiter 方法

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

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 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 指向前面node
            node.prev = pred;
            if (compareAndSetTail(pred, node)) { // 通过 CAS 将 tail 指向新加入的 node
                pred.next = node; // 将前面node.next 指向新node
                return node;
            }
        }
      	// 尾指针为 null 的情况
        enq(node);
        return node;
    }

AQS 类中 compareAndSetTail 方法

    /**
     * CAS tail field. Used only by enq.
     */
    private final boolean compareAndSetTail(Node expect, Node update) {
        return unsafe.compareAndSwapObject(this, tailOffset, expect, update);
    }

AQS 类中 enq 方法

    /**
     * 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 (;;) { // for 循环，通过 CAS 将新node加入到链表尾部
            Node t = tail;
            if (t == null) { // Must initialize
                if (compareAndSetHead(new Node())) // 通过 CAS 初始化 head 结点
                    tail = head; // 将 tail 指向 head
            } else {
              	// 以下逻辑与 addWaiter 方法上面实现一致
                node.prev = t;
                if (compareAndSetTail(t, node)) {
                    t.next = node;
                    return t;
                }
            }
        }
    }

至此，没有获取到锁的线程已经成功加入到等待队列的尾部了，AQS 类 addWaiter 方法会返回新加入到等待队列中的 node

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

我们继续分析 acquireQueued 方法

    /**
     * Acquires in exclusive uninterruptible mode for thread already in
     * queue. Used by condition wait methods as well as acquire.
     *
     * @param node the node
     * @param arg the acquire argument
     * @return {@code true} if interrupted while waiting
     */
    final boolean acquireQueued(final Node node, int arg) {
        boolean failed = true;
        try {
            boolean interrupted = false;
            for (;;) { // 自旋
                final Node p = node.predecessor(); // 新加入node的前一个结点
              	// p == head 说明新加入的node是链表中的第一个数据结点，尝试获取锁
                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 循环自旋，获取当前结点的前一个结点，如果前一个结点是头结点，然后尝试获取锁，如果获取锁成功。更新队列的头结点为当前结点，之前 head 指向一个空的 Node 结点(new Node())，这里将当前结点 node 的thread 和 prev 设置为 null，将 head 指向当前结点。

    /**
     * Sets head of queue to be node, thus dequeuing. Called only by
     * acquire methods.  Also nulls out unused fields for sake of GC
     * and to suppress unnecessary signals and traversals.
     *
     * @param node the node
     */
    private void setHead(Node node) {
        head = node; // 将当前结点设置为头结点
        node.thread = null;
        node.prev = null;
    }

如果新加入等待队列的结点不是第一个结点，或者是第一个结点但是尝试获取锁失败了，会执行shouldParkAfterFailedAcquire 方法，这个方法用于判断当前线程在尝试获取锁失败后是否需要 park（挂起）。

    /**
     * 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;
    }

如果当前线程需要挂起，则执行 parkAndCheckInterrupt 方法

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

ReentrantLock#unlock 非公平锁的实现

    /**
     * Attempts to release this lock.
     *
     * <p>If the current thread is the holder of this lock then the hold
     * count is decremented.  If the hold count is now zero then the lock
     * is released.  If the current thread is not the holder of this
     * lock then {@link IllegalMonitorStateException} is thrown.
     *
     * @throws IllegalMonitorStateException if the current thread does not
     *         hold this lock
     */
    public void unlock() {
        sync.release(1);
    }

调用了 AQS 的 release 方法

    /**
     * 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;
    }

调用 AQS 的 tryRelease 方法尝试释放锁，同样的 AQS 中的 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) {
              	// 如果state 的值为0，表示锁释放，对于可重入锁，加锁和解锁要成对出现
                free = true;
                // 将持有锁的线程设置为 null
                setExclusiveOwnerThread(null);
            }
          	// 更新 state 的值为0
            setState(c);
            return free;
        }

释放锁成功，获取等待队列头结点，唤醒等待队列中的其他线程

    /**
     * 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);
    }

本文主要介绍了 ReentrantLock 非公平锁的底层实现原理，公平锁的实现大家可自行查看源码。