入门AQS锁 - Semaphore


终于来到AQS锁系列的最后一篇文章了。虽然这是AQS系列的最后一篇文章,但并不代表semaphore理解起来会很困难。Semaphore的用法与实现原理和CountDownLatch类似。对于掌握了CountDownLatch的读者来说,Semaphore理解起来应该是游刃有余。 Semaphore直译为信号。实际上Semaphore可以看做是一个信号的集合。不同的线程能够从Semaphore中获取若干个信号量。当Semaphore对象持有的信号量不足时,尝试从Semaphore中获取信号的线程将会阻塞。直到其他线程将信号量释放以后,阻塞的线程会被唤醒,重新尝试获取信号量。

使用Semaphore

下面这个例子可以很好地解释Semaphore的作用。

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;

public class SemaphoreDemo {

    public static void main(String[] args) throws InterruptedException {

        SemaphoreDemo sd = new SemaphoreDemo();
        // 创建一个信号量集合.信号量总数为10
        Semaphore sem = new Semaphore(10);

        ExecutorService es = Executors.newCachedThreadPool();
        // 获取6个信号量
        es.execute(sd.new Thread1(sem, 6));
        // 获取3个信号量
        es.execute(sd.new Thread2(sem, 3));
        // 获取2个信号量
        es.execute(sd.new Thread3(sem, 2));

        es.shutdown();
    }

    public class Thread1 implements Runnable {

        private Semaphore sem;

        private int countDown;

        public Thread1(Semaphore sem, int countDown) {
            this.sem = sem;
            this.countDown = countDown;
        }

        @Override
        public void run() {
            try {
                // 尝试从信号量集合中获取countDown个信号量
                sem.acquire(countDown);
                System.out.printf("%s获取到%d个信号量\n", Thread1.class.getSimpleName(), countDown);
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                System.out.printf("%s释放%d个信号量\n", Thread1.class.getSimpleName(), countDown);
                sem.release(countDown);
            }
        }

    }

    public class Thread2 implements Runnable {

        private Semaphore sem;

        private int countDown;

        public Thread2(Semaphore sem, int countDown) {
            this.sem = sem;
            this.countDown = countDown;
        }

        @Override
        public void run() {
            try {
                // 尝试从信号量集合中获取countDown个信号量
                sem.acquire(countDown);
                System.out.printf("%s获取到%d个信号量\n", Thread2.class.getSimpleName(), countDown);
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                System.out.printf("%s释放%d个信号量\n", Thread2.class.getSimpleName(), countDown);
                sem.release(countDown);
            }
        }
    }

    public class Thread3 implements Runnable {

        private Semaphore sem;

        private int countDown;

        public Thread3(Semaphore sem, int countDown) {
            this.sem = sem;
            this.countDown = countDown;
        }

        @Override
        public void run() {
            try {
                // 尝试从信号量集合中获取countDown个信号量
                sem.acquire(countDown);
                System.out.printf("%s获取到%d个信号量\n", Thread3.class.getSimpleName(), countDown);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                System.out.printf("%s释放%d个信号量\n", Thread3.class.getSimpleName(), countDown);
                sem.release(countDown);
            }
        }
    }

}
---------------------
Thread1获取到6个信号量
Thread2获取到3个信号量
Thread1释放6个信号量
Thread2释放3个信号量
Thread3获取到2个信号量
Thread3释放2个信号量

在main函数中,启动Thread1,Thread2,Thtread3三个线程任务。 Thread1与Thread2首先从总数为10的Semaphore信号对象中获取到9个信号量。Thread3由于Semaphore中的信号量不足,线程阻塞。直到Thread1与Thread2释放掉持有的信号量,Thread3才得以被唤醒并执行。

Semaphore原理

让我们还是同往常一样,先从类的数据结构入手来剖析Semaphore的原理。在接下来的内容里,就如开头所说的那样你会发现Semaphore在设计理念上与CountDownLatch有许多相似之处。

public class Semaphore implements java.io.Serializable {
    private static final long serialVersionUID = -3222578661600680210L;
    // Semaphore中持有sync的引用。sync继承自AQS锁。
    private final Sync sync;

    /**
     * Synchronization implementation for semaphore.  Uses AQS state
     * to represent permits. Subclassed into fair and nonfair
     * versions.
     */
    abstract static class Sync extends AbstractQueuedSynchronizer {
        private static final long serialVersionUID = 1192457210091910933L;

        Sync(int permits) {
            setState(permits);
        }
    ..........
    ........
    ......
    ....
    // Semaphore的默认构造函数。默认创建的是非公平锁。设置锁状态为permits
    public Semaphore(int permits) {
        sync = new NonfairSync(permits);
    }
    // 通过fair创建公平/非公平锁。
    public Semaphore(int permits, boolean fair) {
        sync = fair ? new FairSync(permits) : new NonfairSync(permits);
    }
    ...........
    ........
    .....
    ..
    // 尝试获取permits个信号量。
    public void acquire(int permits) throws InterruptedException {
        if (permits < 0) throw new IllegalArgumentException();
        sync.acquireSharedInterruptibly(permits);
    }
    // 尝试释放permits个信号量。
    public void release(int permits) {
        if (permits < 0) throw new IllegalArgumentException();
        sync.releaseShared(permits);
    }

分析该类大致的结构以后,可见创建Semaphore对象时传入的信号量参数,实际就是用来设置该共享锁的state。根据前面的知识,我们很自然的能够联想到,acquire()获取信号量,实际是否就是在减少state值呢?


    // 尝试获取permits个信号量。
    public void acquire(int permits) throws InterruptedException {
        if (permits < 0) throw new IllegalArgumentException();
        // 调用AQS抽象类中的共通方法来实现“获取信号量”
        sync.acquireSharedInterruptibly(permits);
    }
package java.util.concurrent.locks;

public abstract class AbstractQueuedSynchronizer
    extends AbstractOwnableSynchronizer
    implements java.io.Serializable {

    private static final long serialVersionUID = 7373984972572414691L;
    ....
    ...
    ..
    // 定义在AQS抽象类中的acquireSharedInterruptibly共通方法。
    public final void acquireSharedInterruptibly(int arg)
            throws InterruptedException {
        if (Thread.interrupted())
            throw new InterruptedException();
       // 尝试获取定义在Semaphore中的共享锁。
        if (tryAcquireShared(arg) < 0)
            // 共享锁获取失败,则通过doAcquireSharedInerruptibly()获取共享锁。
            doAcquireSharedInterruptibly(arg);
    }


     private void doAcquireSharedInterruptibly(int arg)
        throws InterruptedException {
        // 将当前线程包装为节点加入到CLH队列末尾
        final Node node = addWaiter(Node.SHARED);
        boolean failed = true;
        try {
            for (;;) {
                // 获取当前线程节点的前继节点
                final Node p = node.predecessor();
                // 若前继节点为head,则尝试获取共享锁
                if (p == head) {
                    int r = tryAcquireShared(arg);
                    if (r >= 0) {
                        setHeadAndPropagate(node, r);
                        p.next = null; // help GC
                        failed = false;
                        return;
                    }
                }
                // 判断是否需要阻塞当前线程。若需要则阻塞当前线程
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    throw new InterruptedException();
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }
}

Semaphore基于公平锁和非公平锁的不同实现,重写了tryAcquireShared方法。

 // 公平锁
 static final class FairSync extends Sync {
        private static final long serialVersionUID = 2014338818796000944L;
        ......
        ....
        ...
        protected int tryAcquireShared(int acquires) {
            for (;;) {
                // 判断是否有比当前线程等的更久的线程存在。有则获取共享锁失败
                if (hasQueuedPredecessors())
                    return -1;
                // 获取到信号量集合中的集合数(锁状态)
                int available = getState();
                // 集合数-尝试获取的信号量
                int remaining = available - acquires;
                // 重设信号量。
                if (remaining < 0 ||
                    compareAndSetState(available, remaining))
                    return remaining;
            }
        }
        .....
        ...
        ..
 }

果然,Semaphore在公平锁的状态下调用acquire尝试获取信号量,实际就是就是对锁的state进行操作。 首先判断当前队列中是否有比尝试获取锁的线程等待的更久的线程,若没有,则获取锁。线程继续执行。若当前线程不是队列中的第一个线程,则通过AQS抽象类的doAcquireSharedInterruptibly方法,将当前线程加入到队列中,然后线程阻塞。

非公平锁的Semaphore的acquire方法与公平锁大同小异。唯一区别就是首次尝试获取信号量时不判断线程是否处于CLH队列中。但是需要注意的是,若获取信号量失败,线程依然会进入CLH排队并阻塞。

    // 非公平锁
    static final class NonfairSync extends Sync {
        private static final long serialVersionUID = -2694183684443567898L;

        NonfairSync(int permits) {
            super(permits);
        }

        protected int tryAcquireShared(int acquires) {
            // 调用非公平所的共享锁获取。
            return nonfairTryAcquireShared(acquires);
        }
    }

 abstract static class Sync extends AbstractQueuedSynchronizer {
        private static final long serialVersionUID = 1192457210091910933L;

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

        final int getPermits() {
            return getState();
        }
        // 尝试在非公平锁的情况下获取信号量。
        final int nonfairTryAcquireShared(int acquires) {
            for (;;) {
                // 直接开始对锁状态进行操作。不判断是否处于CLH队列队首
                int available = getState();
                int remaining = available - acquires;
                if (remaining < 0 ||
                    compareAndSetState(available, remaining))
                    return remaining;
            }
        }

信号量能够被获取,自然也能够被释放。在第一小节中我们也看到了Thread3在另外两个线程释放掉自己的信号量后,Thread3获取到了请求的信号量,继续了线程的工作。释放信号量的操作,实际就是释放锁,重设锁状态的操作。

    // 释放信号量
    public void release(int permits) {
        if (permits < 0) throw new IllegalArgumentException();
        // 调用定义在AQS抽象类中的releaseShared()共通方法实现信号量释放
        sync.releaseShared(permits);
    }

public abstract class AbstractQueuedSynchronizer
    extends AbstractOwnableSynchronizer
    implements java.io.Serializable {

    private static final long serialVersionUID = 7373984972572414691L;

    public final boolean releaseShared(int arg) {
        // 调用Semaphore重写了tryReleaseShared方法
        if (tryReleaseShared(arg)) {
            doReleaseShared();
            return true;
        }
        return false;
    }

    // 唤醒CLH队列中所有等待的线程。
    private void doReleaseShared() {
        for (;;) {
            Node h = head;
            if (h != null && h != tail) {
                int ws = h.waitStatus;
                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;
        }
    }


}

 abstract static class Sync extends AbstractQueuedSynchronizer {
        private static final long serialVersionUID = 1192457210091910933L;

        Sync(int permits) {
            setState(permits);
        }
        // 释放信号量
        protected final boolean tryReleaseShared(int releases) {
            for (;;) {
                // 取得当前的锁状态
                int current = getState();
                // 对锁状态进行计算。
                int next = current + releases;
                if (next < current) // overflow
                    throw new Error("Maximum permit count exceeded"); 
                // 重设锁状态
                if (compareAndSetState(current, next))
                    return true;
            }
        }

}

总结

AQS锁系列的博文到此就告一段落了。本系列是以入门为目的的文章。若你真的想精通AQS锁,在阅读了本系列的文章后我认为还需要长时间深入源码一探究竟。 AQS架构庞大,设计巧妙。就像文章所介绍的那样,要整体理解AQS的运作原理,运用AQS锁解决问题不是很难。难点在于AQS架构中有很多细节上的处理比如node,各种锁状态,线程中断后的处理甚至是写代码的风格,各种设计模式的运用,都是需要花时间,也值得花时间进行钻研的。 随着java版本的提升,AQS的内容还在增多。concurrent包里的内容也变得越来越丰富。使用这些高级工具,不仅能解决更多更复杂的需求,提高代码的阅读性的同事,这些高级工具的设计思想也值得我们借鉴。

Copyright © chuck Lin 2017 all right reserved,powered by Gitbook该文件修订时间: 2018-07-15 10:39:59

results matching ""

    No results matching ""