zoukankan      html  css  js  c++  java
  • JUC--Semaphore信号量

    信号量

    一:什么是信号量

    信号量是对锁的扩展,不管是同步synchronized还是ReentrantLock,一次只能允许一个线程访问一个资源,但是信号量可以使得多个线程,同时访问一个资源。

    基本方法:

    public Semaphore(int permits) {
        sync = new NonfairSync(permits);
    }
    public Semaphore(int permits, boolean fair) {
        sync = fair ? new FairSync(permits) : new NonfairSync(permits);
    }

    permits参数表示能同时申请多少个信号量,主要用来控制线程的并发数量,即同时有多少个线程可以访问资源。

    信号量的主要API:

    public void acquire();
    public void acquireUninterruptibly();
    public boolean tryAcquire();
    public boolean tryAcquire(long timeout, TimeUnit unit);
    public void release();
    • acquire():尝试获得准入的许可,若无法获得,会持续等待,直到线程释放一个许可,或者线程中断。
    • void acquireUninterruptibly():与acquire类似,但是不响应中断。
    • tryAcquire():尝试获得准入许可,得到返回true,没有返回false,不会持续等待。
    • tryAcquire(long timeout, TimeUnit unit):与tryAcquire类似,但是会有一个等待的时间,超过时间立即返回。
    • release():线程访问资源结束后,释放许可。

    先用Semaphore实现一个购票的小例子,来看看如何使用:

    public class Ticket {
    
        public static void main(String[] args) {
            Semaphore semaphore = new Semaphore(5);  // 声明5个令牌
            //同时运行了10个线程,但是只有5个线程在工作,另外5个被阻塞了。
            for (int i = 0; i < 10; i++) {
                new Thread() {
                    @Override
                    public void run() {
                        try {
                            System.out.println(Thread.currentThread().getName() + ": 开始运行。。");
                            semaphore.acquire();  // 占用令牌
                            System.out.println(Thread.currentThread().getName() + ": 开始买票");
                            sleep(2000);  // 睡2秒,模拟买票流程
                            System.out.println(Thread.currentThread().getName() + ": 购票成功");
                            semaphore.release();  // 释放令牌
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }.start();
            }
        }
    Thread-1: 开始运行。。
    Thread-1: 开始买票
    Thread-7: 开始运行。。
    Thread-7: 开始买票
    Thread-3: 开始运行。。
    Thread-3: 开始买票
    Thread-2: 开始运行。。
    Thread-2: 开始买票
    Thread-6: 开始运行。。
    Thread-6: 开始买票
    Thread-5: 开始运行。。
    Thread-9: 开始运行。。
    Thread-0: 开始运行。。
    Thread-4: 开始运行。。
    Thread-8: 开始运行。。
    Thread-7: 购票成功
    Thread-5: 开始买票
    Thread-1: 购票成功
    Thread-9: 开始买票
    Thread-3: 购票成功
    Thread-0: 开始买票
    Thread-2: 购票成功
    Thread-4: 开始买票
    Thread-6: 购票成功
    Thread-8: 开始买票
    Thread-5: 购票成功
    Thread-9: 购票成功
    Thread-0: 购票成功
    Thread-4: 购票成功
    Thread-8: 购票成功
    运行结果

    从结果来看,最多只有5个线程在购票。而这么精确的控制,我们也只是调用了acquire和release方法。下面看看是如何实现的。

    从acquire方法进去,又可以看到老套路:具体调用的还是AbstractQueuedSynchronizer这个类的逻辑

        public final void acquireSharedInterruptibly(int arg)
                throws InterruptedException {
            if (Thread.interrupted())
                throw new InterruptedException();
            if (tryAcquireShared(arg) < 0)
                doAcquireSharedInterruptibly(arg);
        }
    而tryAcquireShared方法留给了子类去实现,Semaphore类里面的两个内部类FairSync和NonfairSync都继承自AbstractQueuedSynchronizer。这两个内部类,从名字来看,一个实现了公平锁,另一个是非公平锁。这里多说一句,所谓公平和非公平是这个意思:假设现在有一个线程A在等待获取锁,这时候又来了线程B,如果这个时候B不考虑A的感受,也去申请锁,显然不公平;反之,只要A是先来的,B一定要排在A的后面,不能马上去申请锁,就是公平的。
    Semaphore默认是调用了NonfairSync的tryAcquireShared方法,主要逻辑:
            final int nonfairTryAcquireShared(int acquires) {
                for (;;) {
                    int available = getState();
                    int remaining = available - acquires;
                    if (remaining < 0 ||
                        compareAndSetState(available, remaining))
                        return remaining;
                }
            }

    这又是一个经典的CAS操作加无限循环的算法,用来保证共享变量的正确性。另外,此处的getState()方法很是迷惑人,你以为是获取状态,实则不然。我们先看看Semaphore的构造方法:

        public Semaphore(int permits) {
            sync = new NonfairSync(permits);
        }
            // 内部类
            NonfairSync(int permits) {
                super(permits);
            }
            // 内部类,NonfairSync的父类
            Sync(int permits) {
                setState(permits);
            }

    我们传进去的参数5,最终传给了setState方法,而getState和setState方法都在AbstractQueuedSynchronizer类里面

        /**
         * The synchronization state.
         */
        private volatile int state;
    
        protected final int getState() {
            return state;
        }
    
        protected final void setState(int newState) {
            state = newState;
        }

    也就是说父类定义了一个属性state,并配有final的get和set方法,子类只需要继承该属性,想代表什么含义都可以,比如Semaphore里面的内部类Sync就把这个属性当作最大允许访问的permits,像CountDownLatch和CyclicBarrier都是这么干的。这种方式似乎不太好理解,为什么不是每个子类都定义自己的具有明确语义的属性,而是把控制权放在父类???我猜是出于安全的考虑。反正,大师的思考深度,我们揣摩不了。

    再回到tryAcquireShared方法,这个方法是有参数的---int型的acquires,代表你要一次占几个坑。我们调用的无参的acquire方法,默认是传入1作为参数调用的这个方法,一次只申请一个坑。但是有的情况下,你可能一次需要多个,比如高富帅需要同时交多个女朋友。方法的返回值是剩余的坑的数量,如果数量小于0,执行AbstractQueuedSynchronizer这个类的doAcquireSharedInterruptibly方法。

        /**
         * Acquires in shared interruptible mode.
         * @param arg the acquire argument
         */
        private void doAcquireSharedInterruptibly(int arg)
            throws InterruptedException {
            final Node node = addWaiter(Node.SHARED);
            boolean failed = true;
            try {
                for (;;) {
                    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;
                        }
                    }
                    if (shouldParkAfterFailedAcquire(p, node) &&
                        parkAndCheckInterrupt())
                        throw new InterruptedException();
                }
            } finally {
                if (failed)
                    cancelAcquire(node);
            }
        }

    这个方法的逻辑与独占模式下的逻辑差不多,可以看看之前讲Condition的那篇,出门一路左拐。当所有的坑都被占着的时候,再来的线程都会被封装成节点,添加到等待的队列里面去。不同的是,这里的节点都是共享模式,而共享模式是实现多个坑同时提供服务的核心。

    再来看看坑的释放,从release方法进去,核心逻辑在tryReleaseShared方法:

            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;
                }
            }

    CAS、无限循环,熟悉的配方,熟悉的味道。同获取一样,这里也可以一次释放多个坑。然而,这里考虑到了next小于current的情况,我是绞尽脑汁也没想出来。传进来的releases一般都是大于0的整数(大部分情况下就是1),最终还是会造成next小于current,实在是想不出来,而且还是抛出Error。但是这种情况,通过代码可以精确的再现。好吧,是在下输了。如果读者中有高人,请指点一二,不胜感激!!!

    前面这么多都只是分析了非公平模式下的处理逻辑,而公平模式下的逻辑多了一个判断,就是看看前面还有没有线程在等待(节点有没有前驱)。具体的细节,希望读者自己玩味。

    最后总结一下:所有的并发核心控制逻辑都在AbstractQueuedSynchronizer这个类中,只有理解了这个类的设计思路,才能真正理解衍生出来的工具类的实现原理。

  • 相关阅读:
    Logistic Regression
    Bootstrap研究2布局系统杂记
    《Programming in Scala》读书笔记(持续更新) passover的个人空间 DOIT博客 多易网
    《Scala, Erlang, F#作者讨论函数式语言》有感
    用python抓取oj题目(0)——重回战场 duoduo3_69 博客园
    Netty vs Apache MINA
    What is Akka?
    Fault Tolerance (Scala) — Akka Documentation
    execute phase · mrdon/mavencliplugin Wiki
    netty和mina的比较
  • 原文地址:https://www.cnblogs.com/jvStarBlog/p/13630535.html
Copyright © 2011-2022 走看看