Java源码阅读之ReentrantLock - lock和unLock方法-阿里云开发者社区

开发者社区> 开发与运维> 正文

Java源码阅读之ReentrantLock - lock和unLock方法

简介: 阅读优秀的源码是提升编程技巧的重要手段之一。 如有不对的地方,欢迎指正 转载请注明出处https://blog.lzoro.com。 碎碎念 如果需要使用或者了解ReentrantLock,证明已经步入并发编程领域了,这里理论基础不多提,需要的自行查阅资料。

阅读优秀的源码是提升编程技巧的重要手段之一。
如有不对的地方,欢迎指正
转载请注明出处https://blog.lzoro.com

碎碎念

如果需要使用或者了解ReentrantLock,证明已经步入并发编程领域了,这里理论基础不多提,需要的自行查阅资料。

但是,相关术语还是要做一下描述的。

ReentrantLock:可重入锁

AQS:AbstractQueuedSynchronized 抽象类,队列式同步器

CAS:Compare and Swap, 比较并交换值

CLH队列:The wait queue is a variant of a "CLH" (Craig, Landin, and
     * Hagersten) lock queue.

ReentrantLock

首先,贴图大家感受一下。

ReentrantLock
Sync

其中SyncReentrantLock的抽象静态内部类,提供了锁的同步措施,具体实现有NonFairSyncFairSync,分别为公平和非公平锁。

从图中我们可以看出,ReentrantLock是实现了Lock接口和Serializable接口,Serializable是Java的序列化接口,这里我们不多做讨论。

那么,开始源码的阅读了~
首先,先看下Lock接口提供的方法(篇幅所限,这里将源码注释去掉),大致可分为三类:获取锁、释放锁、新建条件(可用于高级应用,如等待/唤醒)。

public interface Lock {

    /**
     * 获取锁,若获取失败则进行等待
     */
    void lock();
    
    /**
     * 可中断锁
     */
    void lockInterruptibly() throws InterruptedException;
    
    /**
     * 获取锁,立即返回,成功返回true,否则false
     */
    boolean tryLock();
    
    /**
     * 获取锁,若获取失败则在指定时间内等待,成功返回true,否则false
     */
    boolean tryLock(long time, TimeUnit unit) throws InterruptedException;
    
    /**
     * 释放锁
     */
    void unlock();
    
    /**
     * 新建条件,可用与高级应用
     */
    Condition newCondition();
}

接下来我们具体看下ReentrantLock的实现。

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

可以看到ReentrantLock的lock方法,是调用静态内部类sysc的lock方法的,而synclock方法是抽象方法,具体的实现有两个,NonfairSync(非公平锁)和FairSync(公平锁),我们先来看NonFairSync的实现

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

compareAndSetState(0,1)这个方法是由sysn的父类AbstractQueuedSynchronizer来实现的,也是我们通常说的AQS,而compareAndSetState方法的具体实现是由Unsafe提供的。
Unfafe类的compareAndSwap*系列方法,是虚拟机的本地方法实现,具体的实现不在我们的讨论范围内,简单介绍一下作用,该方法的的作用如下:调用该方法时,若value值与expect值相等,则将value修改为update值,并返回true;若value值与expect值不相等,那么不做任何操作,并返回false,这也就是我们常说的CAS操作,至于存在的ABA等问题和解决方案,有兴趣的可以自己搜索资料。

protected final boolean compareAndSetState(int expect, int update) {
    // See below for intrinsics setup to support this
    return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}

lock方法执行完CAS操作后

若得的一个true返回,则会执行setExclusiveOwnerThread(Thread.currentThread());,该方法作用是为锁设置独占线程,其实也就是一个赋值操作,如下:

  protected final void setExclusiveOwnerThread(Thread thread) {
        exclusiveOwnerThread = thread;
    }

CAS操作返回一个false,则执行acquire方法

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

从上面源码可以看出,若tryAcquire失败并且acquireQueued返回true中断标识的话,将会中断当前线程。
我们先看一下tryAcquire,方法的作用大致如下:判断锁的state值,若当前未有其他线程持有该锁,则执行CAS操作,成功后则设置独占线程;若发现该锁已被线程持有,则判断持有线程是不是当前线程,若是则允许重入,并判断重入的次数是否超过限制,重入成功后修改state并返回一个true布尔值。

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");
        setState(nextc);
        return true;
    }
    return false;
}
        

接下来看一下acquireQueuedaddWaiter方法,作用描述如下,利用addWaiter方法,将当前线程作为一个节点Node加入的CLH队列(The wait queue is a variant of a "CLH" (Craig, Landin, and
* Hagersten) lock queue),这里加入队列有一系列操作,包括尾部判断,前置节点设置等,成功加入队列后,会判断是否有资格去竞争获取锁,有则尝试获取锁,成功后会返回标志位。如果没有资格,则判断是否可以被阻塞,并做相关操作,具体请看注释。

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,前首节点出队,帮助GC 
                failed = false;
                return interrupted;
            }
            //判断是否可以阻塞线程并做相应操作,下面具体阅读这几个方法
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        //判断是否获取失败
        if (failed)
            cancelAcquire(node);
    }
}


private Node addWaiter(Node mode) {
    //封装成node
    Node node = new Node(Thread.currentThread(), mode);
    // 获取队列尾节点(作为当前节点的前置节点)
    Node pred = tail;
    //如果尾节点不为空
    if (pred != null) {
        //设置当前节点的前置节点
        node.prev = pred;
        //CAS操作,设置队列尾部
        if (compareAndSetTail(pred, node)) {
            pred.next = node;
            return node;
        }
    }
    //尾节点为null,调用enq方法
    enq(node);
    //返回当前节点
    return node;
}

 private Node enq(final Node node) {
    
    for (;;) {
        Node t = tail;
        //尾节点为null
        if (t == null) { // Must initialize
            //通过CAS操作设置首节点
            if (compareAndSetHead(new Node()))
                //将首节点赋值给尾节点(初始化)
                tail = head;
        } else {
            //设置当前节点的前置节点
            node.prev = t;
            //通过CAS操作设置尾节点
            if (compareAndSetTail(t, node)) {
                t.next = node;
                return t;
            }
        }
    }
}

接下来是shouldParkAfterFailedAcquireparkAndCheckInterrupt方法

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    //获取前置节点的等待状态
    int ws = pred.waitStatus;
    //如果为SIGNAL
    if (ws == Node.SIGNAL)
        /*
         * This node has already set status asking a release
         * to signal it, so it can safely park.
         */
        //只有当前置节点的状态位SIGNAL的话,当前节点才能进入阻塞,并等待前置节点的唤醒
        return true;
    if (ws > 0) {
        /*
         * Predecessor was cancelled. Skip over predecessors and
         * indicate retry.
         */
        //如果前置节点为取消状态,则不断往前搜索并找到SIGNAL状态的节点,并加在其后面
        do {
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        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.
         */
        //通过CAS操作设置前置节点的等待状态位SIGNAL
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}

/**
 * 如果上面的方法调用返回true,则代表当前节点可以进入阻塞/等待
 */
private final boolean parkAndCheckInterrupt() {
    //通过LockSupport类的park方法来阻塞当前线程
    LockSupport.park(this);
    //被唤醒后,返回中断标志
    return Thread.interrupted();
}  

/**
 * 这里的阻塞具体实现是JVM虚拟机的本地实现,有兴趣者可以自行研究
 */
public static void park(Object blocker) {
    Thread t = Thread.currentThread();
    setBlocker(t, blocker);
    UNSAFE.park(false, 0L);
    setBlocker(t, null);
}
    

看到这里,会有点困惑,如果没有成功获取到锁的线程进入了阻塞状态,那么它什么时候被唤醒呢?
这里有一个不得不提的点,如果使用lock方法来进行加锁,那么必须成对地使用unlock来释放锁,否则容易导致死锁,一般都是在try-catch-finally进行锁的释放。

所以,等待线程的被唤醒是由持有锁的线程调用unlock后触发的。

接下来,从unlock入手来具体看下源码,可以看到unlock方法是调用sync.release(1)实现的,还是以开头的NonFairSync(非公平锁)的实现来看,

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

② 释放锁
public final boolean release(int arg) {
    //判断是否释放成功
    if (tryRelease(arg)) {
        Node h = head;
        //判断CLH队列的首节点是否为null,并判断等待状态是否正常
        if (h != null && h.waitStatus != 0)
            //唤醒节点
            unparkSuccessor(h);
        return true;
    }
    return false;
}

③ 释放锁,并唤醒CLH队列中的合法首节点
protected final boolean tryRelease(int releases) {
    //计算state和释放数量的差值
    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;
}

④ 唤醒阻塞/等待的节点
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;
    //后置节点为null或者为取消状态
    if (s == null || s.waitStatus > 0) {
        s = null;
        //从尾部向前获取到一个不为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);
}
    

被唤醒后的节点,返回是否被中断过的标志,在acquireQueued方法内继续执行循环获取锁的流程。

到这里,NonfairSync非公平锁的分析基本上就告一段落了,而关于FairSync的公平机制,有兴趣的可以去阅读下,实现的机制大同小异。

以上,就是Java可重入锁ReentrantLock的lock和unLock源码分析,膜拜Java源码大神。

总结

1、Lock提供locklockInterruptiblytryLock()tryLock(long time, TimeUnit unit)unlocknewCondition五个方法;

2、lockunlock必须成对调用;

3、ReentrantLock实现了Lock和Serializable两个接口;

4、Sync是ReentrantLock的静态内部类,提供了公平锁(FairSync)和非公平锁(NonFairSync)的实现。

5、CAS操作是基于JVM提供的本地方法实现。

6、待补充

卖萌

喜欢的不妨给个赞呗,溜了溜了。

版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

分享:
开发与运维
使用钉钉扫一扫加入圈子
+ 订阅

集结各类场景实战经验,助你开发运维畅行无忧

其他文章