并发编程之可重入锁ReentrantLock

文章目录

- 前言
- ReentrantLock原理
- ReentrantLock VS Synchronized
- 源码解析
- - ReentrantLock同步机制
  - ReentrantLock可重入机制
  - ReentrantLock可中断机制
  - ReentrantLock超时机制
  - 条件变量Condition
- 写在最后

前言

大家都知道在并发编程中一般会用到多线程技术，多线程技术可以大大增加系统QPS/TPS。但是在一些特殊的业务场景下我们需要限制线程的并发数目，比如秒杀系统、多种商品金额叠加运算等等都是需要限制线程数量。特别是在分布式微服务架构，多线程同步问题尤为明显。一般在实际的开发中我们会使用同步和加锁的方式对线程进行处理，比如synchronized修饰代码块和方法达到同步的目的。但是synchronized默认非公平锁且没有中断和超时功能，所以在实际开发中我们推荐使用可重入锁ReentrantLock。

ReentrantLock原理

可重入锁ReentrantLock内部继承AQS抽象同步队列，并由AQS中的CLH双向FIFO阻塞队列和STATE状态保证锁的同步。ReentrantLock在初始化可以显示指定公平与非公平锁，默认是非公平锁。是否公平锁体现在公平锁在STATE == 0时唤醒CLH阻塞队列头部线程，非公平锁则是CLH队列中的线程与其他线程进行争夺资源。另外ReentrantLock必须显示使用Lock()、unLock()获取与释放锁，并提供中断和超时获取锁等方法，功能和灵活度远远高于synchronized。
在这里插入图片描述

ReentrantLock VS Synchronized

1、用法不同，synchronized修饰方法、静态方法和代码块，ReentrantLock只用用在代码块中；
2、锁性质不同，Synchronized是非公平锁，ReentrantLock默认非公平锁，但可以显示指定公平与非公平锁；
3、获取锁方式不同，Synchronized是底层自动加解锁，ReentrantLock必须用代码加解锁；
4、中断方式不同，Synchronized获取锁不能中断，ReentrantLock可以中断解决死锁问题；
5、实现方式不同，Synchronized 通过JVM监视器Monitor实现，通过增加monitorenter 和 monitorexit指令对代码块同步；ReentrantLock则是通过AQS抽象同步队列的API进行同步。

源码解析

ReentrantLock同步机制

进入package java.util.concurrent.locks 下查看ReentrantLock源码：

//reentrantlock 内部类继承aqs,并显示覆写一些方法
abstract static class Sync extends AbstractQueuedSynchronizer {
    private static final long serialVersionUID = -5179523762034025860L;

    /**
     * Performs {@link Lock#lock}. The main reason for subclassing
     * is to allow fast path for nonfair version.
     */
    abstract void lock();

    /**
     * 非公平尝试获取锁
     */
    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;
    }


    //释放锁
    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;
    }
}

如上源码所示，ReentrantLock 内部类 Sync 继承AQS抽象同步队列，并覆写了尝试获取锁、获取锁、释放锁的方法。从这里可以看出ReentrantLock是通过调用AQS的API实现同步功能。

继续查看源码：

/**
 * 非公平锁同步对象
 */
static final class NonfairSync extends Sync {
    private static final long serialVersionUID = 7316153563782823691L;

    /**
     * Performs lock.  Try immediate barge, backing up to normal
     * acquire on failure.
     */
    final void lock() {
        if (compareAndSetState(0, 1))
            setExclusiveOwnerThread(Thread.currentThread());
        else
            acquire(1);
    }

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

/**
 * 公平锁同步对象
 */
static final class FairSync extends Sync {
    private static final long serialVersionUID = -3000897897090466540L;

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

    /**
     * Fair version of tryAcquire.  Don't grant access unless
     * recursive call or no waiters or is first.
     */
    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;
    }
}

如上源码所示，ReentrantLock有公平锁与非公平锁之分。非公平锁与公平锁都是覆写了tryAcquire()尝试获取锁的方法，并都是调用AQS获取锁acquire()方法。唯一不同的是公平锁在尝试获取锁的时候会调用hasQueuedPredecessors()方法判断线程是否在CLH队列中，非公平则没有这个逻辑直接是 CAS设置 STATE值、标识独占线程和重入验证。

当然ReentrantLock默认是非公平锁，也是可以通过构造方法指定锁类型：

/**
 * Creates an instance of {@code ReentrantLock}.
 * This is equivalent to using {@code ReentrantLock(false)}.
 */
public ReentrantLock() {
    sync = new NonfairSync();
}

/**
 * Creates an instance of {@code ReentrantLock} with the
 * given fairness policy.
 *
 * @param fair {@code true} if this lock should use a fair ordering policy
 */
public ReentrantLock(boolean fair) {
    sync = fair ? new FairSync() : new NonfairSync();
}

ReentrantLock可重入机制

可重入机制就是同一个线程可以多次进入受限同步资源。ReentrantLock当然是可重入的，我们进入源码：

//非公平锁对象尝试获取锁
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;
}

//公平锁对象尝试获取锁
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;
}

如上源码所示无论是用的公平锁还是非公平锁，在STETE == 0时都会用 CAS compareAndSetState(0, acquires)方法改变STATE值，然后调用setExclusiveOwnerThread(current)方法设置独占线程；如果STATE != 0说明资源已经被占用，此时会在getExclusiveOwnerThread()方法会验证是否是独占线程，如果current == getExclusiveOwnerThread() 当前线程就是独占资源的线程，那么该线程直接获取到锁进入资源。

ReentrantLock可中断机制

可中断机制就是线程在阻塞队列中等待过程中，我们可以调用lockInterruptibly() 中断方法打断线程的等待来防止死锁的发生。
我们继续查看中断机制的源码：

//中断方法
public void lockInterruptibly() throws InterruptedException {
    sync.acquireInterruptibly(1);
}
//aqs中断方法
public final void acquireInterruptibly(int arg)
        throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    //尝试获取锁
    if (!tryAcquire(arg))
        //获取不到表示在阻塞队列，则进行中断逻辑
        doAcquireInterruptibly(arg);
}

如上源码所示，当程序主动调用lockInterruptibly()方法是调用的AQS获取中断机制。首先会先尝试获取锁，如果能够获取到锁则不进入中断逻辑，如果没有获取到锁则进入中断方法。

继续查看中断源码：

//aqs中断方法
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);
    }
}

如源码所示，中断机制会自旋获取锁，如果获取到锁则不再执行中断机制；如果没有获取到锁则抛出 InterruptedException() 异常，程序最终如果没有获取到锁会调用 cancelAcquire(node)取消获取锁，从而避免无限等待使程序死锁的发生。

ReentrantLock超时机制

超时机制就是获取锁的方法中传入超时时间，如果在这个时间后还没有获取到锁则取消获取锁并返回false标识。

老规矩查看源码：

//tryLock 带有超时时间的尝试获取锁
public boolean tryLock(long timeout, TimeUnit unit)
        throws InterruptedException {
    return sync.tryAcquireNanos(1, unit.toNanos(timeout));
}

//底层调用aqs获取锁方法
public final boolean tryAcquireNanos(int arg, long nanosTimeout)
        throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    return tryAcquire(arg) ||
        doAcquireNanos(arg, nanosTimeout);
}

如源码所示，tryLock(long timeout, TimeUnit unit)底层是调用AQS tryAcquireNanos(int arg, long nanosTimeout)方法。在调用时会先尝试获取锁，没有获取到则进入超时获取锁逻辑。

继续查看超时获取锁源码：

//AQS超时获取锁源码
private boolean doAcquireNanos(int arg, long nanosTimeout)
        throws InterruptedException {
    if (nanosTimeout <= 0L)
        return false;
    final long deadline = System.nanoTime() + nanosTimeout;
    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 true;
            }
            nanosTimeout = deadline - System.nanoTime();
            if (nanosTimeout <= 0L)
                //超时返回为未获取到锁标识
                return false;
            //未获取到锁且线程应该阻塞且超时时间大于1000ns会睡眠nanosTimeout
            if (shouldParkAfterFailedAcquire(p, node) &&
                nanosTimeout > spinForTimeoutThreshold)
                LockSupport.parkNanos(this, nanosTimeout);
            if (Thread.interrupted())
                throw new InterruptedException();
        }
    } finally {
        if (failed)
            //取消获取锁
            cancelAcquire(node);
    }
}

如源码所示如果 p == head && tryAcquire(arg) 当前线程节点的前驱节点是头部节点且当前线程尝试获取锁成功，会将当前节点设置为头部节点且直接返回获取到锁标识。如果不是头部节点或没有获取到锁则会判断是否获取时间超时，如果超时返回false结束流程，否则shouldParkAfterFailedAcquire(p, node) && nanosTimeout > spinForTimeoutThreshold 节点需要阻塞且超时时间大于1000纳秒则 LockSupport.parkNanos(this, nanosTimeout) 将当前线程睡眠nanosTimeout 纳秒，否则当前线程自旋获取锁。

cancelAcquire(node) 如果没有获取到锁或者超时最终都会直接取消获取锁机制。

条件变量Condition

条件变量就是线程在执行的过程中会可以调用await()阻塞线程、signal()唤醒线程。就相当于休息室，调用await就是让当前线程等待，signal就是让线程继续执行。值得注意的是在await()阻塞线程时候会释放锁，signal()则是唤醒最先调用await()的线程，并且我们可以定义多个condition条件。

查看源码：

//ReentrantLock 中新建一个条件变量
public Condition newCondition() {
    return sync.newCondition();
}

//调用aqs ConditionObject实例化一个条件对象
final ConditionObject newCondition() {
    return new ConditionObject();
}

如源码所示ReentrantLock condition还是用的AQS的ConditionObject。

我们继续分析ConditionObject 中的核心方法await()、signal()方法：

//aqs ConditionObject  await方法
public final void await() throws InterruptedException {
    //线程如果中断则抛出中断异常
    if (Thread.interrupted())
        throw new InterruptedException();
    //将当前节点加入等待队列    
    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);
}

如源码所示，我们在线程执行过程中调用 Condition await()方法会先检查线程是否中断，没有中断会将当前线程加入等待队列，释放同步队列并保存STATE状态。然后会判断是节点是否在同步队列，不在同步队列会执行LockSupport.park(this)方法将线程阻塞。

当然在线程阻塞过程中会自旋验证 (interruptMode = checkInterruptWhileWaiting(node)) != 0 线程是否中断，是否被Condition signal()方法唤醒。

我们继续分析signal()源码：

// condition 唤醒方法
public final void signal() {
    if (!isHeldExclusively())
        throw new IllegalMonitorStateException();
    Node first = firstWaiter;
    if (first != null)
        //唤醒等待队列第一个线程
        doSignal(first);
}

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

如源码所示，signal()唤醒方法默认唤醒条件等待队列中第一个线程。transferForSignal(first)则是将节点从条件队列移动到同步队列。

继续查看移动节点源码transferForSignal：

final boolean transferForSignal(Node node) {
    /*
     * cas 修改 Node.CONDITION == 0，如果修改失败标识节点已经取消
     */
    if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
        return false;

    /*
    将node 移动到同步队列并返回前驱节点
    */
    Node p = enq(node);
    int ws = p.waitStatus;
    //如果前驱节点 waitStatus > 0 或者 cas修改p节点为唤醒状态失败
    if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
        //强制唤醒线程
        LockSupport.unpark(node.thread);
    return true;
}

private Node enq(final Node node) {
    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;
            }
        }
    }
}

如上源码所示，enq(node)方法会将等待队列的第一个节点移动到同步队列，并会返回节点的前驱节点。然后 ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL) 表示节点的前驱节点取消或者cas设置前驱节点waitStatus == SIGNAL失败则执行LockSupport.unpark(node.thread)方法强制唤醒节点。

这里有个很重要的点就是同步队列节点的 waitStatus等待状态，如果是SIGNAL，则表示后续节点是一个阻塞节点需要唤醒的节点，当当前节点执行完成后会主动唤醒后续节点。这里也表明了conditon signal()方法唤醒线程并不会立即执行，而是要当前线程执行完成才会执行唤醒的线程。

写在最后

今天的博文我们讲述了ReentrantLock与Synchronized的区别联系，也解析了ReentrantLock可重入、可中断、超时机制以及conditon变量的源码。通过源码我们可以知道ReentrantLock 是通过AQS保证同步，其他的中断机制、超时机制、以及condition变量也都是基于AQS中的相应底层代码。在实际的开发过程中，我们可以根据自身业务场景选择应用ReentrantLock方法，从而增加系统的健壮性和可维护性。