Java并发编程之ReentrantLock源码分析

Posted 夜里码码

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Java并发编程之ReentrantLock源码分析相关的知识,希望对你有一定的参考价值。

ReentrantLock介绍

从JDK1.5之前,我们都是使用synchronized关键字来对代码块加锁,在JDK1.5引入了ReentrantLock锁。synchronized关键字性能比ReentrantLock锁要差,而且ReentrantLock锁功能要比synchronized关键字功能强大。

特点

synchronized关键字和ReentrantLock锁都是重入锁,可重入锁是指当一个线程获取到锁后,此线程还可继续获得这把锁,在此线程释放这把锁前其他线程则不可获得这边锁。相比synchronized关键字,ReentrantLock锁具有锁获取超时和获取锁响应中断的特点。ReentrantLock锁还分公平锁和非公平锁,公平锁模式是按线程调用加锁的先后排队顺序获取锁,非公平锁模式是已经在排队中的线程按顺序获取锁,但是新来的线程会和排队中的线程进行竞争,并不保证先排先获取锁。

ReentrantLock 源码分析

ReentrantLock实现了java.util.concurrent.locks.Lock接口和java.io.Serializable接口,前者是对实现Java锁的一种规范,后者说明ReentrantLock可以序列化。 ReentrantLock定义了一个成员变量


private final Sync sync;

Sync类型是ReentrantLock的内部类,继承至AbstractQueuedSynchronizer ,AbstractQueuedSynchronizer是一个带空头的双向列表,为ReentrantLock的锁排队提供了基础支持。 ReentrantLock的UML关系图如下

技术分享图片

下面我们解析下ReentrantLock中几个常用方法。

lock()方法源码分析

lock()是ReentrantLock中最常用的方法,用来对代码块加锁。lock()先是调用Sync的lock()的方法,Sync#lock()实现分为非公平模式和公平模式,我们对这2个模式分别讲解

非公平模式

Sync#lock()非公平模式代码如下:


        final void lock() {
        //用CAS方法设置枷锁状态
            if (compareAndSetState(0, 1))
                setExclusiveOwnerThread(Thread.currentThread());
            else
        //抢锁失败,进入后续逻辑。
                acquire(1);
        }

新来线程先调用compareAndSetState(0, 1)方法用CAS方法设置加锁状态,这里是非公平模式实现要点,这样做主要是为了新来的线程和排队中的线程竞争,排队中的线程激活后也会用CAS方法设置加锁状态,就是看哪个线程线程抢的快,哪个能拿到锁。如果设置加锁状态成功,则设置AbstractQueuedSynchronizer中的全局变量线程为当前当前线程。如果设置加锁状态失败即抢锁失败,则调用acquire(1)进入排队逻辑。

AbstractQueuedSynchronizer#acquire(int arg)实现代码如下:


    public final void acquire(int arg) {
    //先调用tryAcquire(arg)再试下能不能获取到锁,无法获取则调用acquireQueued(addWaiter(Node.EXCLUSIVE), arg)进入排队
        if (!tryAcquire(arg) &&
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }

先调用tryAcquire(arg)再试下能不能获取到锁,获取成功则执行结束,无法获取则调用acquireQueued(addWaiter(Node.EXCLUSIVE), arg)进入排队,此方法返回参数为是否中断当前线程,排队过程中如果线程被中断则会返回ture,此时调用selfInterrupt()中断当前线程。

tryAcquire(arg)直接调用了非公平模式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;
        }

调用getState()方法获取加锁状态,如果为0表示当前未被加锁,尝试CAS设置加锁状态获取锁,如果成功同样设置AbstractQueuedSynchronizer中的全局变量线程为当前当前线程。如果已被加锁,这判断当前线程和加锁线程是否是同一线程,如果是同一线程则将获取锁的状态加1返回获取锁成功,这里就是可重入锁实现的核心,状态的值表示当前线程重入了多少次,之后的释放锁就要释放相同的次数。

接下来我们看下acquireQueued(addWaiter(Node.EXCLUSIVE), arg)方法,acquireQueued主要功能是对当前线程阻塞,阻塞到能被上个获取到锁线程释放为止,addWaiter(Node.EXCLUSIVE)则是将当前线程加入到排队队列中。 我们先来看下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
        Node pred = tail;
        //CAS快速添加节点到尾部
        if (pred != null) {
            node.prev = pred;
            if (compareAndSetTail(pred, node)) {
                pred.next = node;
                return node;
            }
        }
        //如果尾节点不存在或者添加失败走最大努力添加节点逻辑
        enq(node);
        return node;
    }
    
private Node enq(final Node node) {
        for (;;) {
            Node t = tail;
            //如果头尾节点为空则创建空节点当头尾节点
            if (t == null) { // Must initialize
                if (compareAndSetHead(new Node()))
                    tail = head;
            } else {
            //CAS添加节点到尾部
                node.prev = t;
                if (compareAndSetTail(t, node)) {
                    t.next = node;
                    return t;
                }
            }
        }
    }
    

创建已当前线程为基础的节点,先走快速添加到尾部逻辑,获取尾节点如果尾节点存在,将当前节点和尾节点相连,并用CAS方式将当前节点设置为尾节点,这边使用CAS方式考虑了多个线程同时操作尾节点的情况,所以如果尾节点已经变更则快速添加节点操作失败,调用enq(node)方法走最大努力添加节点的逻辑。enq(node)最大努力添加逻辑就是一直添加节点直到添加节点到尾部成功。

下面看下acquireQueued(addWaiter(Node.EXCLUSIVE), arg)的实现


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里有个循环,这个循环的主要作用就是在线程激活后重试获取锁直到获取锁。node.predecessor()获取当前线程节点的前一个节点,如果是头节点,则当前线程尝试获取锁,获取锁成功设置当前节点为头节点。如果获取失败或者非头节点则调用shouldParkAfterFailedAcquire(p, node)判断是否需要阻塞等待,如果需要阻塞等待则调用parkAndCheckInterrupt()阻塞当前线程并让出cup资源资质被前一个节点激活,继续循环逻辑。

我们先来看下shouldParkAfterFailedAcquire(p, node)的实现


 private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
        int ws = pred.waitStatus;
        if (ws == Node.SIGNAL)
            
            return true;
        if (ws > 0) {
             
            do {
                node.prev = pred = pred.prev;
            } while (pred.waitStatus > 0);
            pred.next = node;
        } else {
           
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        }
        return false;
    }

先获取前个节点的状态,状态分以下4类


static final int CANCELLED =  1;
static final int SIGNAL    = -1;
static final int CONDITION = -2;
static final int PROPAGATE = -3;

除了CANCELLED关闭状态是非正常,其他状态均正常状态。判断当前状态是否是SIGNAL正常状态,如果是就返回成功,这样当前线程就可以阻塞安心的等待上个节点的激活。如果状态为CANCELLED关闭状态则删除所有当前节点之前状态为CANCELLED的节点,返回失败让当前线程重试获取锁,如果是初始化0状态则CAS方式设置状态为SIGNAL。

接下来看下阻塞方法parkAndCheckInterrupt()


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

方法很简单调用LockSupport.park(this)阻塞当前线程,这里要讲下方法返回时调用Thread.interrupted()判断当前线程是否被中断,如果被中断的话,当前线程获取到锁后会调用Thread.currentThread().interrupt()中断线程。

公平模式

公平模式和非公平模式大部分代码相同,主要是获取锁的逻辑不同,我们就讲下代码不同的部分 lock()代码如下


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

非公平模式模式先尝试设置状态来获取锁,而公平模式则直接调用acquire(1)去走排队逻辑。

尝试获取锁的方法tryAcquire(int acquires)也不一样代码如下


protected final boolean tryAcquire(int acquires) {
            final Thread current = Thread.currentThread();
            int c = getState();
            if (c == 0) {
                if (!hasQueuedPredecessors() &&
                    compareAndSetState(0, acquires)) {
                    setExclusiveOwnerThread(current);
                    return true;
                }
            }
            else if (current == getExclusiveOwnerThread()) {
                int nextc = c + acquires;
                if (nextc < 0)
                    throw new Error("Maximum lock count exceeded");
                setState(nextc);
                return true;
            }
            return false;
}

该方法跟非公平锁基本都一样,只是在获取锁的时候加了hasQueuedPredecessors()判断,这个方法主要判断了当前线程是否在头节点的下个节点,这样保证了获取锁的顺序性。

unlock()方法源码分析

unlock()方法比较简单,直接调用sync.release(1)方法。 release(1)代码如下


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

先尝试释放锁,如果释放产品这判断当前节点是否为0不为0调用unparkSuccessor(h)方法激活下个节点的线程,否则直接返回。这里会有个疑问为什么h.waitStatus为0不去激活下个节点的线程,如果不激活下个节点的线程是否一直阻塞的,答案是否定的。这样做主要是为了释放锁的效率。waitStatus为0是初始化的值,这个值还没被下个节点线程调用shouldParkAfterFailedAcquire(p, node)方法设置成SIGNAL状态,也就说明下个节点线程还没被阻塞,此时如果下个节点线程调用此方法并设置成SIGNAL状态,势必它会重新获取锁,从而获取到锁避免了上述的问题。

下面来看下tryRelease(arg)方法


 protected final boolean tryRelease(int releases) {
        //重入次数减1
            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;
        }

因为锁可重入,因此调用getState()获取状态的值并减去一次重入次数,得到的c就是剩余重入的次数,然后判断当前释放的线程是否是当前占有锁的线程,如果不是抛出异常,否则先判断c是否为0表示当前线程持有的锁是否释放完全,如果是则设置持有锁的线程的变量为空,并设置锁状态为0,否则设置剩余的c到锁的状态。

接下来看下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)
            //激活线程
            LockSupport.unpark(s.thread);
    }

获取当前节点状态,设置如果当前节点正常情况则设置成0,然后取当前节点的下个节点,如果下个节点状态非正常即CANCELLED状态,则从队列的尾部开始查找查到最靠近当前的节点且状态正常的节点,然后调用LockSupport.unpark(s.thread)通知此节点停止阻塞。这边会有个疑问如果调用LockSupport.unpark(s.thread)方法后,此节点才调用LockSupport.park(this)去阻塞,这样会不会发生此节点永久阻塞的问题,答案是否定的,LockSupport.unpark(s.thread)方法的实现其实是为线程设置了一个信号量,LockSupport.park(this)就算后调,如果线程相同也会收到此信号从而激活线程,这里的实现原理就不展开讲。

原文链接:https://my.oschina.net/u/945573/blog/2991876

以上是关于Java并发编程之ReentrantLock源码分析的主要内容,如果未能解决你的问题,请参考以下文章

Java并发编程:ReentrantLock-FairSync源码分析(hasQueuedPredecessors)

Java并发编程:ReentrantLock-NonfairSync源码逐行深度分析(下)

Java并发编程:ReentrantLock-NonfairSync源码逐行深度分析(中)

Java Review - 并发编程_独占锁ReentrantLock原理&源码剖析

Java Review - 并发编程_独占锁ReentrantLock原理&源码剖析

并发编程—— ReentrantLock实现原理及源码分析