[数据结构与算法]ReentrantLock源码分析

前言

本文从源码来介绍下ReentrantLock的实现细节

源码

话不多说直接上源码，从lock方法开始看
当我们不传参是，默认是NonfairSync，非公平锁，本文只看非公平锁，看懂了非公平锁，则公平锁只是一些细节的差异

public ReentrantLock() {
    sync = new NonfairSync();
}

public void lock() {
    sync.lock();
}

final void lock() {
	// 如果state=0，则cas尝试修改state=1
    if (compareAndSetState(0, 1))
    	// 修改成功则设置当前占用锁的线程
        setExclusiveOwnerThread(Thread.currentThread());
    else
    	// 核心代码
        acquire(1);
}

上面前半部分可以看出来非公平锁，一进来就尝试去获取锁，这就是非公平锁的特性，可以插队，即使当前有人在排队，我也可以插队

继续看acquire(1)

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

这里要分步骤看了，先看tryAcquire

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) {
    	// 进来之后还是会去再次抢占一次，和上面代码一样
        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");
        // state的值++
        setState(nextc);
        return true;
    }
    return false;
}

也就是tryAcquire主要是再次抢占锁和判断是否当前可以重入的

继续看，当没有抢占到锁则addWaiter(Node.EXCLUSIVE)

private Node addWaiter(Node mode) {
	// 构建当前节点
    Node node = new Node(Thread.currentThread(), mode);
    // Try the fast path of enq; backup to full enq on failure
    // 下面这段逻辑是设置tail节点，只有tail节点不为空才会去添加到尾结点，并且更新tail尾结点
    // 当tail节点不为空，则说明head节点也不为空，那么只需要添加尾结点和更新
    Node pred = tail;
    if (pred != null) {
        node.prev = pred;
        if (compareAndSetTail(pred, node)) {
            pred.next = node;
            return node;
        }
    }
    // 核心2，初始化或者添加尾结点
    enq(node);
    return node;
}

private Node enq(final Node node) {
	// 循环，cas循环构建head节点和tail节点，因为上面是一次cas，这里for循环，那么最终必定成功
    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;
            }
        }
    }
}

到这里没有抢占到锁的线程构建成双向Node节点形成了一个链

继续看acquireQueued

// node的模式是EXCLUSIVE，独占锁，因为lock是独占锁
final boolean acquireQueued(final Node node, int arg) {
    boolean failed = true;
    try {
        boolean interrupted = false;
        for (;;) {
        	// predecessor方法获取当前的prev节点
            final Node p = node.predecessor();
            // 因为head节点是一个空节点，仅仅只是为了简单而构建的
            // 所以当node的上一节点是head节点时说明当前node是第一个节点了，那么可以去抢占锁了
            if (p == head && tryAcquire(arg)) {
            	// 抢占锁成功则更新head节点，设置head节点的thread为null
                setHead(node);
                p.next = null; // help GC
                failed = false;
                return interrupted;
            }
            // 清理无效节点，比如CANCELLED取消的节点
            if (shouldParkAfterFailedAcquire(p, node) &&
            	// 并且阻塞当前节点
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

parkAndCheckInterrupt()方法

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

到这里lock的方法就分析完了

看一下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;
}

protected final boolean tryRelease(int releases) {
	// state-1
    int c = getState() - releases;
    if (Thread.currentThread() != getExclusiveOwnerThread())
        throw new IllegalMonitorStateException();
    boolean free = false;
    if (c == 0) {
        free = true;
        setExclusiveOwnerThread(null);
    }
    setState(c);
    // 返回是否完全释放了锁，只有释放releases之后state=0，才说明完全释放了，比如重入锁，进入几次就说明需要释放几次
    return free;
}

看unparkSuccessor(h)

private void unparkSuccessor(Node node) {
    int ws = node.waitStatus;
    if (ws < 0)
        compareAndSetWaitStatus(node, ws, 0);

    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)
    	// 唤醒s节点，注意s节点不是传进来的node节点，而是node.next,即head节点的下一个节点，因为head节点始终是空节点
        LockSupport.unpark(s.thread);
}

当s节点被唤醒后会从之前lock方法里面pack的地方继续往下走，也就是这里

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) &&
            	// 之前是阻塞在这里的，唤醒后又进入了for循环
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

到这里源码就看完了，并不是很复杂，其中使用了一个Sync继承了AQS同步器

总结

大致流程：

1. state=0没有锁，state=1有线程占用了锁，并且存储了当前线程exclusiveOwnerThread，state>1重入锁，进入了多次
2. 每次有线程调用lock，则先cas修改state的值，修改成功或者发现当前线程就是exclusiveOwnerThread则说明重入，那么抢占锁成功，直接返回即可
3. 如果抢占锁失败，则构建一个双向的Node链表，并且节点调用LockSupport.park(this)进入阻塞
4. unLock之后唤醒head节点的下一个节点，唤醒节点再次去抢占锁，抢占成功则设置state=1和exclusiveOwnerThread，然后返回，失败则再次加入链表的尾部并且阻塞

上文使用了AQS同步器，AQS是一个扩展性非常好的同步器，juc下的很多组件都使用了AQS，基于AQS实现一个我们需要的阻塞唤醒的组件是非常简单的