ReentrantLock源码解析

青旅半醒 2022-03-15 03:34 248阅读 0赞

前言：

ReentrantLock是java并发包中的同步组件，属于排他锁，其通过CAS的加锁方式相比较synchronized更加轻便，不会造成死锁，其核心实现是其两个内部类FairSync和NonfairSync，意为公平锁和非公平锁，ReentrantLock相关的加锁解锁等方法内部其实都是调用的这两个内部类的对应方法，公平锁和非公平锁都继承了AbstractQueuedSynchronizer，抽象队列同步器，简称AQS，AQS是用来构建锁和其他同步组件的基础框架，他通过一个int类型的成员变量表示同步状态，另外通过一个FIFO队列（Node形式的链表，类似于Linkedlist底层的链表结构）来维护资源获取线程的排队工作，获取锁失败的线程将阻塞等待，并排在队列尾部，公平锁的实现原理就是,优先选择等待时间最长的线程获取锁（即处在队列头部的线程），可以有效的避免线程饥饿，但是高并发情况下效率较低下，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;
    
        /**
         * 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();
    
            /**
             * 非公平模式尝试获取锁与下面公平模式这个获取锁的逻辑的唯一区别就是没有判断是否在等待队列头部，即所有线程均有机会获得锁
             * @see FairSync#tryAcquire(int) 
             * 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) {//剩余获得锁的次数为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 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))//通过CAS获取锁，可以发现这里并没有判断是否处在等待队列头部，因此所有线程都有机会获取到锁
                    setExclusiveOwnerThread(Thread.currentThread());//获取锁成功，设置独占模式的持有者为当前线程
                else
                    acquire(1);
            }
    
            protected final boolean tryAcquire(int acquires) {
                return nonfairTryAcquire(acquires);//非公平模式尝试获取锁
            }
        }
    
        /**
         * 公平锁
         *  等待时间最长的线程（线程等待队列中越靠前的线程代表着等待时间越长）将优先获得锁，避免了线程饥饿，但是性能上比非公平锁低
         * Sync object for fair locks
         */
        static final class FairSync extends Sync {
            private static final long serialVersionUID = -3000897897090466540L;
    
            final void lock() {
                acquire(1);
            }
    
            /**
             * 公平锁尝试获取锁的逻辑，只有在锁未被占用或者无等待队列或者当前线程处在等待队列第一位时可以获取成功
             * Fair version of tryAcquire.  Don't grant access unless
             * recursive call or no waiters or is first.
             */
            protected final boolean tryAcquire(int acquires) {
                final Thread current = Thread.currentThread();
                int c = getState();//获取当前锁被被获取的次数
                if (c == 0) {//如果是0说明锁未被占用，尝试获取
                    if (!hasQueuedPredecessors() &&
                        compareAndSetState(0, acquires)) {//只有当前线程排在线程等待对列中第一位时，尝试CAS获取锁
                        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;//锁已被占用，且持有者非当前线程，获取锁失败
            }
        }
    
        /**
         * 默认的构造方法，将实例化一个不公平的队列同步器
         * 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();
        }
    
    
        public void lock() {
            sync.lock();
        }
    
        /**
         * 可中断的获取锁，与lock不同的是，该方法会响应中断，可以避免发生死锁
         * 
         */
        public void lockInterruptibly() throws InterruptedException {
            sync.acquireInterruptibly(1);
        }
    
        /**
         * 非阻塞尝试获取锁，可以发现其调用了nonfairTryAcquire方法，该方法是非公平的，所以即使当前锁是公平锁，调用此方法也会打破公平优先去获取锁
         *   
         */
        public boolean tryLock() {
            return sync.nonfairTryAcquire(1);
        }
    
        /**
         * 在固定时间内尝试加锁，该方法将响应中断
         * @see Thread#interrupt() 
         */
        public boolean tryLock(long timeout, TimeUnit unit)
                throws InterruptedException {
            return sync.tryAcquireNanos(1, unit.toNanos(timeout));
        }
    
    
        public void unlock() {
            sync.release(1);
        }
    }

AbstractQueuedSynchronizer(省略部分代码，仅保留ReentrantLock相关的代码)

public abstract class AbstractQueuedSynchronizer
        extends AbstractOwnableSynchronizer
        implements java.io.Serializable {
    
        private static final long serialVersionUID = 7373984972572414691L;
    
        /**
         * Creates a new {@code AbstractQueuedSynchronizer} instance
         * with initial synchronization state of zero.
         */
        protected AbstractQueuedSynchronizer() { }
    
        static final class Node {
            /** Marker to indicate a node is waiting in shared mode */
            static final Node SHARED = new Node();
            /** Marker to indicate a node is waiting in exclusive mode */
            static final Node EXCLUSIVE = null;
    
            /** waitStatus value to indicate thread has cancelled */
            static final int CANCELLED =  1;
            /** waitStatus value to indicate successor's thread needs unparking */
            static final int SIGNAL    = -1;
            /** waitStatus value to indicate thread is waiting on condition */
            static final int CONDITION = -2;
            /**
             * waitStatus value to indicate the next acquireShared should
             * unconditionally propagate
             */
            static final int PROPAGATE = -3;
    
    
            volatile int waitStatus;
    
            volatile Node prev;
    
            volatile Node next;
    
            volatile Thread thread;
    
            Node nextWaiter;
    
            /**
             * Returns true if node is waiting in shared mode.
             */
            final boolean isShared() {
                return nextWaiter == SHARED;
            }
    
            /**
             * Returns previous node, or throws NullPointerException if null.
             * Use when predecessor cannot be null.  The null check could
             * be elided, but is present to help the VM.
             *
             * @return the predecessor of this node
             */
            final Node predecessor() throws NullPointerException {
                Node p = prev;
                if (p == null)
                    throw new NullPointerException();
                else
                    return p;
            }
    
            Node() {    // Used to establish initial head or SHARED marker
            }
    
            Node(Thread thread, Node mode) {     // Used by addWaiter
                this.nextWaiter = mode;
                this.thread = thread;
            }
    
            Node(Thread thread, int waitStatus) { // Used by Condition
                this.waitStatus = waitStatus;
                this.thread = thread;
            }
        }
    
        /**
         * Head of the wait queue, lazily initialized.  Except for
         * initialization, it is modified only via method setHead.  Note:
         * If head exists, its waitStatus is guaranteed not to be
         * CANCELLED.
         */
        private transient volatile Node head;
    
        /**
         * Tail of the wait queue, lazily initialized.  Modified only via
         * method enq to add new wait node.
         */
        private transient volatile Node tail;
    
        /**
         *  当前锁被获取的次数
         * The synchronization state.
         */
        private volatile int state;
    
        /**
         * 为当前线程的给定模式创建一个入队的节点
         * 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 = new Node(Thread.currentThread(), mode);//创建一个新的节点
            // Try the fast path of enq; backup to full enq on failure
            Node pred = tail;
            if (pred != null) {//将该节点设置到末尾，追加到原本末位节点的尾部，这里使用了CAS比较
                node.prev = pred;
                if (compareAndSetTail(pred, node)) {
                    pred.next = node;
                    return node;
                }
            }
            enq(node);//CAS+自旋将当前节点加入到队列尾部
            return node;
        }
    
        /**
         *  用于已经入队的线程在排他不可中断模式下获取锁。同时也被用于给定条件（condition）的wait方法的获取锁
         *  该方法，处在等待队列中的方法将不断地尝试去获取锁
         *  该方法是忽略中断的
         * 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();//获取当前节点的前驱节点
                    if (p == head && tryAcquire(arg)) {//如果前驱节点是头节点，说明当前线程是队列中第一个等待获取锁的节点，尝试获取锁
                        setHead(node);//获取成功，将当前节点设置为头节点
                        p.next = null; // help GC  原先的头节点next置空，便于gc回收
                        failed = false;
                        return interrupted;
                    }
                    //如果前驱节点不是头节点（说明当前线程不是队列中第一个等待获取锁的节点），或者当前线程尝试获取锁失败，则尝试去挂起线程并且返回是否中断标识
                    if (shouldParkAfterFailedAcquire(p, node) &&
                        parkAndCheckInterrupt())
                        interrupted = true;
                }
            } finally {
                if (failed)//当前线程由于某种原因未获取到锁，却要结束了，则取消等待获取锁，因为结束意味着当前线程已不再等待，则等待队列中不再须要保留当前线程的节点
                    cancelAcquire(node);
            }
        }
    
        /**
         * 独占式获取同步状态，实现该方法须要查询当前同步状态是否符合预期，然后再进行CAS设置同步状态
         * 
         */
        protected boolean tryAcquire(int arg) {
            throw new UnsupportedOperationException();
        }
    
        /**
         * 独占释放同步状态，等待获取同步状态的线程将有机会获取同步状态
         * 
         */
        protected boolean tryRelease(int arg) {
            throw new UnsupportedOperationException();
        }
    
        /**
         * 共享式获取同步状态，返回大于等于0的值，表示获取成功，反之获取失败
         * 
         */
        protected int tryAcquireShared(int arg) {
            throw new UnsupportedOperationException();
        }
    
        /**
         * 共享式释放同步状态
         * 
         */
        protected boolean tryReleaseShared(int arg) {
            throw new UnsupportedOperationException();
        }
    
        /**
         * 当前同步器是否在独占模式下被线程占用，一般该方法表示是否被当前线程所独占
         * 
         */
        protected boolean isHeldExclusively() {
            throw new UnsupportedOperationException();
        }
    
        /**
         * 尝试获取锁，假如获取失败，则会已阻塞模式加入等待队列中，并不断尝试去获取，该方法将阻塞当前线程
         * 该方法在尝试获取锁的过程中是忽略中断的
         * 
         */
        public final void acquire(int arg) {
            if (!tryAcquire(arg) &&
                acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
                selfInterrupt();//设置线程的中断标识，因为当前线程是忽略中断的
        }
    
    
        /**
         * 释放锁
         * 
         */
        public final boolean release(int arg) {
            if (tryRelease(arg)) {//尝试释放锁，由子类实现
                Node h = head;
                if (h != null && h.waitStatus != 0)//唤醒头节点的后继节点
                    unparkSuccessor(h);
                return true;
            }
            return false;
        }
    }

public interface Condition {
         //使当前线程加入 await() 等待队列中，并释放当锁，当其他线程调用signal()会重新请求锁。与Object.wait()类似。
        void await() throws InterruptedException;
    
        //调用该方法的前提是，当前线程已经成功获得与该条件对象绑定的重入锁，否则调用该方法时会抛出IllegalMonitorStateException。
        //调用该方法后，结束等待的唯一方法是其它线程调用该条件对象的signal()或signalALL()方法。等待过程中如果当前线程被中断，该方法仍然会继续等待，同时保留该线程的中断状态。 
        void awaitUninterruptibly();
    
        // 调用该方法的前提是，当前线程已经成功获得与该条件对象绑定的重入锁，否则调用该方法时会抛出IllegalMonitorStateException。
        //nanosTimeout指定该方法等待信号的的最大时间（单位为纳秒）。若指定时间内收到signal()或signalALL()则返回nanosTimeout减去已经等待的时间；
        //若指定时间内有其它线程中断该线程，则抛出InterruptedException并清除当前线程的打断状态；若指定时间内未收到通知，则返回0或负数。 
        long awaitNanos(long nanosTimeout) throws InterruptedException;
    
        //与await()基本一致，唯一不同点在于，指定时间之内没有收到signal()或signalALL()信号或者线程中断时该方法会返回false;其它情况返回true。
        boolean await(long time, TimeUnit unit) throws InterruptedException;
    
       //适用条件与行为与awaitNanos(long nanosTimeout)完全一样，唯一不同点在于它不是等待指定时间，而是等待由参数指定的某一时刻。
        boolean awaitUntil(Date deadline) throws InterruptedException;
        
        //唤醒一个在 await()等待队列中的线程。与Object.notify()相似
        void signal();
    
       //唤醒 await()等待队列中所有的线程。与object.notifyAll()相似
        void signalAll();
    }

了解了ReentrantLock加锁解锁流程后，其他相关方法就很好理解了，另外与其相对的共享锁ReentrantReadWriteLock的底层设计同ReentrantLock类似，只是加入了共享模式的业务逻辑。