ReentrantLock 源码解析(JDK1.8)

ReentrantLock简单使用demo如下:

Lock lock = new ReentrantLock(); 
lock.lock();
try {
    //业务逻辑  
} finally {
  lock.unlock();
}

注:获取的锁代码要放到try块之外,防止获得锁代码异常,抛出异常的同时,也会导致一次锁的释放。释放代码一定要放到finally块中。

** AQS ** 了解java中的锁,首先的了解AQS。 AQS(AbstractQueuedSynchronizer)队列同步器。是用来构建锁或者其它同步组件的基础框架,他实现了一个int成员变量标识同步状态(更改这个变量值来获取和释放锁),通过内置的FIFO双向队列来完成资源获取线程排队的工作。 AQS可以实现独占锁和共享锁,RenntrantLock实现的是独占锁,ReentrantReadWriteLock实现的是独占锁和共享锁,CountDownLatch实现的是共享锁。

ReentrantLock 类结构信息如下图:

Paste_Image.png

  • ReentrantLock 实现 Lock 和 Serializable 接口
  • RentrantLock 有三个内部类 Sync、NonfairSync 和 FairSync 类
  • Sync 继承 AbstractQueuedSynchronizer 抽象类
  • NonfairSync(非公平锁) 继承 Sync 抽象类
  • FairSync(公平锁) 继承 Sync 抽象类

** 公平锁和非公平锁 **

ReentrantLock 有两种实现方式,公平锁和非公平锁。

  • 公平锁:当前线程不立刻获得锁,而是先直进入等待队列中队尾进行排队获取锁。
  • 非公平锁:当前线程首先尝试获取一下锁(仅仅尝试一下),如果获取不到,则乖乖的进入到等待队列中去排队。

ReentrantLock实现公平锁和非公平锁代码如下:

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

** 获取非公平锁 **

       /**
        * 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);
       }
  1. 首先通过CAS更新AQS中的state变量来获得锁(第一次获得锁),如果获取成功则把当前线程设置为独占锁
  2. 如果是设置失败,进入到acquire方法
    public final void acquire(int arg) {
        if (!tryAcquire(arg) &&
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }
  1. 首先执行tryAcquire方法,尝试获得锁。
  2. 如果获取失败则进入addWaiter方法,构造同步节点,将该节点添加到同步队列尾部,并返回此节点,进入acquireQueued方法。
  3. acquireQueued方法,这个新节点死是循环的方式获取同步状态,如果获取不到则阻塞节点中的线程,阻塞后的节点等待前驱节点来唤醒。
        /**
         * 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;
        }

tryAcquire调用nonfairTryAcquire方法来第二次尝试获得锁

  1. 如果state变量为0,则进行CAS尝试更新state来获得锁,并把该线程设置成独占锁,并返回true。
  2. 如果state变量不为0,则判断当前线程是否为独占锁,如果是,则当前state+1(可重入锁),表示获取锁成功,更新state值,并返回true。这里更新state变量,不需要CAS更新,因为,当前线程已经获得锁。
    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) {
            node.prev = pred;
            if (compareAndSetTail(pred, node)) {
                pred.next = node;
                return node;
            }
        }
        enq(node);
        return node;
    }

将构造的同步节点加入到同步队列中

  1. 使用链表的方式把该Node节点添加到队列尾部,如果tail的前驱节点不为空(队列不为空),则进行CAS添加到队列尾部。
  2. 如果更新失败(存在并发竞争更新),则进入enq方法进行添加
private Node enq(final Node node) {
        for (;;) {
            Node t = tail;
            if (t == null) { // Must initialize
                if (compareAndSetHead(new Node()))
                    tail = head;
            } else {
                node.prev = t;
                if (compareAndSetTail(t, node)) {
                    t.next = node;
                    return t;
                }
            }
        }
    }

该方法使用CAS自旋的方式来保证向队列中添加Node(同步节点简写Node)

  1. 如果队列为空,则把当前Node设置成头节点
  2. 如果队列不为空,则向队列尾部添加Node
    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);
        }
    }

在acquireQueued方法中,当前线程在死循环中尝试获取同步状态,

  1. 如果当前节点的前驱节点头节点才能尝试获得锁,如果获得成功,则把当前线程设置成头结点,把之前的头结点从队列中移除,等待垃圾回收(没有对象引用)
  2. 如果获取锁失败则进入shouldParkAfterFailedAcquire方法中检测当前节点是否可以被安全的挂起(阻塞),如果可以安全挂起则进入parkAndCheckInterrupt方法,把当前线程挂起,并检查刚线程是否执行了interrupted方法。
    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 {
            /*
             * 尝试将当前节点的前驱节点的等待状态设为SIGNAL
             * 1/这为什么用CAS,现在已经入队成功了,前驱节点就是pred,除了node外应该没有别的线程在操作这个节点了,那为什么还要用CAS?而不直接赋值呢?
             * (解释:因为pred可以自己将自己的状态改为cancel,也就是pred的状态可能同时会有两条线程(pred和node)去操作)
             * 2/既然前驱节点已经设为SIGNAL了,为什么最后还要返回false
             * (因为CAS可能会失败,这里不管失败与否,都返回false,下一次执行该方法的之后,pred的等待状态就是SIGNAL了)
             * (网上摘抄的,解释的很明白)
             */
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        }
        return false;
    }

waitStatus状态值

状态

说明

CANCELLED

1

等待超时或者中断,需要从同步队列中取消

SIGNAL

-1

后继节点出于等待状态,当前节点释放锁后将会唤醒后继节点

CONDITION

-2

节点在等待队列中,节点线程等待在Condition上,其它线程对Condition调用signal()方法后,该节点将会从等待同步队列中移到同步队列中,然后等待获取锁。

PROPAGATE

-3

表示下一次共享式同步状态获取将会无条件地传播下去

INITIAL

0

初始状态

  1. 首先获取前驱节点的等待状态ws
  2. 如果ws为SIGNAL则表示可以被前驱节点唤醒,当前线程就可以挂起,等待前驱节点唤醒,返回true(可以挂起)
  3. 如果ws>0说明,前驱节点取消了,并循环查找此前驱节点之前所有连续取消的节点。并返回false(不能挂起)。
  4. 尝试将当前节点的前驱节点的等待状态设为SIGNAL
    private final boolean parkAndCheckInterrupt() {
        LockSupport.park(this);
        return Thread.interrupted();
    }

把当前线程挂起,并检查刚线程是否执行了interrupted方法,并返回true、false。

** 公平锁 **

公平锁和非公平锁实现方式是一样的,唯一不同的是tryAcquire方法的实现,下面是公平锁tryAcquire方法实现:

protected final boolean tryAcquire(int acquires) {
            final Thread current = Thread.currentThread();
            int c = getState();
            if (c == 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;
        }
  1. 首先获取当前锁状态,如果当前state==0(无锁),则进行获取锁操作
  2. hasQueuedPredecessors方法判断头结点是否当前线程,如果是当前线程则进行CAS更新获得锁,获取成功,把当前线程设置成独占锁。
  3. 如果不是头结点或获取锁失败则,则判断当前线程是否为独占锁,如果是,则当前state+1(可重入锁),表示获取锁成功,更新state值,并返回true。这里更新state变量,不需要CAS更新,因为,当前线程已经获得锁。

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

扫码关注云+社区

领取腾讯云代金券

年度创作总结 领取年终奖励