源码分析之AbstractQueuedSynchronizer

java.util.concurrent包中,大部分的同步器都是基于AbstractQueuedSynchronizer(AQS)这个框架实现的。这个框架为同步状态提供原子性管理、线程的阻塞和解除阻塞以及排队提供了一种通用机制。java

同步器通常包含2种方法,一种是acquire,另外一种是releaseacquire操做阻塞线程,获取锁。release经过某种方式改变让被acquire阻塞的线程继续执行,释放锁。为了实现这2种操做,须要如下3个基本组件的相互协做:node

  • 同步状态的原子性管理
  • 线程的阻塞和解除阻塞
  • 队列管理

同步状态

/** * The synchronization state. */
    private volatile int state;复制代码

AQS使用一个int变量来保存同步状态,并暴露出getStatesetState以及compareAndSet来读取或更新这个状态。而且用了volatile来修饰,保证了在多线程环境下的可见性。经过使用compare-and-swap(CAS)指令来实现compareAndSetStateapi

这里的同步状态用int而非long,主要是由于64位long字段的原子性操做在不少平台上是使用内部锁的方式来模拟实现的,这会使得同步器的会有性能问题。绝对多数int型的state足够咱们使用,但JDK也提供了longstate的实现:java.util.concurrent.locks.AbstractQueuedLongSynchronizer安全

阻塞

JDK1.5以前,阻塞线程和解除线程阻塞都是基于Java自身的监控器。在AQS中实现阻塞是用java.util.concurrent包的LockSuport类。方法LockSupport.park阻塞当前线程,直到有个LockSupport.unpark方法被调用。多线程

队列管理

AQS框架关键就在于如何管理被阻塞的线程队列。提供了2个队列,分别是线程安全Sync Queue(CLH Queue)、普通的Condition Queue并发

Sync Queue

Sync Queue是基于FIFO的队列,用于构建锁或者其余相关同步装置。CLH锁能够更容易地去实现取消(cancellation)超时功能,所以咱们选择了CLH锁做为实现的基础。框架

队列中的元素Node是保存线程的引用和线程状态。NodeAQS的一个静态内部类:性能

static final class Node {
        static final Node SHARED = new Node();
        static final Node EXCLUSIVE = null;
        static final int CANCELLED =  1;
        static final int SIGNAL    = -1;
        static final int CONDITION = -2;
        static final int PROPAGATE = -3;
        volatile int waitStatus;
        volatile Node prev;
        volatile Node next;
        volatile Thread thread;
        Node nextWaiter;
    }复制代码

Node类的成员变量如上所示,主要负责保存线程引用、队列的前继和后继节点,以及同步状态:ui

成员 描述
waitStatus 用来标记Node的状态:
CANCELLED:1, 表示当前线程已经被取消
SIGNAL:-1,表示当前节点的后继节点等待运行
CONDITION:-2, 表示当前节点已被加入Condition Queue
PROPAGATE:-3, 共享锁的最终状态是PROPAGATE
thread 当前获取lock的线程
SHARED 表示节点是共享模式
EXCLUSIVE 表示节点是独占模式
prev 前继节点
next 后继节点
nextWaiter 存储Condition Queue中的后继节点

Node元素是Sync Queue构建的基础。当获取锁的时候,请求造成节点挂载在尾部。而锁资源的释放再获取的过程是从开始向后进行的。
this

acquire 获取锁

AQS自身仅定义了相似acquire方法。在实现锁的时候,通常会实现一个继承AQS的内部类Sync。而在Sync类中,咱们根据需求来实现重写tryAcquire方法和tryRelease方法。独占锁acquire方法以下:

public final void acquire(int arg) {

        if (!tryAcquire(arg) &&
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }复制代码
  • 经过tryAcquire(由不一样的实现类实现)尝试获取锁,若是能够获取锁直接返回。获取不到锁,则调用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;
    }

        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;
                }
            }
        }
    }复制代码
  • addWaiter方法做用是把当前线程封装成Node节点,经过CAS操做快速尝试挂载至队列尾部。
    • 若是tail节点t已经有了:将t节点更新为当前节点node的前继节点node.prev,将t.next更新为当前节点node
    • 若是tail节点添加失败:
      • 若是tail节点为空,那么原子化的分配一个头节点,并将尾节点指向头节点,这一步是初始化;
      • 若是tail节点不为空,循环重复addWaiter方法的工做直至当前节点入队为止。
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);
        }
    }复制代码
  • 节点加入Sync Queue以后,接下来就是要进行锁的获取,或者说是访问控制了,只有一个线程可以在同一时刻继续的运行,而其余的进入等待状态。
    • 获取当前节点的前继节点
    • 当前继节点是头结点而且可以获取状态,表明该当前节点占有锁;若是知足上述条件,那么表明可以占有锁,根据节点对锁占有的含义,设置头结点为当前节点。
    • 不然进入等待状态。

至此,能够总结一次acquire的过程大体为:

release 释放锁
public final boolean release(int arg) {
        if (tryRelease(arg)) {
            Node h = head;
            if (h != null && h.waitStatus != 0)
                unparkSuccessor(h);
            return true;
        }
        return false;
    }复制代码
  • 首先经过CAS操做变动同步状态state
  • 释放成功后,经过LockSupport.unpark方法来唤醒后继节点,后继节点继续获取锁。

Condition Queue

AQS框架提供了一个ConditionObject内部类,给维护独占同步的类以及实现Lock接口的类使用。一个锁对象能够关联任意数目的条件对象,能够提供典型的Java监视器风格的awaitsignalsignalAll操做,包括带有超时的,以及一些检测、监控的方法。Condition Queue是普通的队列并不要求是线程安全,缘由是在线程在操做Condition时,要求线程必须独占锁,不须要考虑并发的问题。

Condition Queue也是以Node为基础的队列。

/** First node of condition queue. */
        private transient Node firstWaiter;
        /** Last node of condition queue. */
        private transient Node lastWaiter;复制代码
await操做
  • Condition在执行await操做时,首先会调用addConditionWaiter()方法将当前线程封装的Node节点加入到wait queue
private Node addConditionWaiter() {
            Node t = lastWaiter;
            // If lastWaiter is cancelled, clean out.
            if (t != null && t.waitStatus != Node.CONDITION) {
                unlinkCancelledWaiters();
                t = lastWaiter;
            }
            Node node = new Node(Thread.currentThread(), Node.CONDITION);
            if (t == null)
                firstWaiter = node;
            else
                t.nextWaiter = node;
            lastWaiter = node;
            return node;
        }复制代码

上述addConditionWaiter的逻辑是:

  • 首先清除Condition Queue队列中cancelled状态的尾节点;
  • Condition Queue队列为空,封装当前线程的node节点为Condition QueuefirstWaiter。如Condition Queue队列不为空,则把该节点加至队列尾部。
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);
        }复制代码
  • 加入Condition queue以后,要释放当前线程获取的全部的锁;
  • 若是线程没有在Sync Queue中,将调用LockSupport.park阻塞当前线程,直到signalled或者interrupted唤醒去获取锁。
single 操做
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);
        }复制代码
  • 首先检查线程是否独占锁;
  • 获取Codition QueuefirstWaiter,将节点转移至Sync Queue中去。
singleAll 操做
private void doSignalAll(Node first) {
            lastWaiter = firstWaiter = null;
            do {
                Node next = first.nextWaiter;
                first.nextWaiter = null;
                transferForSignal(first);
                first = next;
            } while (first != null);
        }复制代码

signalAll唤醒Condition Queue的全部等待线程,将全部的Condition Queue中的node元素转移至Sync Queue中去。

其余API

这里只介绍了独占锁模式下,普通acquirerelease方法的原理,AQS还提供了不少能够供咱们选择的API

  • 如优先考虑中断、超时的:acquireInterruptiblytryAcquireNanos
  • 如共享锁模式下acquireSharedreleaseShared

等等...

这里暂时不详细分析,后面有时间的话能够再作了解。

参考:

相关文章
相关标签/搜索