深入理解 ReentrantLock 源码

2022-03-20 本文已影响0人天还下着毛毛雨

image

特性

显示锁，自旋锁，可重入锁
aqs队列锁的实现
支持多条件唤醒
支持打断
支持公平,非公平锁
可尝试加锁

使用方式

ReentrantLock lock = new ReentrantLock();
try {
    // 加锁
    lock.lock();
    // do something
} finally {
    // 在finally中解锁,避免出现异常导致 没释放锁而死锁
    lock.unlock();
}

源码

1. lock() 公平锁

final void lock() {
    acquire(1);
}

AbstractQueuedSynchronizer. acquire(1)

public final void acquire(int arg) {
    // !tryAcquire(arg) 尝试加锁
    // acquireQueued(addWaiter(Node.EXCLUSIVE) 加入队列
    //  selfInterrupt 自我打断
    if (!tryAcquire(arg) &&
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        selfInterrupt();
}

tryAcquire(int acquires)

protected final boolean tryAcquire(int acquires) {
    // 获取当前线程
    final Thread current = Thread.currentThread();
    // 判断当前锁的状态, 0 代表没有人持有
    int c = getState();
    if (c == 0) {
        // 判断队列中是否有线程在等待,如果没有则cas改变锁的状态
        if (!hasQueuedPredecessors() &&
            compareAndSetState(0, acquires)) {
            // cas成功,设置当前线程为持有锁的线程
            setExclusiveOwnerThread(current);
            return true;
        }
    }
    // 有线程占有锁,判断持有线程是否有是自身,是的话重入
    else if (current == getExclusiveOwnerThread()) {
        // 锁状态 + 1
        int nextc = c + acquires;
        // 锁状态 < 0 ? 加锁次数太多,导致整形溢出？
        if (nextc < 0)
            throw new Error("Maximum lock count exceeded");
        // 设置锁的状态
        setState(nextc);
        return true;
    }
    // 抢锁失败
    return false;
}

hasQueuedPredecessors()

判断队列中是否有线程在排队等锁

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;
    // 头结点一般都是 空结点
    // 头结点 != 尾结点
    // 头结点下的next为空  或者 头结点的后一个结点的线程不是当前线程
    return h != t &&
        ((s = h.next) == null || s.thread != Thread.currentThread());
}

acquireQueued(final Node node, int arg) ：入队睡眠

如果当前锁被其他线程持有(非自身重入), 或者队列中有线程在排队,则加锁失败，需要创建Node节点,入队(如果队列还没有初始化则会初始化队列)，并且入队之后,会判断自己Node的前驱节点是不是head,是的话，会自旋拿一次锁。

自旋拿锁成功会把自己做为新的头节点,空Node

addWaiter(Node mode) : 创建Node节点

// 创建线程为当前线程的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节点的前节点设置为当前节点
    node.prev = pred;
    // 将新创建的Node节点设置为新的尾节点
    if (compareAndSetTail(pred, node)) {
        // 之前的尾节点的后驱节点设置为 新创建的Node对象
        pred.next = node;
                // 返回
        return node;
    }
}
// 如果尾节点为空,初始化队列
enq(node);
return node;

enq(Node mode) : 入队

如果队列还没有初始化,就会新建队列,队列的头结点和尾节点都为空的Node

将当前线程的Node节点接在之前的尾节点的后面，并设置成新的尾节点

private Node enq(final Node node) {
    for (;;) {
        // 尾节点
        Node t = tail;
        // 第一次进来,如果为空,说明队列还没有初始化
        // 第二次进来就不为空了
        if (t == null) { // Must initialize
            // 第一次进来,创建空的Node,将头节点和尾节点都设置成这个空的Node
            if (compareAndSetHead(new Node()))
                tail = head;
        } else {
            // 不为空走这里,传进来的Node的前驱设置成之前的尾节点
            node.prev = t;
            // 传进来的Node成为新的尾节点,
            if (compareAndSetTail(t, node)) {
                // 之前的尾节点的后驱节点设置为传进来的Node
                t.next = node;
                return t;
            }
        }
    }
}

acquireQueued ：自旋,失败后睡眠

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

shouldParkAfterFailedAcquire（Node pred, Node node）

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;

更改前驱节点的状态：

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    int ws = pred.waitStatus;
    // 判断前驱Node,是不是唤醒状态，是的话直接返回true,让当前Node去睡眠
    if (ws == Node.SIGNAL)
        /*
         * This node has already set status asking a release
         * to signal it, so it can safely park.
         */
        return true;
    // 如果是> 0，说明是取消状态
    if (ws > 0) {
        /*
         * Predecessor was cancelled. Skip over predecessors and
         * indicate retry.
         */
        do {
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        // 一直往前查找,知道找到非取消的状态的节点，作为当前Node的前驱
        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.
         */
        // 如果是0, cas将前驱Node的状态改为Node.SIGNAL, 外面第二次循环,直接返回true,让当前节点睡眠
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}

parkAndCheckInterrupt()

阻塞当前线程,直到被打断,打断后会清空打断标记,在下一次循环中继续进来阻塞。造成一种无法打断的假象。

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

加锁流程

当锁为无锁状态时：
image
线程t1来加锁：持有锁的线程改成t1,锁的状态改为1:被占有状态
image
线程t1没有释放锁,线程t2来抢锁, t2会创建队列,并且排队
image

线程t1没有释放锁,线程t2入队,线程t3来抢锁

image

lock() : 非公平锁

非公平锁的加锁流程与公平锁不同的是,非公平锁在加锁的时候,不会判断队列中是否有线程在排队,直接cas加锁

final void lock() {
    // cas改锁的状态
    if (compareAndSetState(0, 1))
        // 修改成功直接抢占锁,当锁的线程改为当前线程
        setExclusiveOwnerThread(Thread.currentThread());
    else
        // 尝试加锁
        acquire(1);
}

tryAcquire() & nonfairTryAcquire()

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

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

final boolean nonfairTryAcquire(int acquires) {
    final Thread current = Thread.currentThread();
    int c = getState();
    if (c == 0) {
        // 公平锁这里会判断队列是否有人在排序，非公平锁直接cas上手
        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;
}

再后面抢不到锁，进入队列之后和公平锁的逻辑是一模一样的。

公平和非公平之前在于入队之前尝试加锁，看不看队列中有无线程排队，如果加锁失败,入队了,那么就再没有公平和非公平之分

2. unlock()

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

    public final boolean release(int arg) {
        // 解锁, state一直减到0,锁才会被完全释放
        if (tryRelease(arg)) {
            Node h = head;
            // 当锁完全释放后
            if (h != null && h.waitStatus != 0)
                // 叫醒后面排队的线程
                unparkSuccessor(h);
            return true;
        }
        return false;
    }

tryRelease

释放锁，state - 1,减到0,锁才会被完全释放,当前持有锁线程设置为null

protected final boolean tryRelease(int releases) {
    int c = getState() - releases;
    if (Thread.currentThread() != getExclusiveOwnerThread())
        throw new IllegalMonitorStateException();
    boolean free = false;
    // 状态为0,锁才完全释放, 重入时,state会被累加到 > 1,需要解锁多次
    if (c == 0) {
        // 完全释放
        free = true;
        setExclusiveOwnerThread(null);
    }
    setState(c);
    return free;
}

unparkSuccessor

:唤醒队列自己后面最近的一个非取消状态的线程

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

解锁流程

image

3.lockInterruptibly()

和lock方法不同的是,该加锁方法会响应打断异常

public void lockInterruptibly() throws InterruptedException {
    sync.acquireInterruptibly(1);
}

private void doAcquireInterruptibly(int arg)
    throws InterruptedException {
    final Node node = addWaiter(Node.EXCLUSIVE);
    boolean failed = true;
    try {
        for (;;) {
            final Node p = node.predecessor();
            if (p == head && tryAcquire(arg)) {
                setHead(node);
                p.next = null; // help GC
                failed = false;
                return;
            }
            if (shouldParkAfterFailedAcquire(p, node) && 
                parkAndCheckInterrupt())
                // 如果再睡眠过程中被打断,会抛出异常
                throw new InterruptedException();
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

4.Condition 条件唤醒

await() : 等待

public final void await() throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    // 创建condition等待队列
    Node node = addConditionWaiter();
    int savedState = fullyRelease(node);
    int interruptMode = 0;
    while (!isOnSyncQueue(node)) {
        // 睡眠
        LockSupport.park(this);
        if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
            break;
    }
    if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
        interruptMode = REINTERRUPT;
    if (node.nextWaiter != null) // clean up if cancelled
        unlinkCancelledWaiters();
    if (interruptMode != 0)
        reportInterruptAfterWait(interruptMode);
}

addConditionWaiter()

如果尾节点为空，新建一个Node节点(持有当前线程),state为-2, 头节点和尾节点都等于新建的Node节点

如果尾节点不为空，新建一个Node节点(持有当前线程),接在尾节点的后面,并作为新的尾节点

private Node addConditionWaiter() {
    Node t = lastWaiter;
    // If lastWaiter is cancelled, clean out.
    if (t != null && t.waitStatus != Node.CONDITION) {
        unlinkCancelledWaiters();
        t = lastWaiter;
    }
    Node node = new Node(Thread.currentThread(), Node.CONDITION);
    if (t == null)
        firstWaiter = node;
    else
        t.nextWaiter = node;
    lastWaiter = node;
    return node;
}

signal() : 唤醒

唤醒条件队列的第一个Node

public final void signal() {
    if (!isHeldExclusively())
        throw new IllegalMonitorStateException();
    // 头节点
    Node first = firstWaiter;
    if (first != null)
        // 唤醒
        doSignal(first);
}

doSignal

private void doSignal(Node first) {
    do {
        if ( (firstWaiter = first.nextWaiter) == null)
            lastWaiter = null;
        first.nextWaiter = null;
    } while (!transferForSignal(first) &&
             (first = firstWaiter) != null);
}

transferForSignal(Node node)

final boolean transferForSignal(Node node) {
    /*
     * If cannot change waitStatus, the node has been cancelled.
     */
    if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
        return false;

    /*
     * Splice onto queue and try to set waitStatus of predecessor to
     * indicate that thread is (probably) waiting. If cancelled or
     * attempt to set waitStatus fails, wake up to resync (in which
     * case the waitStatus can be transiently and harmlessly wrong).
     */
    // 将当前节点从条件队列加入到 等待队列中
    // 返回是上一个节点
    Node p = enq(node);
    int ws = p.waitStatus;
    // 然后将上一个节点的状态改为SIGNAL -1
    if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
        // 如果cas失败，unpark
        LockSupport.unpark(node.thread);
    return true;
}

doSignalAll : 唤醒所有

将条件队列中所有的Node按顺序从头到尾加到等待队列中

public final void signalAll() {
    if (!isHeldExclusively())
        throw new IllegalMonitorStateException();
    Node first = firstWaiter;
    if (first != null)
        doSignalAll(first);
}

private void doSignalAll(Node first) {
    lastWaiter = firstWaiter = null;
    // 无限循环，直到条件队列中没有元素
    do {
        Node next = first.nextWaiter;
        first.nextWaiter = null;
        // 从头到尾按顺序,加到等待队列中
        transferForSignal(first);
        first = next;
    } while (first != null);
}

深入理解 ReentrantLock 源码

特性

使用方式

源码

1. lock() 公平锁

AbstractQueuedSynchronizer. acquire(1)

tryAcquire(int acquires)

hasQueuedPredecessors()

acquireQueued(final Node node, int arg) ：入队睡眠

addWaiter(Node mode) : 创建Node节点

enq(Node mode) : 入队

acquireQueued ：自旋,失败后睡眠

shouldParkAfterFailedAcquire（Node pred, Node node）

parkAndCheckInterrupt()

加锁流程

lock() : 非公平锁

tryAcquire() & nonfairTryAcquire()

2. unlock()

tryRelease

unparkSuccessor

解锁流程

3.lockInterruptibly()

4.Condition 条件唤醒

await() : 等待

addConditionWaiter()

signal() : 唤醒

doSignal

transferForSignal(Node node)

doSignalAll : 唤醒所有

猜你喜欢

热点阅读

深入理解 ReentrantLock 源码

特性

使用方式

源码

1. lock() 公平锁

AbstractQueuedSynchronizer. acquire(1)

tryAcquire(int acquires)

hasQueuedPredecessors()

acquireQueued(final Node node, int arg) ： 入队睡眠

addWaiter(Node mode) : 创建Node节点

enq(Node mode) : 入队

acquireQueued ：自旋,失败后睡眠

shouldParkAfterFailedAcquire（Node pred, Node node）

parkAndCheckInterrupt()

加锁流程

lock() : 非公平锁

tryAcquire() & nonfairTryAcquire()

2. unlock()

tryRelease

unparkSuccessor

解锁流程

3.lockInterruptibly()

4.Condition 条件唤醒

await() : 等待

addConditionWaiter()

signal() : 唤醒

doSignal

transferForSignal(Node node)

doSignalAll : 唤醒所有

猜你喜欢

热点阅读

acquireQueued(final Node node, int arg) ：入队睡眠