java高并發(fā)系列 - 第15天：JUC中的Semaphore，最簡單的限流工具類，必備技能

這是java高并發(fā)系列第15篇文章

Semaphore（信號量）為多線程協(xié)作提供了更為強(qiáng)大的控制方法，前面的文章中我們學(xué)了synchronized和重入鎖ReentrantLock，這2種鎖一次都只能允許一個(gè)線程訪問一個(gè)資源，而信號量可以控制有多少個(gè)線程可以同時(shí)訪問特定的資源。

Semaphore常用場景：限流

舉個(gè)例子：

比如有個(gè)停車場，有5個(gè)空位，門口有個(gè)門衛(wèi)，手中5把鑰匙分別對應(yīng)5個(gè)車位上面的鎖，來一輛車，門衛(wèi)會給司機(jī)一把鑰匙，然后進(jìn)去找到對應(yīng)的車位停下來，出去的時(shí)候司機(jī)將鑰匙歸還給門衛(wèi)。停車場生意比較好，同時(shí)來了100兩車，門衛(wèi)手中只有5把鑰匙，同時(shí)只能放5輛車進(jìn)入，其他車只能等待，等有人將鑰匙歸還給門衛(wèi)之后，才能讓其他車輛進(jìn)入。

上面的例子中門衛(wèi)就相當(dāng)于Semaphore，車鑰匙就相當(dāng)于許可證，車就相當(dāng)于線程。

Semaphore主要方法

• Semaphore(int permits)：構(gòu)造方法，參數(shù)表示許可證數(shù)量，用來創(chuàng)建信號量

• Semaphore(int permits,boolean fair)：構(gòu)造方法，當(dāng)fair等于true時(shí)，創(chuàng)建具有給定許可數(shù)的計(jì)數(shù)信號量并設(shè)置為公平信號量
• void acquire() throws InterruptedException：從此信號量獲取1個(gè)許可前線程將一直阻塞，相當(dāng)于一輛車占了一個(gè)車位，此方法會響應(yīng)線程中斷，表示調(diào)用線程的interrupt方法，會使該方法拋出InterruptedException異常
• void acquire(int permits) throws InterruptedException ：和acquire()方法類似，參數(shù)表示需要獲取許可的數(shù)量；比如一個(gè)大卡車要入停車場，由于車比較大，需要申請3個(gè)車位才可以停放
• void acquireUninterruptibly(int permits) ：和acquire(int permits) 方法類似，只是不會響應(yīng)線程中斷
• boolean tryAcquire()：嘗試獲取1個(gè)許可，不管是否能夠獲取成功，都立即返回，true表示獲取成功，false表示獲取失敗
• boolean tryAcquire(int permits)：和tryAcquire()，表示嘗試獲取permits個(gè)許可
• boolean tryAcquire(long timeout, TimeUnit unit) throws InterruptedException：嘗試在指定的時(shí)間內(nèi)獲取1個(gè)許可，獲取成功返回true，指定的時(shí)間過后還是無法獲取許可，返回false
• boolean tryAcquire(int permits, long timeout, TimeUnit unit) throws InterruptedException：和tryAcquire(long timeout, TimeUnit unit)類似，多了一個(gè)permits參數(shù)，表示嘗試獲取permits個(gè)許可
• void release()：釋放一個(gè)許可，將其返回給信號量，相當(dāng)于車從停車場出去時(shí)將鑰匙歸還給門衛(wèi)
• void release(int n)：釋放n個(gè)許可
• int availablePermits()：當(dāng)前可用的許可數(shù)

示例1：Semaphore簡單的使用

package com.itsoku.chat12;

import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

/**
 * 微信公眾號：javacode2018，獲取年薪50萬課程
 */
public class Demo1 {
    static Semaphore semaphore = new Semaphore(2);

    public static class T extends Thread {
        public T(String name) {
            super(name);
        }

        @Override
        public void run() {
            Thread thread = Thread.currentThread();
            try {
                semaphore.acquire();
                System.out.println(System.currentTimeMillis() + "," + thread.getName() + ",獲取許可!");
                TimeUnit.SECONDS.sleep(3);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                semaphore.release();
                System.out.println(System.currentTimeMillis() + "," + thread.getName() + ",釋放許可!");
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        for (int i = 0; i < 10; i++) {
            new T("t-" + i).start();
        }
    }
}

輸出：

1563715791327,t-0,獲取許可!
1563715791327,t-1,獲取許可!
1563715794328,t-0,釋放許可!
1563715794328,t-5,獲取許可!
1563715794328,t-1,釋放許可!
1563715794328,t-2,獲取許可!
1563715797328,t-2,釋放許可!
1563715797328,t-6,獲取許可!
1563715797328,t-5,釋放許可!
1563715797328,t-3,獲取許可!
1563715800329,t-6,釋放許可!
1563715800329,t-9,獲取許可!
1563715800329,t-3,釋放許可!
1563715800329,t-7,獲取許可!
1563715803330,t-7,釋放許可!
1563715803330,t-8,獲取許可!
1563715803330,t-9,釋放許可!
1563715803330,t-4,獲取許可!
1563715806330,t-8,釋放許可!
1563715806330,t-4,釋放許可!

代碼中new Semaphore(2)創(chuàng)建了許可數(shù)量為2的信號量，每個(gè)線程獲取1個(gè)許可，同時(shí)允許兩個(gè)線程獲取許可，從輸出中也可以看出，同時(shí)有兩個(gè)線程可以獲取許可，其他線程需要等待已獲取許可的線程釋放許可之后才能運(yùn)行。為獲取到許可的線程會阻塞在acquire()方法上，直到獲取到許可才能繼續(xù)。

示例2：獲取許可之后不釋放

門衛(wèi)（Semaphore）有點(diǎn)呆，司機(jī)進(jìn)去的時(shí)候給了鑰匙，出來的時(shí)候不歸還，門衛(wèi)也不會說什么。最終結(jié)果就是其他車輛都無法進(jìn)入了。

如下代碼：

package com.itsoku.chat12;

import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

/**
 * 微信公眾號：javacode2018，獲取年薪50萬課程
 */
public class Demo2 {
    static Semaphore semaphore = new Semaphore(2);

    public static class T extends Thread {
        public T(String name) {
            super(name);
        }

        @Override
        public void run() {
            Thread thread = Thread.currentThread();
            try {
                semaphore.acquire();
                System.out.println(System.currentTimeMillis() + "," + thread.getName() + ",獲取許可!");
                TimeUnit.SECONDS.sleep(3);
                System.out.println(System.currentTimeMillis() + "," + thread.getName() + ",運(yùn)行結(jié)束!");
                System.out.println(System.currentTimeMillis() + "," + thread.getName() + ",當(dāng)前可用許可數(shù)量:" + semaphore.availablePermits());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        for (int i = 0; i < 10; i++) {
            new T("t-" + i).start();
        }
    }
}

輸出：

1563716603924,t-0,獲取許可!
1563716603924,t-1,獲取許可!
1563716606925,t-0,運(yùn)行結(jié)束!
1563716606925,t-0,當(dāng)前可用許可數(shù)量:0
1563716606925,t-1,運(yùn)行結(jié)束!
1563716606925,t-1,當(dāng)前可用許可數(shù)量:0

上面程序運(yùn)行后一直無法結(jié)束，觀察一下代碼，代碼中獲取許可后，沒有釋放許可的代碼，最終導(dǎo)致，可用許可數(shù)量為0，其他線程無法獲取許可，會在semaphore.acquire();處等待，導(dǎo)致程序無法結(jié)束。

示例3：釋放許可正確的姿勢

示例1中，在finally里面釋放鎖，會有問題么？

如果獲取鎖的過程中發(fā)生異常，導(dǎo)致獲取鎖失敗，最后finally里面也釋放了許可，最終會怎么樣，導(dǎo)致許可數(shù)量憑空增長了。

示例代碼：

package com.itsoku.chat12;

import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

/**
 * 微信公眾號：javacode2018，獲取年薪50萬課程
 */
public class Demo3 {
    static Semaphore semaphore = new Semaphore(1);

    public static class T extends Thread {
        public T(String name) {
            super(name);
        }

        @Override
        public void run() {
            Thread thread = Thread.currentThread();
            try {
                semaphore.acquire();
                System.out.println(System.currentTimeMillis() + "," + thread.getName() + ",獲取許可,當(dāng)前可用許可數(shù)量:" + semaphore.availablePermits());
                //休眠100秒
                TimeUnit.SECONDS.sleep(100);
                System.out.println(System.currentTimeMillis() + "," + thread.getName() + ",運(yùn)行結(jié)束!");
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                semaphore.release();
            }
            System.out.println(System.currentTimeMillis() + "," + thread.getName() + ",當(dāng)前可用許可數(shù)量:" + semaphore.availablePermits());
        }
    }

    public static void main(String[] args) throws InterruptedException {
        T t1 = new T("t1");
        t1.start();
        //休眠1秒
        TimeUnit.SECONDS.sleep(1);
        T t2 = new T("t2");
        t2.start();
        //休眠1秒
        TimeUnit.SECONDS.sleep(1);
        T t3 = new T("t3");
        t3.start();

        //給t2和t3發(fā)送中斷信號
        t2.interrupt();
        t3.interrupt();
    }
}

輸出：

1563717279058,t1,獲取許可,當(dāng)前可用許可數(shù)量:0
java.lang.InterruptedException
1563717281060,t2,當(dāng)前可用許可數(shù)量:1
    at java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedInterruptibly(AbstractQueuedSynchronizer.java:998)
1563717281060,t3,當(dāng)前可用許可數(shù)量:2
    at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1304)
    at java.util.concurrent.Semaphore.acquire(Semaphore.java:312)
    at com.itsoku.chat12.Demo3$T.run(Demo3.java:21)
java.lang.InterruptedException
    at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1302)
    at java.util.concurrent.Semaphore.acquire(Semaphore.java:312)
    at com.itsoku.chat12.Demo3$T.run(Demo3.java:21)

程序中信號量許可數(shù)量為1，創(chuàng)建了3個(gè)線程獲取許可，線程t1獲取成功了，然后休眠100秒。其他兩個(gè)線程阻塞在semaphore.acquire();方法處，代碼中對線程t2、t3發(fā)送中斷信號，我們看一下Semaphore中acquire的源碼：

public void acquire() throws InterruptedException

這個(gè)方法會響應(yīng)線程中斷，主線程中對t2、t3發(fā)送中斷信號之后，acquire()方法會觸發(fā)InterruptedException異常，t2、t3最終沒有獲取到許可，但是他們都執(zhí)行了finally中的釋放許可的操作，最后導(dǎo)致許可數(shù)量變?yōu)榱?，導(dǎo)致許可數(shù)量增加了。所以程序中釋放許可的方式有問題。需要改進(jìn)一下，獲取許可成功才去釋放鎖。

正確的釋放鎖的方式，如下：

package com.itsoku.chat12;

import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

/**
 * 微信公眾號：javacode2018，獲取年薪50萬課程
 */
public class Demo4 {
    static Semaphore semaphore = new Semaphore(1);

    public static class T extends Thread {
        public T(String name) {
            super(name);
        }

        @Override
        public void run() {
            Thread thread = Thread.currentThread();
            //獲取許可是否成功
            boolean acquireSuccess = false;
            try {
                semaphore.acquire();
                acquireSuccess = true;
                System.out.println(System.currentTimeMillis() + "," + thread.getName() + ",獲取許可,當(dāng)前可用許可數(shù)量:" + semaphore.availablePermits());
                //休眠100秒
                TimeUnit.SECONDS.sleep(5);
                System.out.println(System.currentTimeMillis() + "," + thread.getName() + ",運(yùn)行結(jié)束!");
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                if (acquireSuccess) {
                    semaphore.release();
                }
            }
            System.out.println(System.currentTimeMillis() + "," + thread.getName() + ",當(dāng)前可用許可數(shù)量:" + semaphore.availablePermits());
        }
    }

    public static void main(String[] args) throws InterruptedException {
        T t1 = new T("t1");
        t1.start();
        //休眠1秒
        TimeUnit.SECONDS.sleep(1);
        T t2 = new T("t2");
        t2.start();
        //休眠1秒
        TimeUnit.SECONDS.sleep(1);
        T t3 = new T("t3");
        t3.start();

        //給t2和t3發(fā)送中斷信號
        t2.interrupt();
        t3.interrupt();
    }
}

輸出：

1563717751655,t1,獲取許可,當(dāng)前可用許可數(shù)量:0
1563717753657,t3,當(dāng)前可用許可數(shù)量:0
java.lang.InterruptedException
1563717753657,t2,當(dāng)前可用許可數(shù)量:0
    at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1302)
    at java.util.concurrent.Semaphore.acquire(Semaphore.java:312)
    at com.itsoku.chat12.Demo4$T.run(Demo4.java:23)
java.lang.InterruptedException
    at java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedInterruptibly(AbstractQueuedSynchronizer.java:998)
    at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1304)
    at java.util.concurrent.Semaphore.acquire(Semaphore.java:312)
    at com.itsoku.chat12.Demo4$T.run(Demo4.java:23)
1563717756656,t1,運(yùn)行結(jié)束!
1563717756656,t1,當(dāng)前可用許可數(shù)量:1

程序中增加了一個(gè)變量acquireSuccess用來標(biāo)記獲取許可是否成功，在finally中根據(jù)這個(gè)變量是否為true，來確定是否釋放許可。

示例4：在規(guī)定的時(shí)間內(nèi)希望獲取許可

司機(jī)來到停車場，發(fā)現(xiàn)停車場已經(jīng)滿了，只能在外等待內(nèi)部的車出來之后才能進(jìn)去，但是要等多久，他自己也不知道，他希望等10分鐘，如果還是無法進(jìn)去，就不到這里停車了。

Semaphore內(nèi)部2個(gè)方法可以提供超時(shí)獲取許可的功能：

public boolean tryAcquire(long timeout, TimeUnit unit) throws InterruptedException
public boolean tryAcquire(int permits, long timeout, TimeUnit unit)
        throws InterruptedException 

在指定的時(shí)間內(nèi)去嘗試獲取許可，如果能夠獲取到，返回true，獲取不到返回false。

示例代碼：

package com.itsoku.chat12;

import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

/**
 * 微信公眾號：javacode2018，獲取年薪50萬課程
 */
public class Demo5 {
    static Semaphore semaphore = new Semaphore(1);

    public static class T extends Thread {
        public T(String name) {
            super(name);
        }

        @Override
        public void run() {
            Thread thread = Thread.currentThread();
            //獲取許可是否成功
            boolean acquireSuccess = false;
            try {
                //嘗試在1秒內(nèi)獲取許可，獲取成功返回true，否則返回false
                System.out.println(System.currentTimeMillis() + "," + thread.getName() + ",嘗試獲取許可,當(dāng)前可用許可數(shù)量:" + semaphore.availablePermits());
                acquireSuccess = semaphore.tryAcquire(1, TimeUnit.SECONDS);
                //獲取成功執(zhí)行業(yè)務(wù)代碼
                if (acquireSuccess) {
                    System.out.println(System.currentTimeMillis() + "," + thread.getName() + ",獲取許可成功,當(dāng)前可用許可數(shù)量:" + semaphore.availablePermits());
                    //休眠5秒
                    TimeUnit.SECONDS.sleep(5);
                } else {
                    System.out.println(System.currentTimeMillis() + "," + thread.getName() + ",獲取許可失敗,當(dāng)前可用許可數(shù)量:" + semaphore.availablePermits());
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                if (acquireSuccess) {
                    semaphore.release();
                }
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        T t1 = new T("t1");
        t1.start();
        //休眠1秒
        TimeUnit.SECONDS.sleep(1);
        T t2 = new T("t2");
        t2.start();
        //休眠1秒
        TimeUnit.SECONDS.sleep(1);
        T t3 = new T("t3");
        t3.start();
    }
}

輸出：

1563718410202,t1,嘗試獲取許可,當(dāng)前可用許可數(shù)量:1
1563718410202,t1,獲取許可成功,當(dāng)前可用許可數(shù)量:0
1563718411203,t2,嘗試獲取許可,當(dāng)前可用許可數(shù)量:0
1563718412203,t3,嘗試獲取許可,當(dāng)前可用許可數(shù)量:0
1563718412204,t2,獲取許可失敗,當(dāng)前可用許可數(shù)量:0
1563718413204,t3,獲取許可失敗,當(dāng)前可用許可數(shù)量:0

代碼中許可數(shù)量為1，semaphore.tryAcquire(1, TimeUnit.SECONDS);：表示嘗試在1秒內(nèi)獲取許可，獲取成功立即返回true，超過1秒還是獲取不到，返回false。線程t1獲取許可成功，之后休眠了5秒，從輸出中可以看出t2和t3都嘗試了1秒，獲取失敗。

其他一些使用說明

1. Semaphore默認(rèn)創(chuàng)建的是非公平的信號量，什么意思呢？這個(gè)涉及到公平與非公平。舉個(gè)例子：5個(gè)車位，允許5個(gè)車輛進(jìn)去，來了100輛車，只能進(jìn)去5輛，其他95在外面排隊(duì)等著。里面剛好出來了1輛，此時(shí)剛好又來了10輛車，這10輛車是直接插隊(duì)到其他95輛前面去，還是到95輛后面去排隊(duì)呢？排隊(duì)就表示公平，直接去插隊(duì)爭搶第一個(gè)，就表示不公平。對于停車場，排隊(duì)肯定更好一些咯。不過對于信號量來說不公平的效率更高一些，所以默認(rèn)是不公平的。
2. 建議閱讀以下Semaphore的源碼，對常用的方法有個(gè)了解，不需要都記住，用的時(shí)候也方便查詢就好。
3. 方法中帶有throws InterruptedException聲明的，表示這個(gè)方法會響應(yīng)線程中斷信號，什么意思？表示調(diào)用線程的interrupt()方法，會讓這些方法觸發(fā)InterruptedException異常，即使這些方法處于阻塞狀態(tài)，也會立即返回，并拋出InterruptedException異常，線程中斷信號也會被清除。

java高并發(fā)系列連載中，總計(jì)估計(jì)會有四五十篇文章，可以關(guān)注公眾號：javacode2018，獲取最新文章。