ReentrantLock源码分析

一、引言

ReentrantLock作为concurrent包一下的一员，有着比Synchronized更加直观灵活的使用方式；

private static void lockTest() {
        try {
            lock.lock();
            //dosometing
        } catch (Exception e) {

        } finally {
            //unlock一定要放在finally语句中
            lock.unlock();
        }
    }

ReentrantLock是通过CAS+AQS队列实现（后面会分析），具有以下特性：

1、可重入锁

2、可响应中断

3、可尝试加锁（tryLock）

4、限时等待尝试加锁（tryLock(long timeout, TimeUnit unit)）

5、公平锁、非公平锁（通过构造函数的boolean指定）

ReentrantLock与Synchronized的异同：

不同：

1、synchronized是Java的关键字，使用较为方便；ReentrantLock是类，使用起来需要创建调用方法等

2、ReentrantLock使用较为灵活（可指定是否为公平锁、可尝试加锁等），Synchronized默认只能是非公平锁，没有其它丰富的功能

3、Synchronized在jdk1.6优化之前的性能是远远不如ReentrantLock的，但是优化后（可根据使用场景变为无锁、偏向锁、轻量级锁、重量级锁）；但Synchronized锁的粒度灵活度还是不如ReentrantLock

相同：

1、ReentrantLock与synchronized都是阻塞式锁

2、两者都是可重入锁

常用方法：

1、lock 加锁

2、unLock 解锁

3、tryLock 尝试加锁，成功返回true，失败返回false，不会等待

4、 tryLock(long timeout,TimeUnit unit) 与tryLock大致相同，尝试获取锁，如果超过这段时间依然没有获取锁，返回false

5、 lockInterruptibly 获取锁，跟lock不一样的地方是，过程中会检测是否中断（interrupt），若是会抛出异常

6、 构造函数 public ReentrantLock(boolean fair) 可设置公平锁还是非公平锁

二、源码分析

1、构造函数

    ReentrantLock lock = new ReentrantLock();//里面的参数可指定是公平还是非公平锁

    public ReentrantLock() {
        sync = new NonfairSync();
    }
    //指定是当前是公平锁 还是非公平锁
    public ReentrantLock(boolean fair) {
        sync = fair ? new FairSync() : new NonfairSync();
    }

sync继承自AbstractQueuedSynchronizer（即传说中的AQS）

sync.png

2、AQS数据结构分析

AQS（java.util.concurrent.locks.AbstractQueuedSynchronizer）是ReentrantLock实现线程排队等待的数据结构，本质是一个双向的FIFO队列；结构如下：

AQS结构.png

Node提供了如下常量：

 /** Marker to indicate a node is waiting in shared mode */
        static final Node SHARED = new Node();//共享模式，表示可以多个线程同时执行
        /** Marker to indicate a node is waiting in exclusive mode */
        static final Node EXCLUSIVE = null;//独占模式，表示只有一个线程能执行例如ReentrantLock；共享模式 多个线程可同时执行 例如Semaphore/CountDownLatch

        /** 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;//说明这个结点因为被某一个condition挂起了
        /**
         * waitStatus value to indicate the next acquireShared should
         * unconditionally propagate.
         */
        static final int PROPAGATE = -3;//在共享模式下，下一次获取锁后可以无限传播

还有以下的变量：

waitStatus 当前结点的状态，默认是0，可以是CANCELL、SIGNAL 等
prev 前继结点
next 后继结点
thread 对应的线程
nextWaiter 下一个等待condition的结点
state 状态；是一个int值，表示当前线程占用资源的数量；0表示空闲，没有线程占用；ReentrantLock的state表示线程重入锁的次数

3、lock

先来看非公平锁的情况下逻辑

3.1、非公平锁的lock

    public void lock() {
        //当为非公平锁时 这里的lock会走NonfairSync下的lock函数
        sync.lock();
    }
    
    final void lock() {
            //利用CAS更新当前锁的状态 分析1
            if (compareAndSetState(0, 1))
                //设置当前线程为独占锁的拥有者 分析2
                setExclusiveOwnerThread(Thread.currentThread());
            else
                //如果CAS更新失败 获取锁  分析3
                acquire(1);
    }
    
    ===>分析1
    //AbstractQueuedSynchronizer下方法；通过CAS更新当前锁的状态
    protected final boolean compareAndSetState(int expect, int update) {
        return U.compareAndSwapInt(this, STATE, expect, update);
    }
    
    ===>分析2
    //如果CAS更新成功，设置当前线程为独占锁的拥有者
    //AbstractOwnableSynchronizer中方法；
    //AbstractOwnableSynchronizer把AQS包了一层；主要提供了一些获取、设置当前独占锁拥有者
    protected final void setExclusiveOwnerThread(Thread thread) {
        exclusiveOwnerThread = thread;
    }
    
    ===>分析3
    //如果CAS更新失败 尝试获取锁
    //AQS中的方法
    public final void acquire(int arg) {
        //如果获取锁失败 阻塞排队 
        if (!tryAcquire(arg) &&
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            //如果排队也失败了 中断自己
            selfInterrupt();
    }
    //tryAcquire是AQS中的方法，由子类具体实现；这里看NonfairSync的tryAcquire
    protected final boolean tryAcquire(int acquires) {
            return nonfairTryAcquire(acquires);
    }
    //acquires 是1
    final boolean nonfairTryAcquire(int acquires) {
            final Thread current = Thread.currentThread();
            int c = getState();//获取state值
            if (c == 0) {
                //如果当前没有线程占用锁 通过CAS更新state标志位
                //(这里也是非公平锁的一个原因 不判断前面是否有其它节点 直接CAS尝试拿锁)
                if (compareAndSetState(0, acquires)) {
                    //将当前线程设置为独占锁的拥有者
                    setExclusiveOwnerThread(current);
                    return true;
                }
            }
            //如果state不为0 并且当前线程就是独占锁的拥有者（表示重入）
            else if (current == getExclusiveOwnerThread()) {
                //累加state
                int nextc = c + acquires;
                if (nextc < 0) // overflow
                    throw new Error("Maximum lock count exceeded");
                setState(nextc);
                return true;
            }
            return false;
    }
    //新节点入队方法
    private Node addWaiter(Node mode) {
        Node node = new Node(mode);

        for (;;) {//通过阻塞的方式将当前的node入队
            Node oldTail = tail;//获取尾结点
            if (oldTail != null) {
                //将新节点插入到队列尾部
                U.putObject(node, Node.PREV, oldTail);
                if (compareAndSetTail(oldTail, node)) {
                    oldTail.next = node;
                    return node;
                }
            } else {
                //队列为空 初始化队列
                initializeSyncQueue();
            }
        }
    }
    
    
    //将当前线程入队阻塞
    final boolean acquireQueued(final Node node, int arg) {
        try {
            boolean interrupted = false;
            for (;;) {
                //获取node的前节点
                final Node p = node.predecessor();
                //如果p的前节点就是头结点 尝试获取锁
                if (p == head && tryAcquire(arg)) {
                    //设置当前node为头结点
                    setHead(node);
                    p.next = null; // help GC
                    return interrupted;
                }
                //做一些阻塞前的准备操作
                if (shouldParkAfterFailedAcquire(p, node) &&
                    //阻塞 唤醒后检查中断标志位
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } catch (Throwable t) {
            cancelAcquire(node);
            throw t;
        }
    }

3.2、公平锁的lock

    final void lock() {
            acquire(1);
    }
    
    //acquire的实现与非公平锁相同
    //看一下公平锁的tryAcquire
    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;
    }
    
    //判断头结点后的节点的线程是否为当前线程
    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;
        return h != t &&
            ((s = h.next) == null || s.thread != Thread.currentThread());
    }

3.3、lock的流程总结

调用lock方法，当为非公平锁时，只要有线程过来就尝试获取锁，如果获取成功（AQS的state==0 并且CAS写入成功）将自身设置为独占锁的拥有者 或者state != 0但当前独占锁就是自身（表示重入），获取锁成功并将state累加；如果获取失败就将自己设置到AQS队列的尾部，等待唤醒；当为公平锁时，与非公平锁的主要区别在于公平锁在执行tryAcquire时，需要加一个判断（当前节点是否还有其他节点），如果有其它节点则将自身添加到队列中等待唤醒；具体lock的流程图如下：

ReentrantLock lock流程.png

4、tryLock

直接尝试获取，获取不到返回false，不会阻塞线程

public boolean tryLock() {
        //直接尝试获取
        return sync.nonfairTryAcquire(1);
    }

5、tryLock(long timeout, TimeUnit unit)

直接尝试获取锁，如果失败则在阻塞时间内不断尝试获取锁

public boolean tryLock(long timeout, TimeUnit unit)
            throws InterruptedException {
        return sync.tryAcquireNanos(1, unit.toNanos(timeout));
    }
 public final boolean tryAcquireNanos(int arg, long nanosTimeout)
            throws InterruptedException {
        if (Thread.interrupted())
            //判断是否中断了
            throw new InterruptedException();
        return tryAcquire(arg) ||   //尝试获取锁；否则进入doAcquireNanos
            doAcquireNanos(arg, nanosTimeout);
    }
 private boolean doAcquireNanos(int arg, long nanosTimeout)
            throws InterruptedException {
        if (nanosTimeout <= 0L)
            //如果时间小于等于0 直接返回失败
            return false;
        final long deadline = System.nanoTime() + nanosTimeout;//算出超时的时刻
        final Node node = addWaiter(Node.EXCLUSIVE);//将当前线程节点加入FIFO队列
        try {
            for (;;) {
                final Node p = node.predecessor();//获取当前节点的前继节点
                if (p == head && tryAcquire(arg)) {
                    //如果前继结点是头结点 尝试获取锁 如果成功的话将自身置为头结点
                    setHead(node);
                    p.next = null; // help GC
                    return true;
                }
                nanosTimeout = deadline - System.nanoTime();
                if (nanosTimeout <= 0L) {
                    //超出指定
                    cancelAcquire(node);
                    return false;
                }
                if (shouldParkAfterFailedAcquire(p, node) &&
                    nanosTimeout > SPIN_FOR_TIMEOUT_THRESHOLD)
                    //时间差过大要超过自旋的限定值了 直接挂起线程 nanosTimeOut时间后再唤醒
                    LockSupport.parkNanos(this, nanosTimeout);
                if (Thread.interrupted())
                    throw new InterruptedException();
            }
        } catch (Throwable t) {
            cancelAcquire(node);
            throw t;
        }
    }

6、unLock

解锁并唤醒队列中下一个不为null的节点

    public void unlock() {
        sync.release(1);
    }
    public final boolean release(int arg) {
        if (tryRelease(arg)) {
            Node h = head;//当前执行的节点，当前的head结点指向的是在执行的线程，
            //如果后面还有其他结点需要唤醒，此时的head的status应该会是SINGAL，会继续走到unparkSuccessor()
            if (h != null && h.waitStatus != 0)
                unparkSuccessor(h);
            return true;
        }
        return false;
    }
    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;
    }
     private void unparkSuccessor(Node node) {
     
        int ws = node.waitStatus;
        if (ws < 0)
            node.compareAndSetWaitStatus(ws, 0);
        Node s = node.next;
        //s.waitStatus > 0说明节点为Cancel节点
        if (s == null || s.waitStatus > 0) {
            s = null;
            for (Node p = tail; p != node && p != null; p = p.prev)
                //从后往前遍历节点 找出不为null的节点
                if (p.waitStatus <= 0)
                    s = p;
        }
        if (s != null)
            //唤醒下一个线程
            LockSupport.unpark(s.thread);
    }
    

7、ReentrantLock的唤醒通知机制

同Object的wait notify类似，ReentrantLock也有一套自己的唤醒等待机制；通过ReentrantLock的newCondition创建condition对象，调用condition的await实现等待、调用signal或者signalAll实现唤醒，例如下面的一个生产者消费者例子：

public class ProducterAndConsumerTest {
    private static Queue<Integer> queue = new ArrayDeque<>();
    private static ReentrantLock lock = new ReentrantLock();
    private static Condition condition = lock.newCondition();

    public static void main(String[] args) {
        new Thread(new ProducterRunnable()).start();
        new Thread(new ConsumerRunnable()).start();
    }

    private static class ProducterRunnable implements Runnable {
        @Override
        public void run() {
            try {
                lock.lock();
                while (true) {
                    if (queue.size() >= 10) {
                        //当生产者队列已满 等待 并释放锁
                        System.out.println("生产者车间已满");
                        condition.await();
                    }
                    queue.offer(1);
                    condition.signal();
                    System.out.println("生产者生产了一个苹果");
                    Thread.sleep(1000);
                }
            } catch (Exception e) {

            } finally {
                lock.unlock();
            }
        }
    }

    private static class ConsumerRunnable implements Runnable {
        @Override
        public void run() {
            try {
                lock.lock();
                while (true) {
                    if (queue.size() <= 0) {
                        //当队列为空 将消费者等待 并释放锁
                        System.out.println("消费者都吃完了");
                        condition.await();
                    }
                    System.out.println("消费者吃了一个苹果");
                    condition.signal();
                    queue.poll();
                    Thread.sleep(1000);
                }
            } catch (Exception e) {

            } finally {
                lock.unlock();
            }
        }
    }
}