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

Posted 2022-11-28 qq_22426297

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

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是没有值的）。