AQS源码分析

AQS源码分析

锁的分类 悲观锁和乐观锁 在Java里使用的各种锁,几乎全都是悲观锁。synchronized从偏向锁、轻量级锁到重量级锁,全是悲观锁。JDK提供的Lock实现类全是悲观锁。其实只要有“锁对象”出现,那么就一定是悲观锁。因为乐观锁不是锁,而是一个在循环里尝试CAS的算法。 乐观...
锁的分类
悲观锁和乐观锁

在Java里使用的各种锁,几乎全都是悲观锁。synchronized从偏向锁、轻量级锁到重量级锁,全是悲观锁。JDK提供的Lock实现类全是悲观锁。其实只要有“锁对象”出现,那么就一定是悲观锁。因为乐观锁不是锁,而是一个在循环里尝试CAS的算法。

乐观锁是atomic包下的原子类。

公平锁、非公平锁

多个线程申请一把公平锁,那么当锁释放的时候,先申请的先得到,非常公平。显然如果是非公平锁,后申请的线程可能先获取到锁,是随机或者按照其他优先级排序的。

公平锁与非公平锁的区别在代码中的区别为:

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

就是非公平锁在调用lock()时,先尝试插队直接去拿锁,更改state状态为1,如果成功则把Owner线程设置为当前线程,则表示成功获得锁,没获得锁则acquire,而公平锁直接acquire。

偏向锁 → 轻量级锁 → 重量级锁

初次执行到synchronized代码块的时候,锁对象变成偏向锁(通过CAS修改对象头里的锁标志位),字面意思是“偏向于第一个获得它的线程”的锁。执行完同步代码块后,线程并不会主动释放偏向锁。当第二次到达同步代码块时,线程会判断此时持有锁的线程是否就是自己(持有锁的线程ID也在对象头里),如果是则正常往下执行。由于之前没有释放锁,这里也就不需要重新加锁。如果自始至终使用锁的线程只有一个,很明显偏向锁几乎没有额外开销,性能极高。

一旦有第二个线程加入锁竞争,偏向锁就升级为轻量级锁(自旋锁)。在轻量级锁状态下继续锁竞争,没有抢到锁的线程将自旋,即不停地循环判断锁是否能够被成功获取。获取锁的操作,其实就是通过CAS修改对象头里的锁标志位。先比较当前锁标志位是否为“释放”,如果是则将其设置为“锁定”,比较并设置是原子性发生的。这就算抢到锁了,然后线程将当前锁的持有者信息修改为自己。

长时间的自旋操作是非常消耗资源的,一个线程持有锁,其他线程就只能在原地空耗CPU,执行不了任何有效的任务,这种现象叫做忙等(busy-waiting)。如果多个线程用一个锁,但是没有发生锁竞争,或者发生了很轻微的锁竞争,那么synchronized就用轻量级锁,允许短时间的忙等现象。这是一种折衷的想法,短时间的忙等,换取线程在用户态和内核态之间切换的开销。

此忙等是有限度的(有个计数器记录自旋次数,默认允许循环10次,可以通过虚拟机参数更改)。如果锁竞争情况严重,某个达到最大自旋次数的线程,会将轻量级锁升级为重量级锁(依然是CAS修改锁标志位,但不修改持有锁的线程ID)。当后续线程尝试获取锁时,发现被占用的锁是重量级锁,则直接将自己挂起(而不是忙等),等待将来被唤醒。

独占锁与共享锁

独占锁:队首持锁,唤醒队二后tryAcquire尝试拿锁,队三及以后休眠

共享锁:相较于独占模式只唤醒队二 ,共享模式还唤醒所有mode=shared节点

独占和共享是对于加锁而言(能否多线程同时获锁),释放锁时没有独占和共享的概念。

怎么实现同步

根据参考文章1,提出了4中方式实现同步:

1:自旋实现同步

缺点:耗费cpu资源。没有竞争到锁的线程会一直占用cpu资源进行cas操作

2:yield+自旋实现同步

优点:要解决自旋锁的性能问题必须让竞争锁失败的线程不空转,而是在获取不到锁的时候能把cpu资源给让出来,yield()方法就能让出cpu资源,当线程竞争锁失败时,会调用yield方法让出cpu。

缺点:自旋+yield的方式并没有完全解决问题,当系统只有两个线程竞争锁时,yield是有效的。需要注意的是该方法只是当前让出cpu,有可能操作系统下次还是选择运行该线程

3:sleep+自旋方式实现同步

缺点:sleep的时间怎设置

4:park+自旋方式实现同步

volatile int status=0;
Queue parkQueue;//集合 数组  list

void lock(){
    while(!compareAndSet(0,1)){
        //
        park();
    }
    //lock    10分钟
   unlock()
}

void unlock(){
    lock_notify();
}

void park(){
    //将当期线程加入到等待队列
    parkQueue.add(currentThread);
    //将当期线程释放cpu  阻塞
    releaseCpu();
}
void lock_notify(){
    //得到要唤醒的线程头部线程
    Thread t=parkQueue.header();
    //唤醒等待线程
    unpark(t);
}
ReentrantLock源码分析

而我们的ReentrantLock就是采用第四种方法实现的。

首先来看一段代码

public class MyRunnable implements Runnable {
    private int num = 0;
    private ReentrantLock lock = new ReentrantLock(true);
    @Override
    public void run() {
        while (num 
初始化锁实例

首先来看构造函数,

public ReentrantLock() {
    sync = new NonfairSync();
}

public ReentrantLock(boolean fair) {
    sync = fair ? new FairSync() : new NonfairSync();
}

有两种方式,无参的默认new 一个非公平锁NonfairSync。

在我们上面的例子中,调用了第二个构造函数,并且传入的true,所以使用的是公平锁。(后续都以公平锁进行分析)

lock.lock()
public void lock() {
    sync.lock();
}

final void lock() {
    acquire(1);
}

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

调用lock()时,首先调用acquire(),对于acquire()分两部分来讲。

tryAcquire

这个方法是尝试去获取锁。

protected final boolean tryAcquire(int var1) {
    Thread var2 = Thread.currentThread();
    int var3 = this.getState();
    if (var3 == 0) {
        if (!this.hasQueuedPredecessors() && this.compareAndSetState(0, var1)) {
            this.setExclusiveOwnerThread(var2);
                return true;
        }
    } else if (var2 == this.getExclusiveOwnerThread()) {
        int var4 = var3 + var1;
        if (var4 

要想知道上面代码的意思,我们需要先了解一下AQS的一个静态内部类Node;

static final class Node {
    static final AbstractQueuedSynchronizer.Node SHARED = new AbstractQueuedSynchronizer.Node();//表示共享模式
    static final AbstractQueuedSynchronizer.Node EXCLUSIVE = null;//表示独占模式
    static final int CANCELLED = 1;//因为超时或者中断,结点会被设置为取消状态,被取消状态的结点不应该去竞争锁,只能保持取消状态不变,不能转换为其他状态。处于这种状态的结点会被踢出队列,被GC回收;
    static final int SIGNAL = -1;//表示这个结点的继任结点被阻塞了,到时需要通知它
    static final int CONDITION = -2;//表示这个结点在条件队列中,因为等待某个条件而被阻塞;
    static final int PROPAGATE = -3;//使用在共享模式头结点有可能牌处于这种状态,表示锁的下一次获取可以无条件传播;
    volatile int waitStatus;//用来表示上面4种状态
    volatile AbstractQueuedSynchronizer.Node prev;//前一个节点
    volatile AbstractQueuedSynchronizer.Node next;//后一个节点
    volatile Thread thread;//表示线程
    Node nextWaiter;//条件队列才使用。
}

getState()得到的是一个int类型的volatile int state;被volatile修饰。默认为0,表示无锁状态,1表示上锁状态,>1表示重入。

当第一个线程进来,此时getState()为0,所以进入第一个if中。

public final boolean hasQueuedPredecessors() {
    AbstractQueuedSynchronizer.Node var1 = this.tail;
    AbstractQueuedSynchronizer.Node var2 = this.head;
    AbstractQueuedSynchronizer.Node var3;
    return var2 != var1 && ((var3 = var2.next) == null || var3.thread != Thread.currentThread());
}

该方法是判断是否需要排队。

在这需要分三种情况讨论:

1:队列没有初始化

当队列没有初始化的时候,var1 = null,var2 = null.所以hasQueuedPredecessors()返回false。在第一个if里面进行取反,则表示不需要入队列处理,直接CAS进行加锁操作。并把当前线程设置成Node的thread。然后返回true,而此时在acquire()中又是取反,则不会走if里面的代码。从这里可以看出,其实加锁就是让获取到这个锁的线程能够顺畅的往下执行逻辑。

2:队列初始化且>1个节点

在队列初始化的时候,此时会new一个Node,放在最前面,而在队列的第二才是真正排队的第一个,而new 出来的那个Node可以理解为第一个拿到锁的那个的占位。当大于一个节点的时候var2 != var1为true,因为var2最开始为null,现在>1个节点,表示var2.next是有值的,所以(var3 = var2.next) == null为false。

var3.thread != Thread.currentThread())为true时表示前面已经有人在排队,那我就必须要排队。所以hasQueuedPredecessors()返回true。此时则会跳出第一个if。tryAcquire返回flase,则需在acquire()判断acquireQueued(addWaiter(Node.EXCLUSIVE), arg)在后面讲解。

var3.thread != Thread.currentThread()为flase时,则不需要排队,最简单的理解是前面排队的是自己的女朋友,那她买票等价于我买票了,就不需要排队了。

3:队列初始化只有一个节点

当只有一个节点的时候,相当于就只有初始化队列的时候new的那个Node。那此时队首和队尾就是它自己,但是它又是不算排队的那个,只是算正在持有锁的那个线程在队列中的一个占位而已。

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

当处于上面讲的第二种情况时,就会执行这段代码。addWaiter()是将当前线程封装成Node,然后添加到AQS队列中。

private Node addWaiter(Node mode) {
    Node node = new Node(mode);

    for (;;) {
        Node oldTail = tail;
        if (oldTail != null) {
            U.putObject(node, Node.PREV, oldTail);
            if (compareAndSetTail(oldTail, node)) {
                oldTail.next = node;
                return node;
            }
        } else {
            initializeSyncQueue();
        }
    }
}

将当前的线程封装成Node并且mode为独占锁。

然后进行一个无限循环,第一次oldTail为null,则进入到else中,进行初始化。第二次进来的时候,oldTail则不为null,将当前线程Node的prev节点指向tail,然后通过cas将node加入AQS队列,成功之后,把旧的tail的next指向新的tail。所以AQS并不是第一的时候就进行初始化。

final boolean acquireQueued(final Node node, int arg) {
    try {
        boolean interrupted = false;
        for (;;) {
            final Node p = node.predecessor();
            if (p == head && tryAcquire(arg)) {
                setHead(node);
                p.next = null; // help GC
                return interrupted;
            }
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } catch (Throwable t) {
        cancelAcquire(node);
        throw t;
    }
}

首先获取前一个节点,只有当前节点的前一个节点是head,表示当前节点在队列中是第二个元素。AQS是根据FIFO来的,所以当前一个释放了锁,只有队列中的第二个才能去获取锁。如果索取锁成功,则吧当前的节点设置为head。并把前一个head断开。因为为head的那个node的node.head 和node.thread都是为null,只需要把node.next置为null时,就断开了链表的链接。

如果获取锁失败,则根据waitStatus决定是否需要挂起线程。

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    int ws = pred.waitStatus;
    if (ws == Node.SIGNAL)
        return true;
        if (ws > 0) {
            do {
                node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        pred.next = node;
    } else {
        pred.compareAndSetWaitStatus(ws, Node.SIGNAL);
    }
    return false;
}

如果前一个节点的waitStatus是SIGNAL,表明当前的的线程需要被unpark。

如果waitStatus > 0,则表明是cancel状态,则从前节点开始逐步循环找到下一个没有被cancel的节点设置为当前节点的前节点。目的将cancel状态的节点踢除掉。如果是其他的状态,将前节点改成SIGNAL状态。当返回true时,继续走下面代码。

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

意思就是通过park()将当前线程挂起到WATING状态。

lock.unlock()
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;
}

执行unLock的时候,会执行release();

protected final boolean tryRelease(int var1) {
    int var2 = this.getState() - var1;
    if (Thread.currentThread() != this.getExclusiveOwnerThread()) {
        throw new IllegalMonitorStateException();
    } else {
        boolean var3 = false;
        if (var2 == 0) {
            var3 = true;
            this.setExclusiveOwnerThread((Thread)null);
        }

        this.setState(var2);
        return var3;
    }
}

getState() -1表示可能有重入的,getState()可能>1,只有当var2 ==0时,把当前线程设置为null,并释放锁,返回true。

private void unparkSuccessor(Node node) {
    int ws = node.waitStatus;
    if (ws  0) {
            s = null;
            for (Node p = tail; p != node && p != null; p = p.prev)
                if (p.waitStatus 

该方法是真正的释放锁,传入的是head节点,当前线程被释放后,需要唤醒下一个节点的线程。

这里是从队列尾部向前遍历找到最前面的一个waitStatus小于0的节点。如果从前面向后遍历会导致死循环。

如果获取到head的next节点不为空,则释放许可唤醒线程。

Condition

在上面讲的是AQS的同步队列,除了这个同步队列外,还有一个条件队列。

加入条件队列的前提是,当前线程已经拿到了锁,并处于运行状态,加入条件队列后,就需要释放锁,进入阻塞状态,同时唤醒同步队列的队二拿锁。唤醒操作是将一个node从条件队列,移动到同步队列的队尾,让它返回同步队列park,并不是随机就唤醒一个线程。

流程

1、创建节点并添加到条件队列的尾部。

2、释放线程所持有的锁。

3、判断是否在同步队列中,在的话,进行CAS拿锁操作,不在的话,唤醒进入同步队列的尾部。

public class ConditionDemo {

    private int queueSize = 10;
    private PriorityQueue queue = new PriorityQueue(queueSize);

    private Lock lock = new ReentrantLock();
    private Condition full = lock.newCondition();
    private Condition empty = lock.newCondition();

    class Consumer implements Runnable{

        @Override
        public void run() {
            consume();
        }

        private void consume() {
            while (true){
                lock.lock();
                try {
                    while (queue.size() == 0){
                        try {
                            System.out.println("队列空,等待数据");
                            empty.await();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                    queue.poll();
                    full.signal();
                    System.out.println("从队列取走一个元素,队列剩余"+queue.size()+"个元素");
                }finally {
                    lock.unlock();
                }
            }
        }
    }

    class Producer implements Runnable{

        @Override
        public void run() {
            produce();
        }

        private void produce() {
            while (true){
                lock.lock();
                try {
                    while(queue.size()== queueSize){
                        try {
                            System.out.println("队列满,等待有空余空间");
                            full.await();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                    queue.offer(1);
                    empty.signal();
                }finally {
                    lock.unlock();
                }
            }

        }
    }
}

上面是利用Condition实现生产者消费者模式,当队列的size为0的时候,empty调用await().

休眠
public final void await() throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    Node node = addConditionWaiter();
    int savedState = fullyRelease(node);
    int interruptMode = 0;
    while (!isOnSyncQueue(node)) {
        LockSupport.park(this);
        if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                    break;
    }
    if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
        interruptMode = REINTERRUPT;
        if (node.nextWaiter != null) // clean up if cancelled
            unlinkCancelledWaiters();
            if (interruptMode != 0)
                reportInterruptAfterWait(interruptMode);
}

在ConditionObject中,包含了两个成员对象,一个表示队列的第一个node,一个表示队列的最后一个node。

private transient Node firstWaiter;
private transient Node lastWaiter;


private Node addConditionWaiter() {
    Node t = lastWaiter;
    if (t != null && t.waitStatus != Node.CONDITION) {
        unlinkCancelledWaiters();
        t = lastWaiter;
    }

    Node node = new Node(Node.CONDITION);

    if (t == null)
        firstWaiter = node;
    else
        t.nextWaiter = node;
    lastWaiter = node;
    return node;
}

addConditionWaiter就是创建一CONDITION状态的个node,如果队列中没有,则把这个新创建的赋值给首节点,如果有,则赋值给当前节点的后一个。并且这个也是最后一个节点。条件队列也是先进先出的形式,只不过是单链表的形式。

final int fullyRelease(Node node) {
    try {
        int savedState = getState();
        if (release(savedState))
            return savedState;
        else    
            throw new IllegalMonitorStateException();
    } catch (Throwable t) {
        node.waitStatus = Node.CANCELLED;
        throw t;
    }
}

首先获取当前的state,然后调用release方法进行释放锁。

在第三篇参考博客写到,fullyRelease会一次性释放所有的锁,不管重入多少次。我理解的是重入的时候,还是同一个线程,同一把锁。所以释放的只是一把锁吧。

经过上面的代码之后,节点已经放到条件队列并且释放了所持有的锁,而后需要挂起。但是在前面说了,挂起需要不在同步队列,所以需要判断是否在同步队列。

final boolean isOnSyncQueue(Node node) {
    if (node.waitStatus == Node.CONDITION || node.prev == null)
        return false;
    if (node.next != null) // If has successor, it must be on queue
        return true;
    return findNodeFromTail(node);
}

private boolean findNodeFromTail(Node node) {
    for (Node p = tail;;) {
        if (p == node)
            return true;
        if (p == null)
            return false;
        p = p.prev;
    }
}

当节点的状态为Node.CONDITION或者node.prev == null时,则不在同步队列,node.prev,node.next在条件队列中没有用到。如果含有这个,则表示是在同步队列中。

private void unlinkCancelledWaiters() {
    Node t = firstWaiter;
    Node trail = null;
    while (t != null) {
        Node next = t.nextWaiter;
        if (t.waitStatus != Node.CONDITION) {
            t.nextWaiter = null;
            if (trail == null)
                firstWaiter = next;
            else
               trail.nextWaiter = next;
            if (next == null)
                lastWaiter = trail;
            }else
                trail = t;
                t = next;
        }
    }

如果状态不是CONDITION,就会自动删除。

唤醒
public final void signal() {
    if (!isHeldExclusively())
        throw new IllegalMonitorStateException();
        Node first = firstWaiter;
        if (first != null)
            doSignal(first);
}

private void doSignal(Node first) {
    do {
        if ( (firstWaiter = first.nextWaiter) == null)
            lastWaiter = null;
            first.nextWaiter = null;
        } while (!transferForSignal(first) &&
             (first = firstWaiter) != null);
        }

首先判断是不是同一个线程,不是则抛出异常。然后获取等待队列的头结点,后续的操作都是基于该节点。

final boolean transferForSignal(Node node) {
    if (!node.compareAndSetWaitStatus(Node.CONDITION, 0))
        return false;

    Node p = enq(node);
    int ws = p.waitStatus;
    if (ws > 0 || !p.compareAndSetWaitStatus(ws, Node.SIGNAL))
        LockSupport.unpark(node.thread);
    return true;
}

private Node enq(Node node) {
    for (;;) {
        Node oldTail = tail;
        if (oldTail != null) {
            U.putObject(node, Node.PREV, oldTail);
            if (compareAndSetTail(oldTail, node)) {
                oldTail.next = node;
                return oldTail;
            }
        } else {
            initializeSyncQueue();
        }
    }
}

enq()是将该节点移入到同步队列中去。

signal是只会对等待队列的头结点进行操作,而signalAll则是对每一个节点都移入到同步队列中。

同步队列与条件队列完整配合如下图所示:

1608883350(1).png
参考文献

https://blog.csdn.net/java_lyvee/article/details/98966684

https://segmentfault.com/a/1190000017372067

https://segmentfault.com/a/1190000020345054

0

评论0

鱼翔浅底,鹰击长空,驼走大漠
没有账号? 注册  忘记密码?