ReentrantLock原理浅析

妖狐艹你老母 2022-05-14 09:48 276阅读 0赞

# 一、序言 #

最近一直在看高洪岩著的《java并发编程》，里面介绍了java.util.cocurrent下的常用并发相关类，但是基本上都是介绍如何用这些类，以及这些类的作用，并没有细究到相关原理，加上最近的工作内容很多涉及到异步多线程，用到了一些并发控制相关的类，如ReentrantLock、CountDownLatch等等，所以最近一直在看源码研究这些类的原理，而这篇文章就是从源码的角度写我对ReentrantLock各个特性的原理的理解和总结。

# 二、介绍 #

简单介绍下，ReentrantLock是java的线程安全中对锁进行控制的相关类，类似的还有synchronize关键字（它们有很多区别，这篇文章不打算细究ReentrantLock和synchronize关键字的区别，需要了解的读者可以自行在网上搜索相关资料）。ReentrantLock是在代码层面对锁进行控制，正如其名一样，它是一个可重入锁，并且还可以选择是否使用公平锁（默认为不公平锁）。

# 三、简单例子 #

public class Test {
    
        public static void main(String[] args) throws InterruptedException {
            Lock lock = new ReentrantLock();
            ExecutorService executor = Executors.newFixedThreadPool(2);
            executor.execute(()->{
                try {
                    print(lock);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
            executor.execute(()->{
                try {
                    print(lock);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
    
    
        }
    
        private static void print(Lock lock) throws InterruptedException {
            lock.lock();
            System.out.println("获取锁后打印...");
            Thread.sleep(5_000);
            lock.unlock();
        }
    }

运行的结果：打印“获取锁后打印...”，5秒后再次打印“获取锁后打印...”

# 四、特性 #

## 1.继承关系 ##

ReentrantLock实现了Lock接口，Lock接口有最基本的获取锁或释放锁的基本操作，而ReentrantLock自身是不对锁进行直接的获取和释放，而是通过ReentrantLock内部的一个代理类对锁的状态进行操作。这两个代理类为FairSync和NonfairSync,而它们俩均是继承自一个称为同步器的类AbstractQueuedSynchrnizer（这个同步器很重要，java并发包下的很多类的底层都是通过它来实现）

## 2.非公平锁NonfairSync的lock ##

非公平锁是ReentrantLock实例化时默认使用的锁，其是通过抢占的方式获取锁，大致过程为：线程一开始就尝试获取锁，如果成功则获取到锁，如果不成功最加入到一个等待队列里，在队列里根据FIFO进行锁的获取。

### 2.1NonfairSync的源码浅析 ###

在ReentrantLock的构造方法中可看出在实例化时默认使用非公平锁

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

但是在实例化时也可以选择公平锁

public ReentrantLock(boolean fair) {
        sync = fair ? new FairSync() : new NonfairSync();
    }

NofairSync的实现代码如果

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);
            }
        }

NofairSync在lock()时候尝试通过compareAndSetState去获取锁（即改变state的值，state是什么下文会介绍到），如果当前没有其他线程竞争锁，则获取成功并通过setExclusiveOwnerThread将当前自身线程记录起来，否则则执行acquire方法，传入的参数1和state有关

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

之所以执行到acquire则说明当前的锁发生了竞争，当前线程在acquire方法中通过tryAcquire去试图抢占锁，如果抢占失败，则通过addWaiter将当前节点加入到等待队列的尾部。addWaiter如下

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;
            if (pred != null) {
                node.prev = pred;
                if (compareAndSetTail(pred, node)) {
                    pred.next = node;
                    return node;
                }
            }
            enq(node);
            return node;
        }

Node即为节点，实例化时传入当前线程，以及mode（其实就是waitStatus类型，下文会讲到），添加到队列尾时会通过CAS的方式添加，如果添加失败，则会通过enq自旋的方式添加直到添加成功。

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;
                    }
                }
            }
        }

回退再次回到acquire方法

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

将节点添加到队列尾部后，会通过acquireQueued尝试再次获取锁或者开始阻塞

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);
            }
        }

如果当前节点的前继节点是头结点，那么当前节点对应的会再次尝试获取锁，如果获取成功则直接反馈，反之失败的话，则通过shouldParkAfterFailedAcquire判断当前线程是否需要在获取锁失败后阻塞

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
            int ws = pred.waitStatus;
            if (ws == Node.SIGNAL)
                /*
                 * This node has already set status asking a release
                 * to signal it, so it can safely park.
                 */
                return true;
            if (ws > 0) {
                /*
                 * Predecessor was cancelled. Skip over predecessors and
                 * indicate retry.
                 */
                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.
                 */
                compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
            }
            return false;
        }

如果当前节点的前继节点的waitStatus为SIGNAL(waitStatus不同的状态对应不同的行为是在事前就定义说明的)，说明当前节点需要阻塞等待前继节点的触发，如果前继节点waitStatus大于0说明前继节点需要取消。

如果线程判断需要被阻塞，则调用parkAndCheckInterrupt

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

AbstratQueuedSynchronizer的线程阻塞是通过LockSupport实现，park为阻塞，unpark为唤醒。因为再阻塞期间有可能会被其他线程终中断，所以通过Thread.interrupted检查当前线程是否被中断。之所以返回中断标志位个人的理解是为了让实现AbstratQueuedSynchronizer的子类知道当前线程是被LockSupport.unpark唤醒的还是被其他线程中断的。

### 2.2NonfairSync小结 ###

1、非公平锁获取锁的大致过程：检查当前锁是否被其他线程占用（如果没有被其他线程占用，当前步骤也有可能获取锁失败，因为CAS操作有可能会失败），如果没被占用则获取锁，如果被占用了，则尝试抢占锁，抢占成功则获取锁，抢占失败则将该线程节点添加到等待队列尾部，再次判断如果当前节点的前继节点是头结点，则再次尝试抢占锁，如果抢占成功则获取锁，如果抢占失败则通过waitStatus判断是否需要阻塞，直到前继节点的唤醒，整个过程最多有3次尝试获取锁的操作。

2、公平锁和非公平的获取锁的操作区别就在于tryAcquire()，其也是可重入锁实现的重要部分，其在讲解公平锁部分会详细说明。

## 3.公平锁FairSync的lock ##

公平锁和非公平锁除了lock部分只有tryAcquire()不一样，所以这一节重点将公平锁和非公平所的tryAcquire()的区别

### 3.1公平锁的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;
            }

如果当前state为0，说明锁没有线程占用，再判断等待队列是否有线程正在等待，如果有则返回false，如果没有则直接获取锁，但是如果当前的statue不为0，则判断当前获取锁的线程是否是线程本身，如果是，则增加state的值后返回true。这里也就是可重入锁的实现原理，获取锁多少次（state的值加了多少次），释放锁的时候就要释放多少次（state的值减了多少次），相信到这里，应该大致了解state的作用了。

### 3.2非公平锁的tryAcquire() ###

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;
            }

非公平锁的tryRequire()具体实现是在Sync里，从上面的代码可以很清除的看到，非公平锁的tryRequire()区别与公平锁的的tryRequire()在于非公平锁在获取锁时没有hasQueuedPredecessors方法，该方法是用于判断等待队列是否有线程正在等待，在上面代码也就是说明了非公平锁的线程不用去判断是否有等待队列就就行锁的抢占。

### 3.3hasQueuedPredecessors()判断队列是否有等待的线程 ###

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());
        }

h.next为空有以下两种情况：

1、Node h = head之后，return之前，刚好当前等待的第一个节点作为head获取到锁，而原来的head.next被赋值为null（该操作在方法acquireQueued）,这个时候说明锁被占用，队列里有节点。

2、线程通过enq自旋将节点加入到队列后（tail = node），还没来得及将自身赋值给head头结点的next，这个时候也说明锁已经被占用了，即队列里有节点了。

## 4.unlock ##

### 4.1 unlock代码浅析 ###

公平锁和非公平锁的unlock都是一样的，其代码如下

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

其是调用AbstractQueuedSynchronizer的release,如下:

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

tryRelease尝试释放锁，tryRelease是由AbstractQueuedSynchronizer的子类即ReentrantLock重写实现，具体如下

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;
            }

而唤醒后继节点是通过unparkSuccessor实现，具体如下

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;
            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);
        }

### 4.2 unlock小结 ###

unlock相对与lock简单，其是通过判断state是否为0后（即重出次数是否够了）再对后继节点通过LockSupport.unpark唤醒，就算没有后继节点可唤醒，head阶段字段也不为空，但是其waitStatus的值被置为0。

## 5.state ##

state的值在线程初次获取到锁时，其值由0加至1，如果相同的线程通过n次重入，则state就加了n次1，其值也为n，那么释放锁也需要重出n次才能真正释放（真正释放了锁后state值为0）。

## 6.waitStatus ##

waitStatus是类Node中定义的字段，其值的含义如下

1.CANCELLED=1:该Node会被取消

2.SIGNAL=-1:该Node的后继Node为阻塞状态，需要等待触发后获取锁

3.CONDITION=-2:该Node需要一定的条件才能够被唤醒

4.PROPAGATE=-3:该节点用于AbstractQueuedSynchronizer的共享锁的传播唤醒，如在CountDownLatch中会用到