Java并发编程 AbstractQueuedSynchronizer的设计与实现

Posted 玉树临枫

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Java并发编程 AbstractQueuedSynchronizer的设计与实现相关的知识，希望对你有一定的参考价值。

一 前言

　　上一篇分析AQS的内部结构，其中有介绍AQS是什么，以及它的内部结构的组成，那么今天就来分析下前面说的内部结构在AQS中的具体作用（主要在具体实现中体现）。

二 AQS的接口和简单示例

　　上篇有说到AQS是抽象类，而它的设计是基于模板方法模式的，也就是说：使用者需要继承同步器并重写指定的方法，随后将同步器组合在自定义同步组件的实现中，并调用其提供的模板方法。其中需要子类重写的方法与描述如下表：

方法名称 描述
protected boolean tryAcquire(int arg)

尝试以独占模式获取。 此方法应查询对象的状态是否允许以独占模式获取它，如果是，则获取它。

实现该方法需要查询当前状态并判断同步状态是否预期，然后进行CAS设置同步状态。
protected boolean tryRelease(int arg)

尝试释放独占式的同步状态。

等待获取同步状态的线程将有机会获取同步状态。
protected int tryAcquireShared(int arg)

尝试以共享模式获取。 此方法应查询对象的状态是否允许在共享模式下获取它，如果是，则获取。

实现该方法需要查询当前状态并判断同步状态是否预期，然后进行CAS设置同步状态。
protected boolean tryReleaseShared(int arg)

尝试释放共享式的同步状态。
protected boolean isHeldExclusively() 表示当前同步器是否在独占模式下被线程占用。

　　在重写上面这些方法时，可能需要下面这三个方法（注意其中state是使用volatile关键字修饰的）

方法名 描述
protected final int getState()  获取当前的同步状态
protected final void setState(int newState)  设置当前同步状态
protected final boolean compareAndSetState
(int expect, int update)		 使用CAS设置当前状态，该方法能保证状态设置的原子性

　　其实前面这些都不需要关心，因为这些一般都是在自定义同步组件中实现。自定义同步组件除了重写第一个表格那些方法外，AQS还为其提供了一些公共方法（或者说模板方法），这些才是关键，也是重中之重。下面我先简单列出以及其方法描述，后面一一分析：

方法名称 描述
public final void acquire(int arg)

独占式获取同步状态，忽略中断。

如果当前线程获取同步状态成功，则由该方法返回；否则将会进入同步队列等待（

即上篇说的Node节点队列）。

该方法将会调用重写的tryAcquire(int args)方法。
public final void
acquireInterruptibly(int arg)

与acquire（int args）方法一样，但是该方法响应中断（从方法名就大概知道意思了吧。）

当前线程未获取到同步状态而进入同步队列中，如果当前线程被中断，则该方法会抛出InterruptedException异常
public final boolean release(int arg)

独占式的释放同步状态， 该方法会在释放同步状态后将同步队列中第一个节点包含的线程唤醒。

该方法会调用tryRelease（int args）方法
public final void acquireShared(int arg)

共享式获取同步状态，忽略中断。

如果当前线程获取同步状态成功，则由该方法返回；否则将会进入同步队列等待

（即上篇说的Node节点队列）。

与独占式获取的主要区别是在同一时刻可以有多个线程获取到同步状态。

该方法将会调用重写的tryAcquireShare(int args)方法。
public final void acquireSharedInterruptibly(int arg) 与acquireInterruptibly方法相同
public final boolean
releaseShared(int arg)		  共享式的释放同步状态

该方法会调用tryReleaseShared（int args）方法

　　根据上面提供的方法，同步器主要提供两种模式：独占式和共享式。顾名思义，独占式表示同一时刻只有一个线程才会获取到同步状态，而其他线程都得等待；而共享式就允许同一时刻可以多个线程获取到同步状态。至于示例的话，大家可以查看源码类上注释的Mutx类，表示一个自定义的独占锁。下面我还是直接贴出示例代码。

class Mutex implements Lock, java.io.Serializable {
    // 内部类，自定义同步器
    private static class Sync extends AbstractQueuedSynchronizer {
        // 是否处于占用状态
        protected boolean isHeldExclusively() {
            return getState() == 1;
        }
        // 当状态为0的时候获取锁
        public boolean tryAcquire(int acquires) {
            assert acquires == 1; // Otherwise unused
            if (compareAndSetState(0, 1)) {
                // 将当前线程设置为独占线程
                setExclusiveOwnerThread(Thread.currentThread());
                return true;
            }
            return false;
        }
        // 释放锁，将状态设置为0
        protected boolean tryRelease(int releases) {
            assert releases == 1; // 断言
            if (getState() == 0) throw new IllegalMonitorStateException();
            // 将线程或状态 重置为初始值
            setExclusiveOwnerThread(null);
            setState(0);
            return true;
        }
        // 返回一个Condition，每个condition都包含了一个condition队列
        Condition newCondition() { return new ConditionObject(); }
    }
    // 仅需要将操作代理到Sync上即可
    private final Sync sync = new Sync();
    public void lock()                { sync.acquire(1); }
    public boolean tryLock()          { return sync.tryAcquire(1); }
    public void unlock()              { sync.release(1); }
    public Condition newCondition()   { return sync.newCondition(); }
    public boolean isLocked()         { return sync.isHeldExclusively(); }
    public boolean hasQueuedThreads() { return sync.hasQueuedThreads(); }
    public void lockInterruptibly() throws InterruptedException {
        sync.acquireInterruptibly(1);
    }
    public boolean tryLock(long timeout, TimeUnit unit)
            throws InterruptedException {
        return sync.tryAcquireNanos(1, unit.toNanos(timeout));
    }
}
View Code

　　看了下自定义的独占锁Metux（上面代码来自源码），写个案例测试下它到底是否是独占锁（大家应该知道怎么测试吧）。

public class MutexTest {

    private Lock lock ;
    private MutexTest(Lock lock) {
        this.lock = lock;
    }

    public void runTask() {
        lock.lock();
        try {
            System.out.println(Thread.currentThread().getName() + " 执行任务中...");
            Thread.sleep(3000);
            System.out.println(Thread.currentThread().getName() + " 任务执行完成。");
        } catch (Exception e){
            e.printStackTrace();
        }finally {
            lock.unlock();
        }
    }

    public static void main(String[] args) {
        Lock lock = new Mutex();
        final MutexTest test = new MutexTest(lock);
        for (int i = 0; i < 5; i ++) {
            new Thread(new Runnable() {
                @Override
                public void run() {
                    test.runTask();
                }
            }, "线程" + i).start();
        }
    }
}
View Code

　　运行该案例从打印结果中可以看出，同一时刻只有一个线程在执行（这就是独占锁的特性）。

线程0 执行任务中...
线程0 任务执行完成。
线程2 执行任务中...
线程2 任务执行完成。
线程1 执行任务中...
线程1 任务执行完成。
线程3 执行任务中...
线程3 任务执行完成。
线程4 执行任务中...
线程4 任务执行完成。

三 AQS的核心函数分析

　　关于获取和释放下面只分析acquire函数和release函数，因为其他都与这个函数类似。

1、acquire函数

    /**
     * 独占式获取同步状态，忽略中断。
     */
    public final void acquire(int arg) {
        /**
         * 1 调用子类的tryAcquire(arg)方法，如果获取成功则直接返回，否则以独占模式创建节点加入等待队列
         */
        if (!tryAcquire(arg) &&
                acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }

　　acquire函数主要功能都放在这三个方法中:

  • tryAcquire(arg) 子类提供实现
  • addWaiter(Node) 主要是将节点添加到等待队列中。
  • acquireQueue(Node, int) 主要是提取等待队列中能获取同步状态的节点（遵循FIFO）。

　　1.2 下面先分析下addWaiter(Node)函数：

/**
 * 2 根据给定模式为当前线程创建并排队节点。
 */
private Node addWaiter(Node mode) {
    // 2.1 根据指定模式和当前线程创建节点。（在这就用的上Node了）
    Node node = new Node(Thread.currentThread(), mode);
    // 2.2 尝试下快速通道：判断tail节点是否为空，如果不为空就直接添加到尾节点后面。
    Node pred = tail;
    if (pred != null) {
        node.prev = pred;
        // 2.2.1 进入到这个方法说明线程并没有获取锁，所以需要CAS保证原子性。
        if (compareAndSetTail(pred, node)) {
            pred.next = node;
            return node;
        }
    }
    // 2.3 如果是第一个入队的节点或者compareAndSetTail设置失败，那么就进入enq()方法
    enq(node);
    return node;
}
/**
 * 将节点插入队列，必要时进行初始化。
 */
private Node enq(final Node node) {
    // 自旋，直至设置添加尾节点成功。
    for (;;) {
        Node t = tail;
        if (t == null) {
            // 2.3.1 尾节点为空，则需要初始化队列（同理采取CAS保证原子性）
            if (compareAndSetHead(new Node()))
                tail = head;
        } else {
            // 2.3.2 尾节点不为空，则将节点设置成尾节点（同理采取CAS保证原子性）
            node.prev = t;
            if (compareAndSetTail(t, node)) {
                t.next = node;
                return t;
            }
        }
    }
}

　　上述逻辑主要包括：使用当前线程创建节点，然后将当前节点添加到同步队列中。其中设置节点都是利用CAS设置，保证原子性。

　　具体流程：

　　a 先行尝试在队尾添加（如果尾节点不为空）（另外这一步很重要，如果尾节点存在就可以以最短路径O(1)的效果来完成线程入队，是最大化减少开销的一种方式）:

    • 分配引用prev指向尾节点；
    • 将节点的前驱节点更新为尾节点（current.prev = tail）；
    • 如果尾节点是prev，那么将当尾节点设置为该节点（tail = current，原子更新）；
    • prev的后继节点指向当前节点（prev.next = current）。

　　b 如果是第一个入队的节点或者compareAndSetTail设置失败：

    • 如果尾节点为空，则需要初始化队列（同理采取CAS保证原子性），继续自旋判断；

    •  重复上面a步骤将节点尝试添加至尾节点后，直接添加成功。　　　　　　

　　1.3 进入sync队列之后，接下来就是要进行同步状态的获取，下面请看acquireQueue(Node, arg)函数：　

/**
 * 3 对于已经在队列中的线程，以独占不间断模式获取。
 */
final boolean acquireQueued(final Node node, int arg) {
    boolean failed = true;
    try {
        boolean interrupted = false;
        // 同样采取自旋直至条件满足
        for (;;) {
            // 3.1 获取当前节点的前驱节点
            final Node p = node.predecessor();
            // 3.2 判断前驱节点是否为头节点，并此时是否可以获取到同步状态
            if (p == head && tryAcquire(arg)) {
                // 3.2.1 若上面条件满足，则将当前节点设置为头节点。
                setHead(node);
                p.next = null; // help GC
                failed = false;
                return interrupted;
            }
            if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

　　上述逻辑主要包括：

    • 获取当前节点的前驱节点；需要获取当前节点的前驱节点，而头结点所对应的含义是当前占有锁且正在运行。
    • 当前驱节点是头结点并且能够获取状态，代表该当前节点占有锁；

　　　　　　如果满足上述条件，那么代表能够占有锁，根据节点对锁占有的含义，设置头结点为当前节点。

    • 否则进入等待状态。

　　　　　　如果没有轮到当前节点运行，那么将当前线程从线程调度器上摘下，也就是进入等待状态。也就是调用shouldParkAfterFailedAcquire和parkAndCheckInterrupt函数

 　　下面看下它是怎么将不满足节点摘下来进入等待状态的。

/**
 * 检查并更新获取失败的节点的状态。
 */
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    int ws = pred.waitStatus;
    if (ws == Node.SIGNAL)
        /*
         * 状态处于SIGNAL状态(-1)，表示后继节点随时可以upark
         */
        return true;
    if (ws > 0) {
        /*
         * ws > 0表示处于CANCELLED状态，则需要跳过找到node节点前面不处于取消状态的节点。
         */
        do {
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        pred.next = node;
    } else {
        /*
         * 此时ws为PROPAGATE -3 或者是0 表示无状态,(为CONDITION -2时，表示此节点在condition queue中)
         * 比较并设置前驱结点的状态为SIGNAL
         */
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    /**
     * 此时还不确定Node的前置节点是否处于SIGNAL状态
     * 所以不支持park操作
     */
    return false;
}

/**
 * 进行park操作并且返回该线程是否被中断
 */
private final boolean parkAndCheckInterrupt() {
    LockSupport.park(this);
    return Thread.interrupted();
}

 　　上述主要逻辑包括：

    • 如果前置节点的状态是Signal状态，则返回true。
    • 如果前置节点处于取消状态，则跳过这种取消节点，找到不是前面不是取消状态的节点然后返回false;
    • 如果前置节点处于<0的状态，则利用CAS将其状态设置成Signal状态，然后返回false.
    • 经过上面步骤后，如果返回true，则说明可以中断线程进入等待。

 　　那么acquire函数分析到这就结束了，估计看了一遍还是不太清晰流程那么就多看几遍。下面也对这个流程进行总结下：

2、release函数　　

/**
 * 以独占模式释放
 */
public final boolean release(int arg) {
    // tryRelease由子类实现
    if (tryRelease(arg)) {
        // 获取头结点
        Node h = head;
        // 头结点不为空并且头结点状态不为0
        if (h != null && h.waitStatus != 0)
            // 释放头结点的后继结点
            unparkSuccessor(h);
        return true;
    }
    return false;
}

/**
 * 唤醒后继节点
 */
private void unparkSuccessor(Node node) {
    // 获取节点状态
    int ws = node.waitStatus;
    // 如果节点状态小于0，则将其设置为初始状态。
    if (ws < 0)
        compareAndSetWaitStatus(node, ws, 0);

    // 如果节点状态是取消或节点为空，则从尾部向后移动以找到实际未取消的继任者。
    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);
}

　　上述逻辑主要包括：

    • 尝试释放状态；

　　　　　　tryRelease能够保证原子化的将状态设置回去，当然需要使用compareAndSet来保证。如果释放状态成功过之后，将会进入后继节点的唤醒过程。

    •  唤醒当前节点的后继节点所包含的线程。

　　　　　　通过LockSupport的unpark方法将休眠中的线程唤醒，让其继续acquire状态。

四 总结（获取与释放过程）

  1. 在获取时，维护了一个sync队列，每个节点都是一个线程在进行自旋，而依据就是自己是否是首节点的后继并且能够获取资源；（重点，不清楚的可以看上面的流程图）
  2. 在释放时，仅仅需要将资源还回去，然后通知一下后继节点并将其唤醒。
  3. 这里需要注意，队列的维护（首节点的更换）是依靠消费者（获取时）来完成的，也就是说在满足了自旋退出的条件时的一刻，这个节点就会被设置成为首节点。

 

　　另外送大家一碗心灵鸡汤：）

我从不相信什么懒洋洋的自由，我向往的自由是通过勤奋和努力实现更广阔的人生，那样的自由才是珍贵的、有价值的。我相信一万小时定律，我从来不相信天上掉馅饼的灵感和坐等的成就。做一个自由又自律的人，靠势必实现的决心认真地活着。

以上是关于Java并发编程 AbstractQueuedSynchronizer的设计与实现的主要内容，如果未能解决你的问题，请参考以下文章

Java 并发编程：核心理论

Java并发编程之美

『死磕Java并发编程系列』并发编程工具类之CountDownLatch

Java并发编程：Synchronized及其实现原理

Java并发指南开篇：Java并发编程学习大纲

Java并发编程：Synchronized及其实现原理