java多线程进阶LOCK锁及其原理

Posted 2022-12-14 烟锁迷城

目录

1、实现

2、实现思路

3、源码阅读

3.1、LOCK方法

3.1.1、公平锁

3.1.2、非公平锁

 3.1.3、加入与抢占队列

3.2、UNLOCK

---

1、实现

ReentrantLock是实现Lock接口的锁，是JUC工具包下的一种锁实现，它基于代码层面实现。

这是简单的示例

Lock lock = new ReentrantLock();
try 
    lock.lock()
 catch（）

 finally 
    lock.unlock()

lock命令，抢占锁，没抢占到会阻塞

tryLock命令，抢占锁，没抢占到不会阻塞，返回true抢占成功，false抢占失败

unlock命令，解锁，因为是代码层面的加锁，所以必须要放在finally代码块里进行解锁

synchronized是JVM层面的加锁，所以会自动解锁，无需强行在finally中进行操作

2、实现思路

从实现角度出发，有以下的问题需要解决

• 抢占锁成功的标记

• 抢占锁成功处理

• 抢占锁失败处理

• 抢占成功后释放锁处理

AbstractQueuedSynchronizer简称AQS，是实现整个ReentrantLock的关键，可以用一个流程图展示它的生效方式：

3、源码阅读

在ReentrantLock的源码中，可以看到其内部有两种实现方式，公平锁和非公平锁，在其无参构造中实现的是非公平锁。

private final Sync sync;
public ReentrantLock() 
    sync = new NonfairSync();

 sync是一个final修饰的成员变量，它的类型是Sync，继承了AbstractQueuedSynchronizer，它在ReentrantLock的两个实现就是之前提到过的FairSync和NonfairSync

static class Sync extends AbstractQueuedSynchronizer

3.1、LOCK方法

3.1.1、公平锁

 FairSync，公平锁实现。

lock()加锁方法中只有一个acquire方法，重点看这个。

if (!tryAcquire(arg) &&acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        selfInterrupt();

这个判断很长，所以一段一段看，首先是tryAcquire方法。

final Thread current = Thread.currentThread();——获取到当前的线程

int c = getState();——获取到锁的状态

if (c == 0)——判断状态是否为0，即是否为无锁条件

if (!hasQueuedPredecessors() && compareAndSetState(0, acquires))——这个判断分为两部分，其一是hasQueuedPredecessors()，这个方法用来判断是否已经具有队列，即AQS队列，如果不存在，则进入下一部分判断。compareAndSetState(0, acquires)，这是典型的CAS语句判断，预期值是0，修改值是acquires，当然这里就是1。这个CAS的判断目的是在多线程条件下，可能会有多个线程涌入判断，此刻进行互斥的CAS乐观锁操作，可以保证只有一个线程能成功抢占。

setExclusiveOwnerThread(current);——将当前线程放入exclusiveOwnerThread

return true;——返回true，结束方法。

else if (current == getExclusiveOwnerThread()) ——如果当前锁状态不为0，即已经加锁，则判断当前线程是否是获得锁的线程

 int nextc = c + acquires;——如果是，重入次数将会增加，这里是增加1

if (nextc < 0)——如果重入次数小于0

throw new Error("Maximum lock count exceeded");——抛出异常

setState(nextc);——将重入次数放入锁状态中，即非0即为加锁

return true;——返回true，结束方法

final void lock() 
    acquire(1);

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

protected final boolean tryAcquire(int acquires) 
    //获取到当前的线程
    final Thread current = Thread.currentThread();
    //获取当前锁的状态
    int c = getState();
    //判断锁的状态是否为0
    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;

3.1.2、非公平锁

NonfairSync，非公平锁

可以清楚地看到lock方法和公平锁相比略有不同

if (compareAndSetState(0, 1))——CAS判断，是否加锁

setExclusiveOwnerThread(Thread.currentThread());——如果不加锁，直接插队抢占，这就是非公平锁的第一次插队抢占。

nonfairTryAcquire方法和tryAcquire方法之间的区别就在于在第一次判断的时候缺少了方法：hasQueuedPredecessors()，是否已经具有队列，这是第二次插队抢占

final void lock() 
    if (compareAndSetState(0, 1))
        setExclusiveOwnerThread(Thread.currentThread());
    else
        acquire(1);

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

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

 3.1.3、加入与抢占队列

if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) selfInterrupt()

在这行代码中，已经分析过tryAcquire方法，如果抢占失败，返回false，取反，得到true，继续向下执行，开始执行acquireQueued方法。

在acquireQueued内还有addWaiter方法，此方法顾名思义，是添加等待节点，那么可以继续看其内部数据。

 Node node = new Node(Thread.currentThread(), mode);——创建新节点，此处mode为null，线程为当前线程，可以看Node的这个构造，能够看出来，这个构造内部是将null作为下一个等待节点，当前线程为节点

Node pred = tail;——取出尾节点为pred

if (pred != null)——如果尾结点不为空，即已经存在双向链表，则进入判断内部

node.prev = pred;——pred将作为新节点的上一节点

if (compareAndSetTail(pred, node))——将尾结点替换为新节点

pred.next = node;——将原本的尾结点的下一节点指向新的尾结点，完成双向链表的构建

return node;——返回节点

static final Node EXCLUSIVE = null;
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(Thread thread, Node mode)      
    // Used by addWaiter
    this.nextWaiter = mode;
    this.thread = thread;

enq方法，可以看到for(;;)这是一个死循环，即自旋式的实现。

Node t = tail;——先让t为tail，即尾结点

if (t == null)——如果尾结点为空，证明这是一个全新的队列

if (compareAndSetHead(new Node()))——执行CAS操作，将新创建一个节点并赋予头结点head

tail = head;——头结点和尾结点都指向新节点，此分支结束

接下来，看另一个分支，此分支代表不为新链表的判定结果，因为是自旋，所以必定会执行至此分支。

node.prev = t;——将尾结点赋予当前节点的上一节点，即将尾结点作为此节点指向的上一节点

 if (compareAndSetTail(t, node))——CAS操作，将此节点替换为尾结点

t.next = node;——将此节点作为t节点指向的下一个节点。如果是新的队列，t节点原本既是head节点又是tail节点，此时tail节点已经被换为node节点，所以t节点是head节点，即新创建的那个节点，由此可知，是将head节点的下一节点指向node，即tail节点，完成双向链表构建。如果已经存在队列，那么就是将原本的尾结点的下一节点指向新的尾结点，完成加入尾结点的操作。

return t;——返回t节点

此方法的目的是在没有构建队列的情况下，创造一个空的节点作为头结点，以新加入的Thread作为尾节点，完成双向链表的构建。在有队列的情况下，将新节点作为尾节点加入到双向链表中去。

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;
            
        
    

acquireQueued方法，在获取到节点之后，将执行此方法。

for (;;) ——自旋方法

final Node p = node.predecessor();——将获取node节点的上一节点

if (p == head && tryAcquire(arg))——如果上一节点是头结点，且tryAcquire方法尝试抢占锁成功（也与公平或非公平锁的实现有关）

setHead(node);——将此节点作为头结点，在其内部，是将头结点指向改为当前节点，并且让原本的头结点失去指向。

p.next = null; ——help GC，这句注释是JAVA官方的，这种操作类似于ArrayList的最后一个数组位置为null一样，与可达性分析有关，有助于GC回收

failed = false;——将是否失败的标志位置为false

return interrupted;——返回重点标志，false

if (shouldParkAfterFailedAcquire(p, node) &&parkAndCheckInterrupt())——shouldParkAfterFailedAcquire方法直译过来是在失败的抢占之后应该执行park方法，parkAndCheckInterrupt方法直译过来是park并检查中断。

shouldParkAfterFailedAcquire(p, node)——

interrupted = true;——中断标志为true

finally代码块，这是一定要执行的代码块

if (failed)——是否失败

cancelAcquire(node);——如果失败了，就取消抢占

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

private void setHead(Node node) 
    head = node;
    node.thread = null;
    node.prev = null;

 int ws = pred.waitStatus;——获取上一个节点的等待状态

if (ws == Node.SIGNAL)——查看状态是否等于Node.SIGNA（-1）

return true;——如果是，返回true，代表已进入等待状态

if (ws > 0) ——如果等待状态大于0

do ——进入循环

node.prev = pred = pred.prev;——循环的内容是将节点的前一个节点指向前一个节点的前一个节点，即从后向前，排除掉三个节点中中间的那个，将第三个节点的前一节点直接指向第二个，这样做的目的是为了防止其他线程在新加入节点时，会将下一个节点指向，为了避免中断这个指向操作，这里的去除节点操作就变为从后向前

while (pred.waitStatus > 0);——循环条件是waitStatus大于0，在NODE的状态里只有CANCELLED（取消）为1，是大于0的，也就是说，循环排除取消状态的节点。

pred.next = node;去除完毕后，将前一个节点的下一指向节点置为当前节点。

else

compareAndSetWaitStatus(pred, ws, Node.SIGNAL);——如果不大于0，则将前一节点状态置为-1

return false;——返回false

shouldParkAfterFailedAcquire这个方法代表着这个节点有一次抢占的机会，即不为-1也不大于0的情况下，会进入到返回false的情况，此时不会引发park，中断标志也不会被置为true，一旦执行过一次，就会导致这个节点成为SIGNAL节点，返回true，线程将会被挂起。

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;

 LockSupport.park(this);——挂起当前线程

return Thread.interrupted();——返回当前线程的中断标志

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

cancelAcquire方法是finally内必然会执行的方法。

if (node == null)——如果节点为null

return;——直接结束

node.thread = null;——将node的线程置为null，因为线程被中断，所以要取消这个线程节点。

Node pred = node.prev;——取出当前节点的前一节点

while (pred.waitStatus > 0)——如果前一节点状态为CANCELLED

node.prev = pred = pred.prev;——去除这个失效节点

Node predNext = pred.next;——取得前一节点的下一节点

node.waitStatus = Node.CANCELLED;——当前节点置为失效

if (node == tail && compareAndSetTail(node, pred))——如果当前节点为尾结点且前一节点成功取代当前节点为尾结点

compareAndSetNext(pred, predNext, null);——替换下一节点为null

int ws;——一个ws

if (pred != head &&——前一节点不为头结点

((ws = pred.waitStatus) == Node.SIGNAL ||——节点状态为SIGNAL

(ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL)))——或者节点状态小于0，且被替换为SIGNAL成功

&&pred.thread != null)——并且前一节点线程不能为null

Node next = node.next;——取出当前节点的下一节点

if (next != null && next.waitStatus <= 0)——如果下一节点可用

compareAndSetNext(pred, predNext, next);——前一节点的下一节点被当前节点的下一节点取代，即重新指向

unparkSuccessor(node);——否则取消阻塞

node.next = node; ——help GC，官方注释，与可达性分析有关，保证没有类指向这个node，让它容易被回收。

总结，这个方法的目的是为了将无效的节点排除，并且在线程被中断的情况下，将线程节点排出队列。

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
    

 compareAndSetWaitStatus(node, ws, 0);——如果状态小于0，置为0，不代表任何状态

如果节点未失效，则恢复线程至等待。

如果节点失效或为null，则执行移除。移除的方式是循环，依旧采用从尾结点至头结点的方式进行移除。

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

 

3.2、UNLOCK

可以看到，在unlock方法里，已经不再区分公平与非公平锁。

tryRelease(arg)方法，尝试解锁，如果成功，就会返回true

Node h = head;——获取头结点

if (h != null && h.waitStatus != 0)——如果头结点不为空并且等待状态也不为0

unparkSuccessor(h);——执行线程阻塞恢复

return true;——返回true

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

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

int c = getState() - releases;——获取状态，releases，当然这里的数值是1

if (Thread.currentThread() != getExclusiveOwnerThread())——如果当前线程不等于占有锁的线程，

boolean free = false;——抛出错误

if (c == 0) ——如果锁状态为0

free = true;——代表锁可以被释放

setExclusiveOwnerThread(null);——释放锁

setState(c);——状态重置

return free;——返回是否解锁成功

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;