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

Posted 烟锁迷城

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了java多线程进阶LOCK锁及其原理相关的知识,希望对你有一定的参考价值。

目录

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;

 

以上是关于java多线程进阶LOCK锁及其原理的主要内容,如果未能解决你的问题,请参考以下文章

Java 高并发与多线程;:synchronized 关键字的实现原理

Java-进阶:多线程2

java 多线程中的锁的类别及使用

Java 多线程进阶-并发协作控制

一文看透Java高并发:Synchronized锁的性质原理及其缺陷

Java并发006使用层面:Lock锁机制全解析