java-AbstractQueuedSynchronizer(AQS)

2018-11-18

阅读量

学习java的并发包(JUC, java.util.concurrent)，自然就回想起几个著名的类: ReentrantLock、Semaphore、CountDownLatch等。而在这些类都是基于AQS(AbstractQueuedSynchronizer)实现的,一个抽象模板类。了解和学习AQS，可以更好的理解和掌握JUC。

AQS支持共享和非共享两种模式，非共享对应的有ReentrantLock，而共享对应的有CountDownLatch，semaphore等。

AQS的数据结构其实比较简单，大致这个样子。

head 和 tail 类型都是AQS的内部类-> node类型，
state是一个int类型.也就是互斥/共享资源，state>0 资源被占用， state==0 资源被释放。

两种模式

独占模式:

ReentrantLock:
state为0代表资源未被占有，可以获取锁并设置为1，此时其他线程访问为1，代表资源已经被锁上(占有)，所以线程自身进入lock(AQS)的双向队列中,等待资源释放。

共享模式:

CountDownLatch:
一个线程(一般是主线程) latch = new CountDownLatch(num); 即state设为num
latch.await(); 将线程自身加入latch(AQS)的双向队列，等待资源释放
latch.countDown(); 将state-1,一旦state为0说明资源被释放。(一般是其他线程执行完成)，之前await的线程可以继续执行。

源码分析

talk is cheap show me the code

Semaphore semaphore = new Semaphore(5);
semaphore.acquire();
//doSometing
semaphore.release();

最近用到了semaphore用来限流。也就是dosometing的代码非常消耗内存，如果并发高很容易出现oom。这个时候就可以根据分配的Xmx和dosometing的计算最大并发个数，进行限制。

那么如何实现的并发控制，点进源码see see。

//semaphore 类的定义。
public class Semaphore implements java.io.Serializable {
    private final Sync sync;
    abstract static class Sync extends AbstractQueuedSynchronizer {...}
    public void acquire() throws InterruptedException {
        sync.acquireSharedInterruptibly(1);
    }
    public void release() {
        sync.releaseShared(1);
    }
    // somemethods 几乎都和sync 内部类对象有关，而Sync实现的就是AQS.

acquire方法

那么先来看看acquire方法。实际执行的是sync的acquireSharedInterruptibly(1)

该方法没有被重写，在AbstractQueuedSynchronizer类中

public final void acquireSharedInterruptibly(int arg)
        throws InterruptedException {
    if (Thread.interrupted())   // 线程是否中断
        throw new InterruptedException();
    // 尝试获取资源 小于0 获取失败
    if (tryAcquireShared(arg) < 0)
        // 加到阻塞队列中，会再次尝试是否能获取资源。
        doAcquireSharedInterruptibly(arg);
}

也就是尝试获取资源，发现此时资源已经被用完，所以自身线程加入aqs队列，将自己阻塞起来。

tryAcquireShared方法:

该方法重写了，在Semaphore类中。

public boolean tryAcquire() {return sync.nonfairTryAcquireShared(1) >= 0;}

final int nonfairTryAcquireShared(int acquires) {
  for (;;) {
    int available = getState();
    int remaining = available - acquires;
    // 没有资源， 获取资源失败/成功。返回剩余资源数量。没有则 < 0.
    // compareAndSetState cas操作。 尝试将资源 state = available - acquires
    if (remaining < 0 || compareAndSetState(available, remaining))
        return remaining;
  }
}

可以看到，如果有资源，则资源减一，返回的数大于等于0，不阻塞。反之，没有资源，那么执行doAcquireSharedInterruptibly() 操作，再次尝试，如果失败则阻塞。

doAcquireSharedInterruptibly 方法:

该方法没有被重写，在AbstractQueuedSynchronizer类中

private void doAcquireSharedInterruptibly(int arg)
    throws InterruptedException {
    /**这个addWaiter 很讲究，大致就是将自己这个线程加入到aqs的阻塞队列的末尾。
     *  Node node = new Node(Thread.currentThread(), mode); 并返回node
     *  如果是第一个阻塞， 会生产一个空head.
     * 不过为了保证线程安全，用了cas操作
     * 先尝试一遍放在队列末尾，如果失败，进入enq方法，
     * 无限循环尝试放在阻塞队列的末尾. (自旋锁)
     * 而且head节点是一个空节点，new Node(); 不存储数据。
    **/
    final Node node = addWaiter(Node.SHARED);
    boolean failed = true;
    try {
        /** 无限循环，尝试获取资源，
         *  是否应该阻塞， 调用原语阻塞自身。
         *  当前线程停止在if(xxx && parkAndCheckInterrupt)处。
         *  一旦有资源被释放， 从此处唤醒，继续无限循环尝试获取资源。
        **/
        for (;;) {
            /** 如果当前节点的前一个是head，而head是个标记，不存储数据，
             *  说明前面没有阻塞的node节点，那么再尝试获取一遍资源。
            **/
            final Node p = node.predecessor();
            if (p == head) {
                int r = tryAcquireShared(arg);
                if (r >= 0) {
                    setHeadAndPropagate(node, r);
                    p.next = null; // help GC
                    failed = false;
                    return;
                }
            }
            /** shouldParkAfterFailedAcquire 从阻塞队列末尾往前查看，
             * 看看前面的资源是否被阻塞，如果被阻塞自己就排在后面也阻塞起来，
             * 如果没有被阻塞，说明前面这个线程放弃了，从队列中删除
             * 回到上面循环，重新尝试获取资源。然后再阻塞自己。
             * 将自身的前一个节点的waitStatus设置为SIGNAL，等待资源唤醒。
             * 所以node.waitStatus 是指下一个节点需要的状态(等待唤醒, 取消...)
             * parkAndCheckInterrupt 调用原语，阻塞自己。
            **/
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                throw new InterruptedException();
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

到这里阻塞就算是完成了。线程没有获取到资源，全都暂停在parkAndCheckInterrupt。等待其他拥有资源的线程release，那么就会唤醒阻塞的线程，重新回到for(;;)。重新获取资源。

release方法

线程释放资源调用 release方法

    public void release() {sync.releaseShared(1);}

// 该方法没有被重写，在AbstractQueuedSynchronizer类中
    public final boolean releaseShared(int arg) {
       // 无限循环尝试释放资源， state=state+1， 释放成功
       if (tryReleaseShared(arg)) {
            // 释放成功， 唤醒线程抢夺资源。。。
           doReleaseShared();
           return true;
       }
       return false;
   }

和acquire方法相反， release方法释放资源，然后doReleaseShared 唤醒之前被阻塞在parkAndCheckInterrupt 方法的线程。

doReleaseShared 方法:

 private void doReleaseShared() {
     for (;;) {
        Node h = head;
        // 如果阻塞队列中有数据。
        if (h != null && h != tail) {   
            int ws = h.waitStatus;
            // 如果h的状态是等待唤醒
            if (ws == Node.SIGNAL) {
                if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                    continue;            // loop to recheck cases
                // 唤醒一个后继节点
                unparkSuccessor(h);
            }
            // 将设置为共享模式。到了这一步，说明阻塞队列为空。
            else if (ws == 0 &&
                     !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                continue;                // loop on failed CAS
        }
        // 这次循环头节点没有变化，阻塞队列全部被唤醒。
        if (h == head)                   // loop if head changed
            break;
    }
}

这个代码也是一个无限循环(闭锁)，然后唤醒所有没有放弃的后继节点。线程就会回到上面doAcquire中继续循环。抢占资源或者继续阻塞。等待唤醒。

unpartSuccessor方法：

private void unparkSuccessor(Node node) {
    int ws = node.waitStatus;
    // 修改状态
    if (ws < 0)
        compareAndSetWaitStatus(node, ws, 0);

    // head.next 才是需要唤醒的线程节点。
    Node s = node.next;
    // 如果next放弃了。修改s位置。
    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;
    }
    // s不为空，执行原语，唤醒阻塞线程。
    if (s != null)
        LockSupport.unpark(s.thread);
}

到了这里才是真正唤醒线程。而互斥与之差不多，不过是资源数变成1，互斥访问。