多线程知识梳理(11) - 队列同步器实现原理 & 应用

2019-05-09 本文已影响17人泽毛

一、基本概念

队列同步器在Java并发包中的实现是AbstractQueuedSynchronizer，简称为AQS，它是用来构建锁或者其它同步组件的基础框架。了解其实现原理有助于：

理解同步组件ReentrantLock和ReentrantReadWriteLock的原理
理解Condition实现等待通知/模式的原理
根据业务场景，自定义同步组件（较少用到）

队列同步器AQS和同步组件（ReentrantLock、ReentrantReadWriteLock等）的区别在于：

同步组件面向使用者，它定义了使用者与锁交互的接口，隐藏了实现细节。
队列同步器面向锁的实现者，它简化了锁的实现方式，屏蔽了同步状态管理、线程排队、等待与唤醒等底层操作。

队列同步器基于模板方法模式，它的使用方式为如下：

创建自定义同步组件的实现类。
实现者继承AbstractQueuedSynchronizer（一般会将它作为自定义同步组件的内部类），根据业务场景，重写指定的模板方法tryAcquire(int acquires)、tryRelease(int releases)等。
创建同步器子类的对象，作为自定义同步组件的成员变量，在对外提供的公共方法中，调用队列同步器的指定方法。

以ReentrantLock为例，ReentrantLock为自定义的同步组件，NonfairSync和FairSync就是队列同步器的实现类。

public class ReentrantLock implements Lock, java.io.Serializable {

    private final Sync sync;

    abstract static class Sync extends AbstractQueuedSynchronizer {
        //...
    }
    
    static final class NonfairSync extends Sync {
        //...
    }
    
    static final class FairSync extends Sync {
        //...
    }
    
    public void lock() {
        sync.lock();
    }

}

二、同步状态 & 同步队列

在队列同步器中有两个关键的元素：

同步状态
同步队列

2.1 同步状态

同步状态，用一个int型表示，访问或者修改同步状态需要使用指定的方法：

getState()：获取当前同步状态。
setState(int newState)：设置当前同步状态。
compareAndSetState(int except, int update)：使用CAS设置当前状态，该方法可以保证状态设置的原子性。

2.2 同步队列

同步队列用来完成同步状态的管理，当前线程获取同步状态失败时，会将当前线程以及等待状态等信息构造成一个结点，将其加入同步队列，同时会阻塞当前线程，当同步状态释放时，会把首结点的后继结点的线程唤醒，使其再次尝试获取同步状态。

同步队列中的结点保存了以下的信息：

thread：线程引用
waitStatus：等待状态
- CANCELLED：由于在同步队列中等待的线程等待超时或者被中断，需要从同步队列中取消等待。
- SIGNAL：后继结点的线程处于等待状态，而当前结点的线程如果释放了同步状态或取消，将会通知后继结点，使后继结点的线程运行。
- CONDITION：结点在等待队列中，结点线程等待在Condition上，当其他线程对Condition调用了signal方法后，该结点会从等待队列移到同步队列中。
- PROPAGATE：表示下一次共享式同步状态获取将会无条件传播下去。
- INITIAL：初始状态。
prev & next：前驱 & 后继结点

同步器包含了两个类型结点的引用：头结点和尾结点。设置的区别在于：

头结点是获取同步状态成功的结点，头结点的线程在释放同步状态时，将会唤醒后继结点，而后继结点将会在获取同步状态成功时将自己设置成头结点。由于头结点是通过获取同步状态成功的线程来完成的，因此设置头结点的方式不需要CAS来保证。
如果已经有一个线程获取了同步状态，那么其他线程无法获取同步状态，就会构造结点假如到同步队列当中，为了保证线程安全，因此需要采用compareAndSetTail(Node expect, Node update)的CAS方式来设置尾结点。

同步队列

三、同步器提供的模板方法

独占式获取 & 释放同步状态
共享式获取 & 释放同步状态
独占式超时获取同步状态

3.1 独占式同步状态获取与释放

3.1.1 获取

通过调用同步器的acquire(int arg)方法可以获取同步状态，该方法对中断不敏感。当同步状态获取成功后，当前线程从acquire(int arg)方法返回，如果对于锁这种并发组件，代表着当前线程获取了锁。其实现为：

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

我们将其工作拆解称为三个部分：

Step 1 tryAcquire(arg)

调用 自定义同步器 实现的tryAcquire(int arg)方法，该方法需要保证线程安全的获取同步状态。

Step 2 addWaiter(Node.EXCLUSIVE), arg)

如果同步状态获取失败（tryAcquire方法返回false），则构造同步结点，并通过addWaiter(Node node)方法加入到同步队列的尾部。为了保证能够线程安全地添加，它采用了两层保护机制：

CAS：compareAndSetTail。
死循环：enq。

    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;
            //1.确保节点能够线程安全地被添加
            if (compareAndSetTail(pred, node)) {
                pred.next = node;
                return node;
            }
        }
        //2.通过死循环来确保节点的正确添加，在"死循环"中只有通过`CAS`将节点设置为尾节点之后，当前线程才能从该方法返回，否则当前线程不断地进行尝试。
        enq(node);
        return node;
    }

Step 3 acquireQueued(Node node, int arg)

调用acquireQueued(Node node, int arg)方法，使得新构造的同步结点以“死循环”的方式获取同步状态。

    final boolean acquireQueued(final Node node, int arg) {
        boolean failed = true;
        try {
            boolean interrupted = false;
            for (;;) {
                //1.1 得到当前节点的前驱节点
                final Node p = node.predecessor();
                //1.2 如果当前节点的前驱节点是头节点，只有在这种情况下获取同步状态成功
                if (p == head && tryAcquire(arg)) {
                    //1.3 将当前节点设为头节点
                    setHead(node);
                    p.next = null; // help GC
                    failed = false;
                    return interrupted;
                }
                //2 前驱结点不是头结点，或者无法获取到同步状态的情况。
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

这里需要注意的有以下几点：

只有前驱结点是头结点时，才会尝试再次调用tryAcquire(int arg)来获取同步状态。这么做是为了维护同步队列的FIFO原则，并且也便于对过早通知的处理。
如果当前结点成功获取到同步状态，那么会通过setHead方法将其设为新的头结点，并断开旧的头结点与后继结点的关系（p.next = null）。
如果无法获取到同步状态，那么会通过shouldParkAfterFailedAcquire判断是否要调用parkAndCheckInterrupt进入阻塞状态。waitStatus的默认值是0，因此第一次会进入最后一个判断，并将其waitStatue设置为Node.SIGNAL，之后进入第二次的for循环，假如其前驱结点仍然不是头结点或者无法获取到同步状态，那么将会进入第一个判断，阻塞当前线程。

    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 {
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        }
        return false;
    }
    
    private final boolean parkAndCheckInterrupt() {
        //阻塞当前线程。
        LockSupport.park(this);
        //如果当前线程已经被中断，那么返回 true。
        return Thread.interrupted();
    }

如果当前线程在获取同步状态时，已经被中断，即interrupted=true，那么会调用selfInterrupt方法。

3.1.2 释放

当前线程获取同步状态并执行了相应的逻辑后，就需要释放同步状态，并调用unparkSuccessor来唤醒后继结点。

  public final boolean release(int arg) {
      if (tryRelease(arg)) {
          Node h = head;
          if (h != null && h.waitStatus != 0)
              unparkSuccessor(h);
          return true;
      }
      return false;
  }
  
  private void unparkSuccessor(Node node) {
      int ws = node.waitStatus;
      if (ws < 0)
          node.compareAndSetWaitStatus(ws, 0);
      Node s = node.next;
      if (s == null || s.waitStatus > 0) {
          s = null;
          for (Node p = tail; p != node && p != null; p = p.prev)
              if (p.waitStatus <= 0)
                  s = p;
      }
      if (s != null)
          LockSupport.unpark(s.thread);
  }

3.1.3 小结

由以上两点分析，可以看出独占式获取 & 释放同步状态采用了模板方法的设计模式，AQS内部会基于当前线程创建一个结点，并负责管理线程的入队和出队，线程的阻塞和唤醒，我们只需要重写重要的模板方法：

protected boolean tryAcquire(int arg)：独占式获取同步状态，实现该方法需要查询当前状态并判断同步状态是否符合预期，然后再进行CAS设置同步状态。
protected boolean tryRelease(int arg)：独占式释放同步状态。

下面是一个独占式获取 & 释放同步状态的简单实现示例：

/**
 * @author lizejun
 **/
public class ExclusiveLock implements Lock {

    private static class Sync extends AbstractQueuedSynchronizer {

        protected boolean isHeldExclusively() {
            return getState() == 1;
        }

        @Override
        protected boolean tryAcquire(int acquires) {
            if (compareAndSetState(0, 1)) {
                setExclusiveOwnerThread(Thread.currentThread());
                return true;
            }
            return false;
        }

        @Override
        protected boolean tryRelease(int releases) {
            if (getState() == 0) {
                throw new IllegalMonitorStateException();
            }
            setExclusiveOwnerThread(null);
            setState(0);
            return true;
        }

        Condition newCondition() {
            return new ConditionObject();
        }
    }

    private final Sync sync = new Sync();

    @Override
    public void lock() {
        sync.acquire(1);
    }

    @Override
    public boolean tryLock() {
        return sync.tryAcquire(1);
    }

    @Override
    public void unlock() {
        sync.release(1);
    }

    @Override
    public void lockInterruptibly() throws InterruptedException{
        sync.acquireInterruptibly(1);
    }

    @Override
    public boolean tryLock(long timeout, TimeUnit unit) throws InterruptedException {
        return sync.tryAcquireNanos(1, unit.toNanos(timeout));
    }

    @Override
    public Condition newCondition() {
        return sync.newCondition();
    }

    public boolean isLocked() {
        return sync.isHeldExclusively();
    }

    public boolean hasQueuedThreads() {
        return sync.hasQueuedThreads();
    }


    public static void run() {
        final ExclusiveLock lock = new ExclusiveLock();
        for (int i = 0; i < 3; i++) {
            final int index = i;
            new Thread() {

                @Override
                public void run() {
                    lock.lock();
                    System.out.println("begin:" + index);
                    try {
                        System.out.println("run:" + index);
                        Thread.sleep(500);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } finally {
                        System.out.println("end:" + index);
                        lock.unlock();
                    }
                }

            }.start();
        }
    }
}

运行结果为如下，可以看到，处于lock ~ unlock之间的逻辑在同一时间只能由一个线程执行。

独占锁实现示例

假如我们注释掉lock.lock()和lock.unlock()，那么打印的结果为，这时不能保证是线程安全的。

独占锁关闭

3.2 共享式获取 & 释放同步状态

3.2.1 获取

共享式获取与独占式获取最主要的区别在于 同一时刻能否有多个线程同时获取到同步状态。

调用同步器的acquireShared(int arg)方法可以共享式地获取同步状态。锁的实现者需要重写tryAcquireShared方法，如果该方法的返回值大于0，表示能够获取到同步状态。

    public final void acquireShared(int arg) {
        if (tryAcquireShared(arg) < 0)
            doAcquireShared(arg);
    }

    private void doAcquireShared(int arg) {
        final Node node = addWaiter(Node.SHARED);
        boolean failed = true;
        try {
            boolean interrupted = false;
            for (;;) {
                final Node p = node.predecessor();
                if (p == head) {
                    //1.如果前驱结点是头结点，那么会尝试获取同步状态。
                    int r = tryAcquireShared(arg);
                    if (r >= 0) {
                        //2.获取同步状态成功，设为头结点。
                        setHeadAndPropagate(node, r);
                        p.next = null; // help GC
                        if (interrupted)
                            selfInterrupt();
                        failed = false;
                        return;
                    }
                }
                //3.获取同步状态失败，那么阻塞当前线程。
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

3.2.2 释放

通过调用releaseShared(int args)方法可以释放同步状态，该方法在释放同步状态后，将会唤醒后续处于阻塞状态的结点。对于支持多个线程同时访问的并发组件，它和独占式的区别在于，tryReleaseShared(int arg)方法必须保证同步状态安全释放，一般是通过循环和CAS来保证的，因为释放同步状态的操作会同时来自多个线程。

    public final boolean releaseShared(int arg) {
        if (tryReleaseShared(arg)) {
            doReleaseShared();
            return true;
        }
        return false;
    }

3.2.3 小结

与独占式获取 & 释放同步状态类似，在继承AQS时主要关注两个重要的模板方法：

protected int tryAcquireShared(int arg)：共享式获取同步状态，返回值>=0表示获取成功，否则失败。
protected boolean tryReleaseShared(int arg)：共享式释放同步状态。

下面是一个简单地共享式获取 & 释放同步状态的示例：

/**
 * @author lizejun
 **/
public class TwinsLock implements Lock {

    private Sync sync = new Sync(2);

    private static final class Sync extends AbstractQueuedSynchronizer {

        Sync(int count) {
            setState(count);
        }

        @Override
        protected int tryAcquireShared(int reduceCount) {
            for (;;) {
                int current = getState();
                int newCount = current - reduceCount;
                if (newCount < 0 || compareAndSetState(current, newCount)) {
                    return newCount;
                }
            }
        }

        @Override
        protected boolean tryReleaseShared(int returnCount) {
            for (;;) {
                int current = getState();
                int newCount = current + returnCount;
                if (compareAndSetState(current, newCount)) {
                    return true;
                }
            }
        }

        Condition newCondition() {
            return new ConditionObject();
        }
    }

    @Override
    public void lock() {
        sync.acquireShared(1);
    }

    @Override
    public void unlock() {
        sync.releaseShared(1);
    }

    @Override
    public void lockInterruptibly() throws InterruptedException {
        sync.acquireSharedInterruptibly(1);
    }

    @Override
    public boolean tryLock(long timeout, TimeUnit unit) throws InterruptedException {
        return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
    }

    @Override
    public Condition newCondition() {
        return sync.newCondition();
    }

    @Override
    public boolean tryLock() {
        return sync.tryAcquireShared(1) >= 0;
    }

    public static void run() {
        final TwinsLock lock = new TwinsLock();
        for (int i = 0; i < 5; i++) {
            final int index = i;
            new Thread() {

                @Override
                public void run() {
                    lock.lock();
                    System.out.println("begin:" + index);
                    try {
                        System.out.println("run:" + index);
                        Thread.sleep(500);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } finally {
                        System.out.println("end:" + index);
                        lock.unlock();
                    }
                }

            }.start();
        }
    }
}

运行结果为如下，可以看到在同一时刻只有两个线程能够获取到锁：

共享锁实现示例

3.3 独占式超时获取同步状态

通过调用同步器的doAcquireNanos(int arg, long nanosTimeout)可以超时获取同步状态。

synchronized：当一个线程获取不到锁而被阻塞在synchronized之外时，对该线程进行中断操作，此时线程中断标志位会被修改，但线程仍然会阻塞在synchronized方法上。
acquireInterruptibly(int arg)：在等待获取同步状态时，如果当前线程被中断，会立刻返回InterruptedException。
acquireInterruptibly(int arg, long nanosTimeout)：与独占式获取的区别在于，在获取同步状态失败后，不是简单地进入阻塞状态，而是会判断是否超时。
- 如果超时 > 1000ns，重新计算超时时间，通过LockSupport.parkNanos等待对应的超时时间后返回
- 如果超时 <= 1000ns，不进入超时等待，而是进入快速的自旋过程。

    private boolean doAcquireNanos(int arg, long nanosTimeout)
            throws InterruptedException {
        if (nanosTimeout <= 0L)
            return false;
        //1.计算出截止时间.
        final long deadline = System.nanoTime() + nanosTimeout;
       //2.加入节点
        final Node node = addWaiter(Node.EXCLUSIVE);
        boolean failed = true;
        try {
            for (;;) {
                //3.取出前驱节点
                final Node p = node.predecessor();
                //4.如果获取成功则直接返回
                if (p == head && tryAcquire(arg)) {
                    setHead(node);
                    p.next = null; // help GC
                    failed = false;
                    return true;
                }
                nanosTimeout = deadline - System.nanoTime();
                //5.如果到了超时时间，则直接返回
                if (nanosTimeout <= 0L)
                    return false;
                if (shouldParkAfterFailedAcquire(p, node) &&
                    nanosTimeout > spinForTimeoutThreshold)
                    LockSupport.parkNanos(this, nanosTimeout);
                //6.如果在自旋过程中被中断，那么抛出异常返回
                if (Thread.interrupted())
                    throw new InterruptedException();
            }
        } finally {
            if (faisled)
                cancelAcquire(node);
        }
    }

四、阻塞和唤醒线程

在前面，我们多次提到了阻塞和唤醒线程，它的实现依靠的是LockSupport工具类，它定义了一组的公共静态方法，提供了最基本的线程阻塞和唤醒功能：

park：阻塞当前线程，直到调用unpark(Thread thread)或者当前线程被中断才返回。
parkNanos(long nanos)：在park基础上增加了超时返回。
parkUntil(long deadline)：直到deadline时间才返回。
unpark(Thread thread)：唤醒处于阻塞状态的线程thread。

五、Condition

ConditionObject是同步器AQS的内部类，每个Condition对象都包含着一个等待队列，该队列是实现等待/通知功能的关键，其用法也很简单，就是通过await和signal/signalAll方法来进行等待和通知。

5.1 等待队列

每一个Condition包含了一个等待队列，该队列结点的类型为AbstractQueuedSynchronizer.Node，与AQS中同步队列结点的类型相同，头结点firstWaiter和尾结点lastWaiter。

等待队列

这里需要注意的是 一个AQS对象只有一个同步队列，但是它可以关联到多个同步队列上。

5.2 等待

调用Condition的await()方法的前提是当前线程已经获取了同步状态。它会执行下面的操作：

使用当前线程创建一个新的结点加入等待队列。
从同步队列中移除当前线程，即释放当前线程的同步状态，再唤醒其在同步队列中的后继结点。

    public final void await() throws InterruptedException {
        if (Thread.interrupted()) throw new InterruptedException();
        //1.创建新的结点加入等待队列。
        Node node = addConditionWaiter();
        //2.释放同步状态。
        long savedState = fullyRelease(node);
        int interruptMode = 0;
        //3.判断是否在同步队列当中，如果不再则阻塞。
        while (!isOnSyncQueue(node)) {
            //3.1.等待被唤醒，唤醒的方法就是调用 signal 方法。
            LockSupport.park(this);
            if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                break;
        }
        //4.被唤醒后从等待队列移到同步队列中，继续参与同步状态的竞争。
        if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
            interruptMode = REINTERRUPT;
        if (node.nextWaiter != null) // clean up if cancelled
            unlinkCancelledWaiters();
        if (interruptMode != 0)
            reportInterruptAfterWait(interruptMode);
    }

5.3 通知

调用Condition的signal()方法的前提条件是当前线程已经获取了同步状态，它会执行下面的操作：

调用同步器的end(Node node)方法，将等待队列中的头结点（注意是等待队列的头结点，而不是当前线程关联的结点）线程安全地移动到同步队列。
结点移动到同步队列后，当前线程再使用LockSupport唤醒该结点的线程。

    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;
        //1.加入到同步队列中。    
        Node p = enq(node);
        int ws = p.waitStatus;
        if (ws > 0 || !p.compareAndSetWaitStatus(ws, Node.SIGNAL))
            //2.唤醒。
            LockSupport.unpark(node.thread);
        return true;
    }

被唤醒的线程，从await方法第3步中的while循环中退出，进而调用同步器的acquireQueued()方法加入到获取同步状态的竞争中（也就是5.2小结中从3循环退出后的逻辑）。

signalAll相当于对等待队列中的每个结点均执行一次signal方法，效果就是将等待队列中所有节点全部移动到同步队列中，并唤醒每个节点的线程。

5.4 生产者消费者模型

生产者消费者模型是多线程协作的经典示例。在这个模型中包含三个角色。

生产者：向缓冲区中写入，当缓冲区满时等待；当缓冲区有新的数据后，通知消费者。
消费者：从缓存区获取，当缓冲区为空时等待，并通知生产者。
缓冲区：存储数据，这里我们用LinkedList来表示。

下面是实现的代码：

/**
 * @author lizejun
 **/
public class Demo {

    private static final int MAX_SIZE = 5;

    private final LinkedList<Integer> factory = new LinkedList<>();

    private Lock lock = new ReentrantLock();
    private Condition consumer = lock.newCondition();
    private Condition producer = lock.newCondition();

    private void producer2() {
        lock.lock();
        try {
            while (factory.size() == MAX_SIZE) {
                try {
                    System.out.println("-- factory is full --");
                    producer.await();
                    Thread.sleep(Math.round(Math.random() * 100));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            System.out.println("produce product");
            factory.add(factory.size());
            consumer.signalAll();
        } finally {
            lock.unlock();
        }
    }

    private void consumer2() {
        lock.lock();
        try {
            while (factory.size() == 0) {
                try {
                    System.out.println("-- factory is empty --");
                    consumer.await();
                    Thread.sleep(Math.round(Math.random() * 100));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            System.out.println("consume product");
            factory.removeLast();
            producer.signalAll();
        } finally {
            lock.unlock();
        }
    }

    public static void run2() {
        final Demo demo = new Demo();
        Thread pThread = new Thread() {

            @Override
            public void run() {
                for (int i = 0; i < 100; i++) {
                    demo.producer2();
                }
            }

        };
        Thread cThread = new Thread() {

            @Override
            public void run() {
                for (int i = 0; i < 100; i++) {
                    demo.consumer2();
                }
            }
        };
        pThread.start();
        cThread.start();
    }
}

运行结果为：

生产者消费者模型

5.5 Condition 提供的其它方法

void await() throw InterruptedException：当前线程进入等待队列直到被signal通知或中断。
void awaitUninterruptibly()：当前线程进入等待状态直到被通知，对中断不敏感。
long awaitNanos(long nanosTimeout) throws InterruptedException：当前线程进入等待状态直到被通知，返回值表示剩余的时间，如果返回值<=0，那么就是超时。
boolean waitUntil(Data deadline) throws InterruptedException：当前线程进入等待状态直到被通知、中断或者到某个时间，如果没有到指定时间就通知，则返回true。
void signal()：唤醒一个等待在Condition上的线程，该线程从等待方法返回前必须获得和Condition相关的锁。
void signalAll()：与signal()的区别在于，它会唤醒所有等待在Condition上的线程。