AQS中非公平锁的实现原理简介

Posted qq_22426297

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了AQS中非公平锁的实现原理简介相关的知识,希望对你有一定的参考价值。

先了解非公平锁,公平锁自然就很简单了

Lock lock =new ReentrantLock();

lock.lock();

在ReentrantLoc类中,有Sync,FairSync,NonfaiSync

类图如下,

 

首先,public void lock()

sync.lock();

调用sync的lock,sync会调用子类公平锁或者非公平锁的lock。

先看非公平锁的lock方法

 

final void lock()

if (compareAndSetState(0, 1))

setExclusiveOwnerThread(Thread.currentThread());

else

acquire(1);

 

compareAndSetState 是CAS state,由jvm来实现底层的具体操作,是个线程安全的操作,第一个参数是期望值,第二个是修改值,如果state=0,那么把排它锁给当前线程,否则进入acquire获取锁的方法中。

 

public final void acquire(int arg)

    if (!tryAcquire(arg) &&

        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))

        selfInterrupt();

 

这个写法很简洁,第一个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;

意思是,如果当前的state是0,则可以去获取锁,否则,判断是否又是个线程由执行了一次lock,导致重入,则把state加1,后面使用unlock的时候,state也要一次一次调用,才能减到0.

 

回到!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)

先看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(Thread thread, Node mode) // Used by addWaiter

    this.nextWaiter = mode;

    this.thread = thread;

可以查看static final Node EXCLUSIVE = null;(也就是mode是null)  EXCLUSIVE意思是排他锁,也就不是共享锁。首先,把当前这个结点包装成Node结点,Node是AQS的内部类,判断tail尾节点是否为null,如果是null,则进入enq(node)。

 

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;

        

    

 

提一下,为什么用for(;;)而不用while(true)?因为for(;;)生成的字节码要少,节约点空间。这里是个死循环,如果tail为空的话,就要初始化这个AQS,大致如下

第二次进入循环后,tail就有值了,此时再来次CAS操作,这次设置的是tail,如果失败就再次进入循环,来把当前线程设置成tail,addWaiter中tail如果不为空,也是这样的操作

再回到acquireQueued(addWaiter(Node.EXCLUSIVE), arg)

 

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

    

先看for循环,p获取到node的前一个节点

final Node predecessor() throws NullPointerException

    Node p = prev;

    if (p == null)

        throw new NullPointerException();

    else

        return p;

如果p是头节点,并且tyrAcquire成功了(意味着这个node获得了锁),就把node设置成头结点。有的小伙伴可能就会疑惑,为什么会成功了?因为除了lock,还有unlock方法,会把state重新设置为0,这样就会有机会得到锁了。具体后面再讲。

private void setHead(Node node)

    head = node;

    node.thread = null;

    node.prev = null;

这个时候队列就变成了

这样一来,很关键的一步,3.Thread=null,这样就把前面那个空的头结点取代了,而且这样就保证了一致性,初始化的头结点不会影响队列的效果。

 

再看看,如果没有成功会怎样

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;

看方法名,应该在获取锁失败后park吗?park这个方法以后再讲,现在就理解为wait()或者阻塞,或者被挂起。先看前面节点的ws,我们先看看有哪些waitState

/** waitStatus value to indicate thread has cancelled */

static final int CANCELLED = 1;

/** waitStatus value to indicate successor's thread needs unparking */

static final int SIGNAL = -1;

/** waitStatus value to indicate thread is waiting on condition */

static final int CONDITION = -2;

/**

* waitStatus value to indicate the next acquireShared should

* unconditionally propagate

*/

static final int PROPAGATE = -3;

 

ws大于0的情况暂时先不管(等待超时或者其他原因,状态就被标记为cancelled),看看等于0(初始化为0)和小于0且不等于-1(signal)的,又会进行一个CAS操作把前节点的ws标记为-1,这样做,可以再次进入for循环,然后再进行一次 获得锁 的方法,这次再没有获得,就会进入parkAndCheckInterrupt()方法,另外此节点的ws还是0,只是把前面节点的ws设置为-1,那么可以推出来,tail尾节点的ws等待状态为0.

private final boolean parkAndCheckInterrupt()

    LockSupport.park(this);

    return Thread.interrupted();

 

此时就将该线程park(挂起,wait,阻塞之类的,先理解),补充个小知识

① interrupt():中断本线程

myThread.interrupt();//中断的是调用interrupt()方法的线程

小结:阻塞于wait/join/sleep的线程,中断状态会被清除掉,同时收到著名的InterruptedException;而其他情况中断状态都被设置,并不一定收到异常。

② isInterrupted():检测本线程是否已经中断

myThread.isInterrupted();//判断本线程myThread是否中断

如果已经中断,则返回true,否则false。中断状态不受该方法的影响。

如果中断调用时线程已经不处于活动状态,则返回false。

③ interrupted():检测当前线程是否已经中断

Thread.interrupted();//判断该语句所在线程是否中断

如果已经中断,则返回true,否则false,并清除中断状态。换言之,如果该方法被连续调用两次,第二次必将返回false,除非在第一次与第二次的瞬间线程再次被中断。

如果中断调用时线程已经不处于活动状态,则返回false。

---------------------

作者:li_mengjun

来源:CSDN

原文:https://blog.csdn.net/li_mengjun/article/details/78162031

版权声明:本文为博主原创文章,转载请附上博文链接!

 

返回true则说明,线程被成功中断,这样一来,再次进入for循环的时候,如果获得到了锁就会返回true,再次回到

if (!tryAcquire(arg) &&

    acquireQueued(addWaiter(Node.EXCLUSIVE), arg))

    selfInterrupt();

如果后面为true才会调用sefInterrupt()方法。

 

static void selfInterrupt()

Thread.currentThread().interrupt();

清掉阻塞的状态,也可以说是复位。

得到结论,获得锁的同时,并没有唤醒线程,两步是分开进行的。

 

接下来看unlock(),会调用release方法

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;

第一个tryRelease(arg)

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;

大致意思就是,调用一次unlock就会使得当前state减一(为什么会超过1?因为可以重入,同一个线程可以调用多次lock方法,所以每次unlock只能开一个lock),当全部开完后,把排他锁的拥有者设置为null。继续看

if (h != null && h.waitStatus != 0)

    unparkSuccessor(h);

return true;

 

如果头不为空,h的ws为0(意思是只有一个节点的时候,但有时候,后面来了节点,但此时还没来得及更改)

就不会调用后面的方法,只有当有多个节点的时候才会调用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);

有时候,前面lock代码刚好在执行node.next=null的时候,s就为null了,所以还是再需要判断一下,或者s已经超时了,等待状态被标记为取消了,那么s=null(让垃圾回收器尽早回收内存)再从tail开始找离node最近的一个等待状态小于等于0的线程,如果没有找到s就为null,然后再次判断,如果s!=null才执行唤醒方法,s有可能是tail(头指针和尾指针相等的时候,thread是没有值的)。

以上是关于AQS中非公平锁的实现原理简介的主要内容,如果未能解决你的问题,请参考以下文章

ReentrantLock原理ReentrantReadWriteLock原理

AQS组件总结

可重入的独占锁ReentrantLock概述-公平与非公平锁的实现

并发编程 —— 源码分析公平锁和非公平锁

21.Lock锁原理

试验局ReentrantLock中非公平锁与公平锁的性能测试