并发编程——ReentrantLock

た入场券 2023-10-16 11:42 24阅读 0赞

#### 文章目录 ####

*  ReentrantLock介绍
 *  synchronized和lock区别
 *  ReentrantLock 源码
 *   *  ReentrantLock的lock方法
     *  ReentrantLock的acquire方法
     *  ReentrantLock的tryAcquire方法
     *  ReentrantLock的addWaiter方法
     *  ReentrantLock的acquireQueued方法
     *  ReentrantLock的unlock方法

## ReentrantLock介绍 ##

Java中提供锁，一般就是synchronized和lock锁，ReentrantLock跟synchronized一样都是互斥锁。如果竞争比较激烈，推荐lock锁，效率更高。如果几乎没有竞争，推荐synchronized。

## synchronized和lock区别 ##

lock锁的使用相对synchronized成本更高，synchronized是非公平锁，lock是公平+非公平锁。

lock锁提供的功能更完善，lock可以使用tryLock指定等待锁的时间。

lock锁还提供了lockInterruptibly允许线程在获取锁的期间被中断。

synchronized基于对象实现，lock锁基于AQS+CAS实现。

public static void main(String[] args) {
        
        ReentrantLock lock = new ReentrantLock();
        lock.lock();
        try{
        
            // 业务代码
        }finally{
        
            lock.unlock();
        }
    }

## ReentrantLock 源码 ##

### ReentrantLock的lock方法 ###

lock方法是如何实现让当前线程获取到锁资源的？

先尝试将state从0修改为1，如果成功，代表获取锁资源。如果没有成功，调用acquire。

state是AQS中的一个由volatile修饰的int类型变量，多个线程会通过CAS的方式修改state，在并发情况下，只会有一个线程成功的修改state（从0~1）。

如果线程没有拿到锁资源，会到AQS的双向链表中排队等待

AQS中的双向链表是基于内部类Node在维护，Node中包含prev，next，thread属性，并且在AQS中还有两个属性，分别是head，tail。

AQS的核心

![image.png][]

公平&非公平的方法源码

// 公平锁的sync的lock方法
    final void lock() {
        
        acquire(1);
    }
    
    // 非公平锁的sync的lock方法
    final void lock() {
        
        if (compareAndSetState(0, 1))
            setExclusiveOwnerThread(Thread.currentThread());
        else
            acquire(1);
    }

### ReentrantLock的acquire方法 ###

acquire是一个业务方法，里面并没有实际的业务处理，都是在调用其他方法

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

1.  调用tryAcquire方法：尝试获取锁资源（非公平、公平），拿到锁资源，返回true，直接结束方法。
2.  当没有获取锁资源后，会先调用addWaiter：会将没有获取到锁资源的线程封装为Node对象，并且插入到AQS的队列的末尾，并且作为tail。
3.  调用acquireQueued方法，查看当前排队的Node是否在队列的前面，如果在前面（head的next），尝试获取锁资源。如果没在前面，尝试将线程挂起，阻塞起来。

### ReentrantLock的tryAcquire方法 ###

tryAcquire分为公平和非公平两种，主要做了两件事：

*  如果state为0，尝试获取锁资源
 *  如果state不为0，看一下是不是锁重入操作

非公平：

// 非公平锁实现！
    final boolean nonfairTryAcquire(int acquires) {
        
        // 拿到当前线程！
        final Thread current = Thread.currentThread();
        // 拿到AQS的state
        int c = getState();
        // 如果state == 0，说明没有线程占用着当前的锁资源
        if (c == 0) {
        
            // 没人占用锁资源，直接抢一波（不管有没有线程在排队）
            if (compareAndSetState(0, acquires)) {
        
                // 将当前占用这个互斥锁的线程属性设置为当前线程
                setExclusiveOwnerThread(current);
                // 返回true，拿锁成功
                return true;
            }
        }
        // 当前state != 0，说明有线程占用着锁资源
        // 判断拿着锁的线程是不是当前线程（锁重入）
        else if (current == getExclusiveOwnerThread()) {
        
            // 将state再次+1
            int nextc = c + acquires;
            // 锁重入是否超过最大限制
            // 01111111 11111111 11111111 11111111   + 1
            // 10000000 00000000 00000000 00000000
            // 抛出error
            if (nextc < 0) 
                throw new Error("Maximum lock count exceeded");
            // 将值设置给state
            setState(nextc);
            // 返回true，拿锁成功
            return true;
        }
        return false;
    }

公平锁：

// 公平锁实现
    protected final boolean tryAcquire(int acquires) {
        
        // 拿到当前线程！
        final Thread current = Thread.currentThread();
        // 拿到AQS的state
        int c = getState();
        if (c == 0) {
        
            // 判断是否有线程在排队，如果有线程排队，返回true，配上前面的!，那会直接不执行返回最外层的false
            if (!hasQueuedPredecessors() &&
                // 如果没有线程排队，直接CAS尝试获取锁资源
                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的addWaiter方法 ###

在获取锁资源失败后，需要将当前线程封装为Node对象，并且插入到AQS队列的末尾

// 将当前线程封装为Node对象，并且插入到AQS队列的末尾
    private Node addWaiter(Node mode) {
        
        // 将当前线程封装为Node对象，mode为null，代表互斥锁
        Node node = new Node(Thread.currentThread(), mode);
        // pred是tail节点
        Node pred = tail;
        // 如果pred不为null，有线程正在排队
        if (pred != null) {
        
            // 将当前节点的prev，指定tail尾节点
            node.prev = pred;
            // 以CAS的方式，将当前节点变为tail节点
            if (compareAndSetTail(pred, node)) {
        
                // 之前的tail的next指向当前节点
                pred.next = node;
                return node;
            }
        }
        // 添加的流程为，  自己prev指向、tail指向自己、前节点next指向我
        // 如果上述方式，CAS操作失败，导致加入到AQS末尾失败，如果失败，就基于enq的方式添加到AQS队列
        enq(node);
        return node;
    }
    
    // enq，无论怎样都添加进入
    private Node enq(final Node node) {
        
        for (;;) {
        
            // 拿到tail
            Node t = tail;
            // 如果tail为null，说明当前没有Node在队列中
            if (t == null) {
         
                // 创建一个新的Node作为head，并且将tail和head指向一个Node
                if (compareAndSetHead(new Node()))
                    tail = head;
            } else {
        
                node.prev = t;
                if (compareAndSetTail(t, node)) {
        
                    t.next = node;
                    return t;
                }
            }
        }
    }

### ReentrantLock的acquireQueued方法 ###

acquireQueued方法会查看当前排队的Node是否是head的next，如果是，尝试获取锁资源，如果不是或者获取锁资源失败那么就尝试将当前Node的线程挂起（unsafe.park()）。

在挂起线程前，需要确认当前节点的上一个节点的状态必须是小于等于0，如果为1，代表是取消的节点，不能挂起；如果为-1，代表挂起当前线程；如果为-2，-3，需要将状态改为-1之后，才能挂起当前线程。

// acquireQueued方法
    // 查看当前排队的Node是否是head的next，
    // 如果是，尝试获取锁资源，
    // 如果不是或者获取锁资源失败那么就尝试将当前Node的线程挂起（unsafe.park()）
    final boolean acquireQueued(final Node node, int arg) {
        
        // 标识。
        boolean failed = true;
        try {
        
            // 循环走起
            for (;;) {
        
                // 拿到上一个节点
                final Node p = node.predecessor();
                if (p == head && // 说明当前节点是head的next
                    tryAcquire(arg)) {
         // 竞争锁资源，成功：true，失败：false
                    // 进来说明拿到锁资源成功
                    // 将当前节点置位head，thread和prev属性置位null
                    setHead(node);
                    // 帮助快速GC
                    p.next = null; 
                    // 设置获取锁资源成功
                    failed = false;
                    // 不管线程中断。
                    return interrupted;
                }
                // 如果不是或者获取锁资源失败,尝试将线程挂起
                // 第一个事情，当前节点的上一个节点的状态正常！
                // 第二个事情，挂起线程
                if (shouldParkAfterFailedAcquire(p, node) &&
    				// 通过LockSupport将当前线程挂起
                    parkAndCheckInterrupt())
            }
        } finally {
        
            if (failed)
                cancelAcquire(node);
        }
    }
    
    // 确保上一个节点状态是正确的
    private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
        
        // 拿到上一个节点的状态
        int ws = pred.waitStatus;
        // 如果上一个节点为 -1
        if (ws == Node.SIGNAL)
            // 返回true，挂起线程
            return true;
        // 如果上一个节点是取消状态
        if (ws > 0) {
        
            // 循环往前找，找到一个状态小于等于0的节点
            do {
        
                node.prev = pred = pred.prev;
            } while (pred.waitStatus > 0);
            pred.next = node;
        } else {
        
            // 将小于等于0的节点状态该为-1
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        }
        return false;
    }

### ReentrantLock的unlock方法 ###

释放锁资源：

*  将state-1。
 *  如果state减为0了，唤醒在队列中排队的Node。

// 真正释放锁资源的方法
    public final boolean release(int arg) {
        
        // 核心的释放锁资源方法
        if (tryRelease(arg)) {
        
            // 释放锁资源释放干净了。  （state == 0）
            Node h = head;
            // 如果头节点不为null，并且头节点的状态不为0，唤醒排队的线程
            if (h != null && h.waitStatus != 0)、
                // 唤醒线程
                unparkSuccessor(h);
            return true;
        }
        // 释放锁成功，但是state != 0
        return false;
    }
    // 核心的释放锁资源方法
    protected final boolean tryRelease(int releases) {
        
        // 获取state - 1
        int c = getState() - releases;
        // 如果释放锁的线程不是占用锁的线程，抛异常
        if (Thread.currentThread() != getExclusiveOwnerThread())
            throw new IllegalMonitorStateException();
        // 是否成功的将锁资源释放利索 （state == 0）
        boolean free = false;
        if (c == 0) {
        
            // 锁资源释放干净。
            free = true;
            // 将占用锁资源的属性设置为null
            setExclusiveOwnerThread(null);
        }
        // 将state赋值
        setState(c);
        // 返回true，代表释放干净了
        return free;
    }
    
    // 唤醒节点
    private void unparkSuccessor(Node node) {
        
        // 拿到头节点状态
        int ws = node.waitStatus;
        // 如果头节点状态小于0，换为0
        if (ws < 0)
            compareAndSetWaitStatus(node, ws, 0);
        // 拿到当前节点的next
        Node s = node.next;
        // 如果s == null ，或者s的状态为1
        if (s == null || s.waitStatus > 0) {
        
            // next节点不需要唤醒，需要唤醒next的next
            s = null;
            // 从尾部往前找，找到状态正常的节点。（小于等于0代表正常状态）
            for (Node t = tail; t != null && t != node; t = t.prev)
                if (t.waitStatus <= 0)
                    s = t;
        }
        // 经过循环的获取，如果拿到状态正常的节点，并且不为null
        if (s != null)
            // 唤醒线程
            LockSupport.unpark(s.thread);
    }

为什么唤醒线程时，为啥从尾部往前找，而不是从前往后找？

因为在addWaiter操作时，是先将当前Node的prev指针指向前面的节点，然后是将tail赋值给当前Node，最后将上一个节点的next指针，指向当前Node。如果从前往后，通过next去找，可能会丢失某个节点，导致这个节点不会被唤醒。如果从后往前找，肯定可以找到全部的节点。

[image.png]: https://img-blog.csdnimg.cn/img_convert/4ac50a886d2b69f514ac42e693813284.png