Java并发编程 - 深入剖析ReentrantLock之非公平

作者: HRocky | 来源:发表于2019-03-13 22:46 被阅读1次

Java并发编程 - 深入剖析ReentrantLock之非公平锁解锁流程（第2篇）

前面两篇文章我们调试的方式梳理了一下流程，这篇文章我们对流程中的一些小细节点进行单独详细得说明。

1. head节点在什么时候创建的

当我们创建了一个ReentrantLock对象之后，对象内部的head节点是null的，那么什么会对这个head赋值呢？

经过分析我们可以知道，节点的创建和入同步队列是通过addWaiter方法，方法如下：

AbstractQueuedSynchronizer.java

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

同步队列初始状态，tail是null，所以这里if条件不成立，不会执行。

接着执行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;
            }
        }
    }
}

tail=null，第一个if条件成立，

if (compareAndSetHead(new Node()))
    tail = head;

这里就是初始head节点，同时将head赋值给了tail，tail也被初始化，结果是head和tail被初始化后执行了同一个节点对象。

2. 出现争抢设置head和tail是怎么处理的

上面我们讲了head和tail的初始化，不过有个问题，在多线程的条件下，不同的线程会有竞争，这种竞争也会发生在争抢设置head和tail上，比如说两个线程到来时，同步队列的head和tail还未初始化，那么两个线程都有可能会几乎同时设置它们，怎么处理？

这个是通过CAS来解决的，我们可以看到head和tail的设置都是CAS话的，我们还是来看上面关于head的设置：

if (compareAndSetHead(new Node()))
    tail = head;

这里就是通过CAS保证了设值的原子性和有序性，同时有个地方也很关键，那就是上层是个无限循环，这样就保证了假如一个线程执行到这里来，刚好另外一个线程已经把head给设置了，这时候这个if的条件就不满足了，就继续循环，tail不为空了，也就执行了另外的逻辑。

对于tail的设置也是一样的，多个不同的线程同样会争抢将代表自己的Node节点设置为tail，跟上面head初始化一样，也是通过CAS进行处理：

if (compareAndSetTail(t, node)) {
    t.next = node;
    return t;
}

通过CAS的帮助，直到把代表自己的Node节点设置为tail。

3. 同步队列中被唤醒的线程一定能获得到锁吗

AbstractQueuedSynchronized.java

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

通过前两篇的分析我们知道，当有其他线程占用锁的时候，代表当前线程的Node节点入同步队列，然后线程执行到parkAndCheckInterrupt被挂起，这里的for循环就不再执行，线程被唤醒后继续执行，根据我们之前的分析，这个被唤醒的线程所在的节点肯定是head的后继节点（注意这里说的是被唤醒，这里不考虑等待线程被中断），也就是说被唤醒的线程重新执行for循环，那么第一个if语句的p==head是满足的，然后被唤醒的线程调用tryAcquire方法重新申请获取锁的使用权，但是如果此时刚好一个新的线程（非同步队列等待线程）到来，那么会发生怎样的情况呢？

新来的线程来说可能会执行下面的逻辑。

第1处：

NonfairSync

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

新来的线程刚好执行到if这里，通过CAS成功将state设置为1了，那么它就立即获取到了锁。被唤醒线程执行tryAcquire会返回false，获取锁失败。假如被唤醒的唤醒抢到了，那么新来的线程入同步队列。

第2处：

在前一个持有锁的线程将state设置为0，而还未执行唤醒逻辑之前，新来的线程可能会执行到：

 acquire(1);

这处逻辑，接下来和被唤醒线程一起通过tryAcquire争抢锁的所有权。

NonfairSync

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

Sync

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

这里同样的处理逻辑，谁先CAS修改state成功，谁就获得了锁的使用权。

上面说过，新来的线程获取锁失败，就会入同步队列，那么被唤醒线程获取锁失败会怎么样？

回到AbstractQueuedSynchronizer类的acquireQueued方法中的这段逻辑：

AbstractQueuedSynchronizer->acquireQueued

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

此时tryAcquire返回是false的，if块代码不会执行，外层是无限循环，被唤醒线程重新执行，只要是它没抢到锁，那么就会一直循环下去。

我先来的为什么我被唤醒了我还抢不到锁？这就是为什么说是“非公平锁”的原因。也就是说被唤醒的线程不一定可以获得锁。

同步队列中等待的线程被中断了会出现什么情况

如果同步队列中的正在等待的线程，被其他线程中断了，是如何处理的？

依然拿我们第1篇的示例代码测试，加上线程中断逻辑，代码如下：

ReentrantLockExample.java

import java.util.concurrent.locks.ReentrantLock;

public class ReentrantLockExample {

    private ReentrantLock lock = new ReentrantLock();

    public void doSomething() {

        lock.lock();

        try {
            System.out.println(Thread.currentThread().getName() + " has been acquired the lock...");
        } finally {
            lock.unlock();
            System.out.println(Thread.currentThread().getName() + " has been released the lock...");
        }

    }

    public static void main(String[] args) {

        ReentrantLockExample lockExample = new ReentrantLockExample();

        Thread A = new Thread(()->{
            lockExample.doSomething();
        }, "thread-A");

        Thread B = new Thread(()->{
            lockExample.doSomething();
        }, "thread-B");

        Thread C = new Thread(()->{
                lockExample.doSomething();
            }, "thread-C");

        Thread D = new Thread(()->{
                lockExample.doSomething();
        }, "thread-D");


        A.start();
        B.start();
        C.start();
        D.start();

        Thread interruptThread = new Thread(()->{
            B.interrupt();// 根据具体测试情况修改
        });
        oneThread.start();

    }

}

来看一下，线程的挂起点，根据之前的分析，挂起点是LockSupport.park(this)处：

AbstractQueuedSynchronizer.java

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

调用处：

AbstractQueuedSynchronizer.java

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

下面进行测试：

同步队列中的代表线程的节点我们分3种：头直接后继节点、中间节点（非头直接后继）和尾节点。

中断头直接后继节点代表的线程

1. 执行顺序：线程A释放锁 -> 中断线程B -> 执行线程B

第一步：线程A释放锁；

AbstractQueuedSynchronizer.java

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

执行完tryRelease后，暂停线程A。测试可以看到线程A已经将state改为了0。

第二步：中断线程B；

B.interrupt();

第三步：执行线程B

测试可以看到，线程B被唤醒后，从挂起点之后的代码开始执行：

AbstractQueuedSynchronizer.java

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

因为此时当前线程也就是线程B是中断的，所以这个方法返回true。

这里也有个需要注意的地方，上面是通过调用Thread.interrupted()返回线程的中断状态的，而这个方法调用后会清除线程的中断状态。

回到acquireQueued方法：

if(shouldParkAfterFailedAcquire(p, node) &&
    parkAndCheckInterrupt())
    interrupted = true;

if条件测试，interrupted被设值为true。

继续下次循环，第一个if条件成立。现在会出现两种情况，因为可能会有新来的线程争抢锁，所以线程B可能争抢到锁也可能争抢不到：

线程B通过执行tryAcquire争抢到锁

执行if代码块内容，这就和我们正常唤醒逻辑一样了，把线程B代表的节点设置为head节点，唯一不同的是这时候interrupted是true。

当一个代表线程的节点成功被设置为头节点后，这个节点内部的thread属性就被设置成null了，那么这个节点就不再代表线程了。

这里要注意了，也就是说执行完后acquireQueued返回的是true，回到acquireQueued方法的调用处：

ReentrantLock.java

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

这时候因为acquireQueued返回的是true，所以if条件成立。if代码块会执行：

AbstractQueuedSynchronizer.java

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

这里我什么要再调用一次自己中断自己呢？

我的猜测是等待的线程被其他线程打断，而我们上面调用了Thread.interrupted()这个清除了线程的中断状态，但是从lock（这些执行都相当于lock调用的内部）外部来看我并不知道内部做了这些事，如果lock执行结束获取线程的中断状态，发现它是false的，那么就会很奇怪，明明它被中断过为什么这个状态突然消失了呢，所以这里调用自打断重新设置状态。

这里只是猜测，中断状态复位，那为什么parkAndCheckInterrupt方法不调用不清位的isInterrupted()方法呢？待研究~

线程B通过tryAcquire未争抢到锁

未争抢到锁，第一个if代码块不执行，执行if条件判断，执行完线程B会被重新挂起。

线程A在线程B未释放锁之前继续之前的释放锁未完成的代码

线程B获得了锁，这时候线程B未释放锁，而是由线程A执行释放锁后的唤醒操作，情况是这么样的。

这里有点意思，来分析一下。

线程A继续执行release的逻辑：

AbstractQueuedSynchronizer.java

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

有一个地方是关键就是：

Node h = head;

如果被唤醒的线程B在将代表它的线程设置为头节点之前，线程A还未执行这段代码，那么线程A重新来执行后，这个h引用指向的就是线程B操作后新的头结点，那么h的next就是代表线程C的节点，那么线程A就会唤醒线程C。
如果被唤醒的线程B在将代表它的线程设置为头节点之前，线程A还已经执行了这段代码，因为线程B在将代表它的点设置头节点后，会把原头结点的next设置为null，而这个h还指向的是原节点，unparkSuccessor方法的表现就有些不同了,我们来看看：

AbstractQueuedSynchronizer.java

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

在上面代码中，此时node为原头节点，node.next为空，则s=null, 会执行if里面的逻辑。

for循环的逻辑是从尾节点开始不断地找满足waitStatus <= 0的节点，在我们的例子中，最终找到的是头节点，头节点是无线程绑定的，所以s.thread == null，

LockSupport.java

public static void unpark(Thread thread) {
    if (thread != null)
        UNSAFE.unpark(thread);
}

upark不唤醒任何线程。

2. 执行顺序：线程A未释放锁 -> 中断线程B -> 执行线程B

在线程A未释放锁的时候，线程B被中断，然后继续执行：

AbstractQueuedSynchronizer.java

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

这个就比较简单了，p==head满足，因为A还未释放锁，所以这里tryAcquire返回false。第一个if条件不满足。

p为头结点，waitStatus状态为-1，shouldParkAfterFailedAcquire返回true，执行parkAndCheckInterrupt，线程B又重新被挂起。

中断中间节点代表的线程

我们这里测试C节点代表的线程的唤醒。

1. 执行顺序：线程A释放锁 -> 中断线程C -> 执行线程C

其中的一些具体细节和上面一样，这里就不做具体的分析。只关注线程C的执行这个分支。

线程C被唤醒后执行循环，此时C的前置节点是B（p==B），非头结点，所以第一个if语句不执行，B的waitStatus状态为-1，执行parkAndCheckInterrupt()，线程C被重新挂起。

2. 执行顺序：线程A未释放锁 -> 中断线程C -> 执行线程C

和上面一致。

中断尾节点代表的线程

和上面一致。

总结

同步队列中的线程T被中断：

若它为非head直接后继节点，那么它无权利参与锁的争抢，会重新被挂起，等待唤醒；
若它为head直接后继节点，那么它有权利参与锁的争抢，结果是争抢到锁或者重新被挂起。

4. 可重入性是怎么实现的

可重入性解决的是持有锁的线程重复请求锁的问题。ReentrantLock从名字上就可以知道它是支持重入的，那么它是怎么实现的呢？

我们知道synchronized是可重入的，它在锁上绑定持有锁的线程，然后通过计数来记录锁重入的次数，加锁就+1，解锁就-1。计数为0表示锁被释放。

ReentrantLock.java

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

看else if部分，也是判断当前线程是否是之前持有锁的线程，然后是的话计数加1。

看下解锁：

ReentrantLock.java

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

首先判断当前的线程是不是持有锁的线程，是的话计数减1，减到0表示锁释放。

Java并发编程 - 深入剖析ReentrantLock之非公平

1. head节点在什么时候创建的

2. 出现争抢设置head和tail是怎么处理的

3. 同步队列中被唤醒的线程一定能获得到锁吗

同步队列中等待的线程被中断了会出现什么情况

中断头直接后继节点代表的线程

中断中间节点代表的线程

中断尾节点代表的线程

总结

4. 可重入性是怎么实现的

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

Java并发编程