@@ -1,58 +0,0 @@ ReentrantLock源码分析(二)-获取锁细节 | 凤凰涅槃进阶之路

ReentrantLock源码分析(二)-获取锁细节

Abel sun2022年12月24日约 5082 字大约 17 分钟

ReentrantLock源码分析(二)-获取锁细节

1. ReentrantLock 中线程加入等待队列

1.1 加入队列的时机

当执行Acquire(1)时,会通过tryAcquire获取锁。在这种情况下,如果获取锁失败,就会调用 addWaiter加入到等待队列中去。

在AQS 中

 public final void acquire(int arg) {
        // 当执行Acquire(1)时,会通过tryAcquire获取锁。在这种情况下,如果获取锁失败,就会调用 addWaiter加入到等待队列中去。
        if (!tryAcquire(arg) &&
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }

在 ReentrantLock

/**
     * 非公平锁
     * Sync object for non-fair locks
     */
    static final class NonfairSync extends Sync {

        /**
         * 获得锁
         */
        final void lock() {
            /**
             * 若通过CAS设置变量State(同步状态)成功,也就是获取锁成功,则将当前线程设置为独占线程。
             * 若通过CAS设置变量State(同步状态)失败,也就是获取锁失败,则进入Acquire方法进行后续处理。
             */
            if (compareAndSetState(0, 1))
                // 把当前线程设置独占了锁
                setExclusiveOwnerThread(Thread.currentThread());
            else// 锁已经被占用,或者set失败
                // 以独占模式获取对象,忽略中断
                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() {
            // 以独占模式获取对象,忽略中断
            // aqs 的 acquire 方法
            acquire(1);
        }
    }

1.2 如何加入队列

获取锁失败后,会执行 addWaiter(Node.EXCLUSIVE)加入等待队列,具体实现方法如下:

 /**
     * 获取锁失败后,会执行 addWaiter(Node.EXCLUSIVE)加入等待队列,具体实现方法如下:
     * 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) {
        // 1. 通过当前的线程和锁模式新建一个节点。
        Node node = new Node(Thread.currentThread(), mode);
        // Try the fast path of enq; backup to full enq on failure
        // 2. Pred指针指向尾节点Tail。
        Node pred = tail;
        if (pred != null) {
            // 3. 将New中Node的Prev指针指向Pred。
            node.prev = pred;
            // 4. 通过compareAndSetTail方法,完成尾节点的设置。这个方法主要是对 tailOffset和 Expect进行比较,
            // 如果 tailOffset的 Node和 Expect的 Node地址是相同的,那么设置 Tail的值为 Update的值。
            if (compareAndSetTail(pred, node)) {
                pred.next = node;
                return node;
            }
        }
        enq(node);
        return node;
    }

private final boolean compareAndSetTail(Node expect, Node update) {
    return unsafe.compareAndSwapObject(this, tailOffset, expect, update);
}

主要的流程如下:

  1. 通过当前的线程和锁模式新建一个节点。

  2. Pred指针指向尾节点Tail。

  3. 将New中Node的Prev指针指向Pred。

  4. 通过compareAndSetTail方法,完成尾节点的设置。这个方法主要是对 tailOffset和 Expect进行比较,如果 tailOffset的 Node和 Expect的 Node地址是相同的,那么设置 Tail的值为 Update的值。

     // java.util.concurrent.locks.AbstractQueuedSynchronizer
     static {
         try {
             stateOffset = unsafe.objectFieldOffset(AbstractQueuedSynchronizer.class.getDeclaredField("state"));
             headOffset = unsafe.objectFieldOffset(AbstractQueuedSynchronizer.class.getDeclaredField("head"));
             tailOffset = unsafe.objectFieldOffset(AbstractQueuedSynchronizer.class.getDeclaredField("tail"));
             waitStatusOffset = unsafe.objectFieldOffset(Node.class.getDeclaredField("waitStatus"));
             nextOffset = unsafe.objectFieldOffset(Node.class.getDeclaredField("next"));
        } catch (Exception ex) { 
        throw new Error(ex); 
      }
    }
    

    从AQS的静态代码块可以看出,都是获取一个对象的属性相对于该对象在内存当中的偏移量,这样我们就可以根据这个偏移量在对象内存当中找到这个属性。tailOffset指的是 tail对应的偏移量,所以这个时候会将 new出来的 Node置为当前队列的尾节点。同时,由于是双向链表,也需要将前一个节点指向尾节点。

  5. 如果 Pred指针是 Null(说明等待队列中没有元素),或者当前 Pred指针和 Tail指向的位置不同(说明被别的线程已经修改),就需要看一下 Enq的方法。

      // java.util.concurrent.locks.AbstractQueuedSynchronizer
      
      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;
                 }
             }
         }
     }
    
    

    如果没有被初始化,需要进行初始化一个头结点出来。但请注意,初始化的头结点并不是当前线程节点,而是调用了无参构造函数的节点。如果经历了初始化或者并发导致队列中有元素,则与之前的方法相同。其实,addWaiter就是一个在双端链表添加尾节点的操作,需要注意的是,双端链表的头结点是一个无参构造函数的头结点

1.3 线程获取锁 过程总结

  1. 当没有线程获取到锁时,线程1获取锁成功。
  2. 线程2申请锁,但是锁被线程1占有。

image-20220521202652797

  1. 如果再有线程要获取锁,依次在队列中往后排队即可。

1.4 公平锁加锁时判断等待队列中是否存在有效节点

回到上边的代码,hasQueuedPredecessors是公平锁加锁时判断等待队列中是否存在有效节点的方法。如果返回False,说明当前线程可以争取共享资源;如果返回True,说明队列中存在有效节点,当前线程必须加入到等待队列中。

 
 public final boolean hasQueuedPredecessors() {
     // The correctness of this depends on head being initialized
     // before tail and on head.next being accurate if the current
     // thread is first in queue.
     Node t = tail; // Read fields in reverse initialization order
     Node h = head;
     Node s;
     return h != t && ((s = h.next) == null || s.thread != Thread.currentThread());
 }

看到这里,我们理解一下h != t && ((s = h.next) == null || s.thread != Thread.currentThread());为什么要判断的头结点的下一个节点?第一个节点储存的数据是什么?

双向链表中,第一个节点为虚节点,其实并不存储任何信息,只是占位。真正的第一个有数据的节点,是在第二个节点开始的。当h != t时: 如果(s = h.next) == null,等待队列正在有线程进行初始化,但只是进行到了Tail指向Head,没有将Head指向Tail,此时队列中有元素,需要返回True(这块具体见下边代码分析)。 如果(s = h.next) != null,说明此时队列中至少有一个有效节点。如果此时s.thread == Thread.currentThread(),说明等待队列的第一个有效节点中的线程与当前线程相同,那么当前线程是可以获取资源的;如果s.thread != Thread.currentThread(),说明等待队列的第一个有效节点线程与当前线程不同,当前线程必须加入进等待队列。

 // java.util.concurrent.locks.AbstractQueuedSynchronizer#enq
 
 if (t == null) { // Must initialize
     if (compareAndSetHead(new Node()))
         tail = head;
 } else {
     node.prev = t;
     if (compareAndSetTail(t, node)) {
         t.next = node;
         return t;
     }
 }

节点入队不是原子操作,所以会出现短暂的head != tail,此时Tail指向最后一个节点,而且Tail指向Head。如果Head没有指向Tail(可见5、6、7行),这种情况下也需要将相关线程加入队列中。所以这块代码是为了解决极端情况下的并发问题。

1.5 等待队列中线程出队列时机

回到最初的源码:

// java.util.concurrent.locks.AbstractQueuedSynchronizer
public final void acquire(int arg) {
    if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        selfInterrupt();
}

上文解释了addWaiter方法,这个方法其实就是把对应的线程以 Node的数据结构形式加入到双端队列里,返回的是一个包含该线程的Node。而这个 Node会作为参数,进入到 acquireQueued方法中。acquireQueued方法可以对排队中的线程进行“获锁”操作。总的来说,一个线程获取锁失败了,被放入等待队列,acquireQueued会把放入队列中的线程不断去获取锁,直到获取成功或者不再需要获取(中断)。

下面我们从“何时出队列?”和“如何出队列?”两个方向来分析一下acquireQueued源码:

// java.util.concurrent.locks.AbstractQueuedSynchronizer

final boolean acquireQueued(final Node node, int arg) {
    // 标记是否成功拿到资源
    boolean failed = true;
    try {
        // 标记等待过程中是否中断过
        boolean interrupted = false;
        // 开始自旋,要么获取锁,要么中断
        for (;;) {
            // 获取当前节点的前驱节点
            final Node p = node.predecessor();
            // 如果p是头结点,说明当前节点在真实数据队列的首部,就尝试获取锁(别忘了头结点是虚节点)
            if (p == head && tryAcquire(arg)) {
                // 获取锁成功,头指针移动到当前node
                setHead(node);
                p.next = null; // help GC
                failed = false;
                return interrupted;
            }
            // 说明p为头节点且当前没有获取到锁(可能是非公平锁被抢占了)或者是p不为头结点,这个时候就要判断当前node是否要被阻塞(被阻塞条件:前驱节点的waitStatus为-1),防止无限循环浪费资源。具体两个方法下面细细分析
            if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

注:setHead方法是把当前节点置为虚节点,但并没有修改waitStatus,因为它是一直需要用的数据。

// java.util.concurrent.locks.AbstractQueuedSynchronizer

private void setHead(Node node) {
    head = node;
    node.thread = null;
    node.prev = null;
}

// java.util.concurrent.locks.AbstractQueuedSynchronizer

// 靠前驱节点判断当前线程是否应该被阻塞
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    // 获取头结点的节点状态
    int ws = pred.waitStatus;
    // 说明头结点处于唤醒状态
    if (ws == Node.SIGNAL)
        return true; 
    // 通过枚举值我们知道waitStatus>0是取消状态
    if (ws > 0) {
        do {
            // 循环向前查找取消节点,把取消节点从队列中剔除
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        pred.next = node;
    } else {
        // 设置前任节点等待状态为SIGNAL
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}

parkAndCheckInterrupt主要用于挂起当前线程,阻塞调用栈,返回当前线程的中断状态。

// java.util.concurrent.locks.AbstractQueuedSynchronizer

private final boolean parkAndCheckInterrupt() {
    LockSupport.park(this);
    return Thread.interrupted();
}

上述方法的流程图如下:

image-20220521205100169

从上图可以看出,跳出当前循环的条件是当“前置节点是头结点,且当前线程获取锁成功”。为了防止因死循环导致CPU资源被浪费,我们会判断前置节点的状态来决定是否要将当前线程挂起,具体挂起流程用流程图表示如下(shouldParkAfterFailedAcquire流程):

image-20220521205119638

从队列中释放节点的疑虑打消了,那么又有新问题了:

  • shouldParkAfterFailedAcquire中取消节点是怎么生成的呢?什么时候会把一个节点的waitStatus设置为-1?
  • 是在什么时间释放节点通知到被挂起的线程呢?

2. CANCELLED状态节点生成

acquireQueued方法中的 Finally代码:

// java.util.concurrent.locks.AbstractQueuedSynchronizer

final boolean acquireQueued(final Node node, int arg) {
    boolean failed = true;
    try {
    ...
        for (;;) {
            final Node p = node.predecessor();
            if (p == head && tryAcquire(arg)) {
                ...
                failed = false;
        ...
            }
            ...
    } finally {
        if (failed)
            cancelAcquire(node);
        }
}

通过cancelAcquire方法,将Node的状态标记为CANCELLED。接下来,我们逐行来分析这个方法的原理:

// java.util.concurrent.locks.AbstractQueuedSynchronizer

private void cancelAcquire(Node node) {
  // 将无效节点过滤
    if (node == null)
        return;
  // 设置该节点不关联任何线程,也就是虚节点
    node.thread = null;
    Node pred = node.prev;
  // 通过前驱节点,跳过取消状态的node
    while (pred.waitStatus > 0)
        node.prev = pred = pred.prev;
  // 获取过滤后的前驱节点的后继节点
    Node predNext = pred.next;
  // 把当前node的状态设置为CANCELLED
    node.waitStatus = Node.CANCELLED;
  // 如果当前节点是尾节点,将从后往前的第一个非取消状态的节点设置为尾节点
  // 更新失败的话,则进入else,如果更新成功,将tail的后继节点设置为null
    if (node == tail && compareAndSetTail(node, pred)) {
        compareAndSetNext(pred, predNext, null);
    } else {
        int ws;
    // 如果当前节点不是head的后继节点,1:判断当前节点前驱节点的是否为SIGNAL,2:如果不是,则把前驱节点设置为SINGAL看是否成功
    // 如果1和2中有一个为true,再判断当前节点的线程是否为null
    // 如果上述条件都满足,把当前节点的前驱节点的后继指针指向当前节点的后继节点
        if (pred != head && ((ws = pred.waitStatus) == Node.SIGNAL || (ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) && pred.thread != null) {
            Node next = node.next;
            if (next != null && next.waitStatus <= 0)
                compareAndSetNext(pred, predNext, next);
        } else {
      // 如果当前节点是head的后继节点,或者上述条件不满足,那就唤醒当前节点的后继节点
            unparkSuccessor(node);
        }
        node.next = node; // help GC
    }
}

当前的流程:

  • 获取当前节点的前驱节点,如果前驱节点的状态是CANCELLED,那就一直往前遍历,找到第一个waitStatus <= 0的节点,将找到的Pred节点和当前Node关联,将当前Node设置为CANCELLED。
  • 根据当前节点的位置,考虑以下三种情况:

(1) 当前节点是尾节点。

(2) 当前节点是Head的后继节点。

(3) 当前节点不是Head的后继节点,也不是尾节点。

根据上述第二条,我们来分析每一种情况的流程。

当前节点是尾节点。

image-20220521205416054

当前节点是Head的后继节点。

image-20220521205431385

当前节点不是Head的后继节点,也不是尾节点。

image-20220521205446769

通过上面的流程,我们对于CANCELLED节点状态的产生和变化已经有了大致的了解,但是为什么所有的变化都是对Next指针进行了操作,而没有对Prev指针进行操作呢?什么情况下会对Prev指针进行操作?

执行cancelAcquire的时候,当前节点的前置节点可能已经从队列中出去了(已经执行过Try代码块中的shouldParkAfterFailedAcquire方法了),如果此时修改Prev指针,有可能会导致Prev指向另一个已经移除队列的Node,因此这块变化Prev指针不安全。 shouldParkAfterFailedAcquire方法中,会执行下面的代码,其实就是在处理Prev指针。shouldParkAfterFailedAcquire是获取锁失败的情况下才会执行,进入该方法后,说明共享资源已被获取,当前节点之前的节点都不会出现变化,因此这个时候变更Prev指针比较安全。

do {
    node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);

3. 如何解锁

我们已经剖析了加锁过程中的基本流程,接下来再对解锁的基本流程进行分析。由于 ReentrantLock在解锁的时候,并不区分公平锁和非公平锁,所以我们直接看解锁的源码:

// java.util.concurrent.locks.ReentrantLock

public void unlock() {
    sync.release(1);
}

可以看到,本质释放锁的地方,是通过框架来完成的。

// java.util.concurrent.locks.AbstractQueuedSynchronizer

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

在ReentrantLock里面的公平锁和非公平锁的父类Sync定义了可重入锁的释放锁机制。

// java.util.concurrent.locks.ReentrantLock.Sync

// 方法返回当前锁是不是没有被线程持有
protected final boolean tryRelease(int releases) {
    // 减少可重入次数
    int c = getState() - releases;
    // 当前线程不是持有锁的线程,抛出异常
    if (Thread.currentThread() != getExclusiveOwnerThread())
        throw new IllegalMonitorStateException();
    boolean free = false;
    // 如果持有线程全部释放,将当前独占锁所有线程设置为null,并更新state
    if (c == 0) {
        free = true;
        setExclusiveOwnerThread(null);
    }
    setState(c);
    return free;
}

我们来解释下述源码:

// java.util.concurrent.locks.AbstractQueuedSynchronizer

public final boolean release(int arg) {
    // 上边自定义的tryRelease如果返回true,说明该锁没有被任何线程持有
    if (tryRelease(arg)) {
        // 获取头结点
        Node h = head;
        // 头结点不为空并且头结点的waitStatus不是初始化节点情况,解除线程挂起状态
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);
        return true;
    }
    return false;
}

这里的判断条件为什么是h != null && h.waitStatus != 0?

h == null Head还没初始化。初始情况下,head == null,第一个节点入队,Head会被初始化一个虚拟节点。所以说,这里如果还没来得及入队,就会出现head == null 的情况。

h != null && waitStatus == 0 表明后继节点对应的线程仍在运行中,不需要唤醒。

h != null && waitStatus < 0 表明后继节点可能被阻塞了,需要唤醒。

再看一下 unparkSuccessor方法:

// java.util.concurrent.locks.AbstractQueuedSynchronizer

private void unparkSuccessor(Node node) {
    // 获取头结点waitStatus
    int ws = node.waitStatus;
    if (ws < 0)
        compareAndSetWaitStatus(node, ws, 0);
    // 获取当前节点的下一个节点
    Node s = node.next;
    // 如果下个节点是null或者下个节点被cancelled,就找到队列最开始的非cancelled的节点
    if (s == null || s.waitStatus > 0) {
        s = null;
        // 就从尾部节点开始找,到队首,找到队列第一个waitStatus<0的节点。
        for (Node t = tail; t != null && t != node; t = t.prev)
            if (t.waitStatus <= 0)
                s = t;
    }
    // 如果当前节点的下个节点不为空,而且状态<=0,就把当前节点unpark
    if (s != null)
        LockSupport.unpark(s.thread);
}

为什么要从后往前找第一个非Cancelled的节点呢?原因如下。

之前的 addWaiter方法:

// java.util.concurrent.locks.AbstractQueuedSynchronizer

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

我们从这里可以看到,节点入队并不是原子操作,也就是说,node.prev = pred; compareAndSetTail(pred, node) 这两个地方可以看作Tail入队的原子操作,但是此时pred.next = node;还没执行,如果这个时候执行了unparkSuccessor方法,就没办法从前往后找了,所以需要从后往前找。还有一点原因,在产生CANCELLED状态节点的时候,先断开的是Next指针,Prev指针并未断开,因此也是必须要从后往前遍历才能够遍历完全部的Node。

综上所述,如果是从前往后找,由于极端情况下入队的非原子操作和CANCELLED节点产生过程中断开Next指针的操作,可能会导致无法遍历所有的节点。所以,唤醒对应的线程后,对应的线程就会继续往下执行。继续执行acquireQueued方法以后,中断如何处理?

4. 中断恢复后的执行流程

唤醒后,会执行 return Thread.interrupted();,这个函数返回的是当前执行线程的中断状态,并清除。

// java.util.concurrent.locks.AbstractQueuedSynchronizer

private final boolean parkAndCheckInterrupt() {
    LockSupport.park(this);
    return Thread.interrupted();
}

再回到 acquireQueued代码,当 parkAndCheckInterrupt返回True或者False的时候,interrupted的值不同,但都会执行下次循环。如果这个时候获取锁成功,就会把当前 interrupted返回。

// java.util.concurrent.locks.AbstractQueuedSynchronizer

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为True,就会执行selfInterrupt方法。

// java.util.concurrent.locks.AbstractQueuedSynchronizer

static void selfInterrupt() {
    Thread.currentThread().interrupt();
}

该方法其实是为了中断线程。但为什么获取了锁以后还要中断线程呢?这部分属于Java提供的协作式中断知识内容,感兴趣同学可以查阅一下。这里简单介绍一下:

  1. 当中断线程被唤醒时,并不知道被唤醒的原因,可能是当前线程在等待中被中断,也可能是释放了锁以后被唤醒。因此我们通过Thread.interrupted()方法检查中断标记(该方法返回了当前线程的中断状态,并将当前线程的中断标识设置为False),并记录下来,如果发现该线程被中断过,就再中断一次。
  2. 线程在等待资源的过程中被唤醒,唤醒后还是会不断地去尝试获取锁,直到抢到锁为止。也就是说,在整个流程中,并不响应中断,只是记录中断记录。最后抢到锁返回了,那么如果被中断过的话,就需要补充一次中断。

这里的处理方式主要是运用线程池中基本运作单元Worder中的runWorker,通过Thread.interrupted()进行额外的判断处理,感兴趣的同学可以看下ThreadPoolExecutor源码。

5. 小结

  • Q:某个线程获取锁失败的后续流程是什么呢?

    A:存在某种排队等候机制,线程继续等待,仍然保留获取锁的可能,获取锁流程仍在继续。

  • Q:既然说到了排队等候机制,那么就一定会有某种队列形成,这样的队列是什么数据结构呢?

    A:是CLH变体的FIFO双端队列。

  • Q:处于排队等候机制中的线程,什么时候可以有机会获取锁呢?

    A:可以详细看下前面章节

  • Q:如果处于排队等候机制中的线程一直无法获取锁,需要一直等待么?还是有别的策略来解决这一问题?

    A:线程所在节点的状态会变成取消状态,取消状态的节点会从队列中释放,具体可见2.3.2小节。

  • Q:Lock函数通过Acquire方法进行加锁,但是具体是如何加锁的呢?

    A:AQS的Acquire会调用tryAcquire方法,tryAcquire由各个自定义同步器实现,通过tryAcquire完成加锁过程。

参考文章

ReentrantLock 锁详解open in new window

评论
  • 按正序
  • 按倒序
  • 按热度
Powered by Waline v2.9.1