ReentrantLock和AQS解析

骑猪看日落 2023-07-24 08:33 67阅读 0赞

本文参考了https://blog.csdn.net/javazejian/article/details/75043422  
ReentrantLock是在多线程问题处理中常用的锁结构，作为一种排它锁 他比 synchronized要灵活，使用上却要比 synchronized复杂一些。  
要了解ReentrantLock的原理就不得不提到AQS，AQS 即 AbstractQueuedSynchronizer 的缩写 翻译为抽象的同步队列。提供了以Node为基础构建的双向链表存储请求锁的线程。  
我们来看以下ReentrantLock的类结构  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NjE2MTcz_size_16_color_FFFFFF_t_70]  
我们这里关注他的三个内部类，公平锁 ，非公平锁 和 Sync  
![在这里插入图片描述][20200410111303663.png]  
可以看到Sync正是继承自 AbstractQueuedSynchronizer，我们再来看一下AbstractQueuedSynchronizer  
![在这里插入图片描述][20200410111308838.png]

public abstract class AbstractOwnableSynchronizer
        implements java.io.Serializable { 
    
        /** Use serial ID even though all fields transient. */
        private static final long serialVersionUID = 3737899427754241961L;
    
        /** * Empty constructor for use by subclasses. */
        protected AbstractOwnableSynchronizer() {  }
    
        /** * The current owner of exclusive mode synchronization. */
        private transient Thread exclusiveOwnerThread;
    
        /** * Sets the thread that currently owns exclusive access. * A {@code null} argument indicates that no thread owns access. * This method does not otherwise impose any synchronization or * {@code volatile} field accesses. * @param thread the owner thread */
        protected final void setExclusiveOwnerThread(Thread thread) { 
            exclusiveOwnerThread = thread;
        }
    
        /** * Returns the thread last set by {@code setExclusiveOwnerThread}, * or {@code null} if never set. This method does not otherwise * impose any synchronization or {@code volatile} field accesses. * @return the owner thread */
        protected final Thread getExclusiveOwnerThread() { 
            return exclusiveOwnerThread;
        }
    }

这个类就很简洁了，他定义了当前持有锁的线程。我们再回头来看AbstractQueuedSynchronizer。  
![在这里插入图片描述][20200410111317488.png]  
我们来看一下Node的定义

static final class Node { 
            /** 标记为共享锁*/
            static final Node SHARED = new Node();
            /** 标记为排他锁*/
            static final Node EXCLUSIVE = null;
            //以下4个状态都是 waitStatus 所支持的状态 标识当前Node 中线程的状态 
            /** 当前线程可以取消 */
            static final int CANCELLED = 1;
            /** 当前线程可以被唤醒 */
            static final int SIGNAL = -1;
            /** 当前线程等待condition 条件 */
            static final int CONDITION = -2;
            /** * 在共享模式中使用表示获得的同步状态会被传播 */
            static final int PROPAGATE = -3;
    
            /** 等待状态 */
            volatile int waitStatus;
    
            /** 上一个节点*/
            volatile Node prev;
    
            /** 下一个节点*/
            volatile Node next;
    
            /** 当前节点持有的线程*/
            volatile Thread thread;
    
            /** * 等待队列中的后继结点，这个与Condition有关 */
            Node nextWaiter;
    
            /** * 是否共享锁 */
            final boolean isShared() { 
                return nextWaiter == SHARED;
            }
    
            /** * 获取上一个节点 */
            final Node predecessor() throws NullPointerException { 
                Node p = prev;
                if (p == null)
                    throw new NullPointerException();
                else
                    return p;
            }
    
            Node() {  // Used to establish initial head or SHARED marker
            }
    
            Node(Thread thread, Node mode) {  // Used by addWaiter
                this.nextWaiter = mode;
                this.thread = thread;
            }
    
            Node(Thread thread, int waitStatus) {  // Used by Condition
                this.waitStatus = waitStatus;
                this.thread = thread;
            }
        }

看完了Node 我们来看一下AbstractQueuedSynchronizer 自身的属性

/** * 队首 */
        private transient volatile Node head;
    
        /** * 队尾 */
        private transient volatile Node tail;
    
        /** * 锁状态 为 0 时 没有线程持有锁 */
        private volatile int state;

在AbstractQueuedSynchronizer 中提供了很多方法维护Node的数据结构，以及定义了获取锁和释放锁的基础方法。了解了他们的内部结构，我们来看这张图  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NjE2MTcz_size_16_color_FFFFFF_t_70 1]  
这张图描述了AQS的相关类的关系。  
AbstractOwnableSynchronizer：抽象类，定义了存储独占当前锁的线程和获取的方法

AbstractQueuedSynchronizer：抽象类，AQS框架核心类，其内部以虚拟队列的方式管理线程的锁获取与锁释放，其中获取锁(tryAcquire方法)和释放锁(tryRelease方法)并没有提供默认实现，需要子类重写这两个方法实现具体逻辑，目的是使开发人员可以自由定义获取锁以及释放锁的方式。

Node：AbstractQueuedSynchronizer 的内部类，用于构建虚拟队列(链表双向链表)，管理需要获取锁的线程。

Sync：抽象类，是ReentrantLock的内部类，继承自AbstractQueuedSynchronizer，实现了释放锁的操作(tryRelease()方法)，并提供了lock抽象方法，由其子类实现。

NonfairSync：是ReentrantLock的内部类，继承自Sync，非公平锁的实现类。

FairSync：是ReentrantLock的内部类，继承自Sync，公平锁的实现类。

ReentrantLock：实现了Lock接口的，其内部类有Sync、NonfairSync、FairSync，在创建时可以根据fair参数决定创建NonfairSync(默认非公平锁)还是FairSync。  
接下来我们从一个实例出发，从源码上看一个线程从获取锁到释放锁的过程。

/** * @Description * @Author changyandong * @Emoji (゜ - ゜)つ干杯 * @Created Date: 2020/4/1 9:12 * @ClassName TestThread * @Version: 1.0 */
    public class TestThread implements Runnable { 
        static ReentrantLock reentrantLock = new ReentrantLock();
        public static int i = 0;
        @Override
        public void run() { 
    
            reentrantLock.lock();
            try { 
                IntStream.range(0,10000).forEach(k->i++);
            }finally { 
                reentrantLock.unlock();
            }
    
        }
    
        public static void main(String[] args) throws InterruptedException { 
            TestThread testThread = new TestThread();
            Thread thread1 = new Thread(testThread);
            Thread thread2 = new Thread(testThread);
            thread1.start();
            thread2.start();
            thread1.join();
            thread2.join();
            System.out.println(TestThread.i);
        }
    }

我启动两个线程操作共享资源i 我们先来分析lock()

public void lock() { 
            sync.lock();
        }
     // 默认情况下我们使用的是非公平锁实现 NonfairSync
        final void lock() { 
                if (compareAndSetState(0, 1))
                    setExclusiveOwnerThread(Thread.currentThread());
                else
                    acquire(1);
            }

可以看到，他尝试使用cas将state从0设置为1，如果设置成功，当前线程就是持有锁的线程否则acquire(1);  
acquire是AbstractQueuedSynchronizer中的方法定义如何获取锁和进入等待队列

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

我们这里先看看tryAcquire

// 非公平锁的尝试获取
    protected final boolean tryAcquire(int acquires) { 
                return nonfairTryAcquire(acquires);
            }
    
    final boolean nonfairTryAcquire(int acquires) { 
                final Thread current = Thread.currentThread();
                int c = getState();
                if (c == 0) { 
                  // cas 如果当前没有人持有锁 设置state 为1 
                    if (compareAndSetState(0, acquires)) { 
                        setExclusiveOwnerThread(current);
                        return true;
                    }
                }
                // 如果是重入 将state + 1
                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;
            }

如果这里返回false表示尝试获取锁失败，那么就应该将当前线程置入队列中也就是acquireQueued的逻辑，但是在他之前我们看一下addWaiter()  
addWaiter() 也是 AbstractQueuedSynchronizer定义的创建node并加入队尾的方法

private Node addWaiter(Node mode) { 
            // 将当前线程封装为node
            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的上一个节点为 之前的队尾
                node.prev = pred;
                // cas 尝试将 tail设置为node
                if (compareAndSetTail(pred, node)) { 
                    pred.next = node;
                    return node;
                }
            }
            // cas设置失败 
            enq(node);
            return node;
        }
    
    private Node enq(final Node node) { 
           // 无限循环的for
            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;
                    }
                }
            }
        }

接着来看 添加队列方法acquireQueued

final boolean acquireQueued(final Node node, int arg) { 
            boolean failed = true;
            try { 
                boolean interrupted = false;
               // 无限循环
                for (;;) { 
                // 获取 node 的前置节点 
                    final Node p = node.predecessor();
                // 如果前置为 head 那么就尝试获取锁
                    if (p == head && tryAcquire(arg)) { 
                    // 获取锁成功，当前node就变成了 head 并释放原head 
                        setHead(node);
                        p.next = null; // help GC
                        failed = false;
                        return interrupted;
                    }   
                   //如果p不是head 或者获取锁失败，那么判断是否需要Park线程
                    if (shouldParkAfterFailedAcquire(p, node) &&
                        parkAndCheckInterrupt())
                        interrupted = true;
                }
            } finally { 
            // 如果出了问题 比如 shouldParkAfterFailedAcquire的空指针 那么放弃 node
                if (failed)
                    cancelAcquire(node);
            }
        }
    
     private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) { 
            // 获取上一个节点的等待状态
            int ws = pred.waitStatus;
            // 如果是唤醒状态 返回true
            if (ws == Node.SIGNAL)
                /* * This node has already set status asking a release * to signal it, so it can safely park. */
                return true;
            // 如果状态大于 0 说明线程应该被结束 释放节点
            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;
            // ws = 0 或者 -3 将它cas为 -1
            } 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;
        }
         // 锁住当前线程
    private final boolean parkAndCheckInterrupt() { 
            LockSupport.park(this);
            return Thread.interrupted();
        }
    
    private void cancelAcquire(Node node) { 
            // Ignore if node doesn't exist
            if (node == null)
                return;
    
            node.thread = null;
    
            // Skip cancelled predecessors
            Node pred = node.prev;
            while (pred.waitStatus > 0)
                node.prev = pred = pred.prev;
    
            // predNext is the apparent node to unsplice. CASes below will
            // fail if not, in which case, we lost race vs another cancel
            // or signal, so no further action is necessary.
            Node predNext = pred.next;
    
            // Can use unconditional write instead of CAS here.
            // After this atomic step, other Nodes can skip past us.
            // Before, we are free of interference from other threads.
            node.waitStatus = Node.CANCELLED;
    
            // If we are the tail, remove ourselves.
            if (node == tail && compareAndSetTail(node, pred)) { 
                compareAndSetNext(pred, predNext, null);
            } else { 
                // If successor needs signal, try to set pred's next-link
                // so it will get one. Otherwise wake it up to propagate.
                int ws;
                if (pred != head &&
                    ((ws = pred.waitStatus) == Node.SIGNAL ||
                     (ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) &&
                    pred.thread != null) { 
                    Node next = node.next;
                    if (next != null && next.waitStatus <= 0)
                        compareAndSetNext(pred, predNext, next);
                } else { 
                    unparkSuccessor(node);
                }
    
                node.next = node; // help GC
            }
        }

到这里 lock方法的流程就结束了。接下来看unLock() 这个就相对简单了

public void unlock() { 
            sync.release(1);
        }
    
    public final boolean release(int arg) { 
            if (tryRelease(arg)) { 
                // 如果锁已经没有线程持有了 就唤醒一个队列里的Node
                Node h = head;
                if (h != null && h.waitStatus != 0)
                    unparkSuccessor(h);
                return true;
            }
            return false;
        }
    //tryRelease 很简单 就是改state 如果 state为0 将持有线程设置为0
    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) { 
            /* * 设置waitStatus为 0 */
            int ws = node.waitStatus;
            if (ws < 0)
                compareAndSetWaitStatus(node, ws, 0);
    
            /* * 判断下一个节点是否应该被唤醒，如果不是 倒序找到一个需要被唤醒的节点 唤醒 他 */
            Node s = node.next;
            if (s == null || s.waitStatus > 0) { 
                s = null;
                // 这里循环没有退出 找到的是node节点的后续节点中最早进入队列的一个 所有unpark后 他的prev 就是head 
                for (Node t = tail; t != null && t != node; t = t.prev)
                    if (t.waitStatus <= 0)
                        s = t;
            }
            if (s != null)
                LockSupport.unpark(s.thread);
        }

在这里我再放上一张流程图  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NjE2MTcz_size_16_color_FFFFFF_t_70 2]  
unLock 过程  
![在这里插入图片描述][20200410111354368.png]  
接下来我们来看公平锁，公平锁保证严格的FIFO，不会再acquire之前先尝试设置state，公平锁与非公平锁的区别就在tryAcquire()方法的实现略有区别。

protected final boolean tryAcquire(int acquires) { 
                final Thread current = Thread.currentThread();
                int c = getState();
                if (c == 0) { 
                    // 如果c == 0 hasQueuedPredecessors 判断的是 队列是否为 空或者 当前线程是重入才执行cas操作
                    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.
            // aqs 会维护head 赋值一定在tail 之前，这样我们获取时就要先获取tail 保证线程安全
            Node t = tail; // Read fields in reverse initialization order
            Node h = head;
            Node s;
            // h== t 队列为空 h.next == null 队列为空 s.thread == Thread.currentThread() 重入 因为下一个获取锁的线程为本线程所以符合公平锁定义
            return h != t &&
                ((s = h.next) == null || s.thread != Thread.currentThread());
        }

下面放上我自己画的 lock流程  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NjE2MTcz_size_16_color_FFFFFF_t_70 3]  
关于Condition 之后再补充

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NjE2MTcz_size_16_color_FFFFFF_t_70]: /images/20230528/2d1e04fb8c614eaba3c4bf0636837e32.png
[20200410111303663.png]: /images/20230528/ac2430c65ee74a4492759f4b00324d58.png
[20200410111308838.png]: /images/20230528/38a86e659cf64fd08f5502a921c37aea.png
[20200410111317488.png]: /images/20230528/e9598f36fc274a90b68356695c061896.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NjE2MTcz_size_16_color_FFFFFF_t_70 1]: /images/20230528/109547ba0ced47e6b048fb6539426227.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NjE2MTcz_size_16_color_FFFFFF_t_70 2]: /images/20230528/dfc13083c7d74488b69ef4c4912aa323.png
[20200410111354368.png]: /images/20230528/48eb7ccab8784005b99ebae080163ddb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NjE2MTcz_size_16_color_FFFFFF_t_70 3]: /images/20230528/222ae8e7bd1f4f06abfadde97e6541ed.png