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

時間  2019-12-10
標籤 java 併發 系列 15天 juc semaphore 最簡單 限流 工具 必備 技能 欄目 Java 简体版
原文   原文鏈接

這是java高併發系列第15篇文章 java

Semaphore（信號量）爲多線程協做提供了更爲強大的控制方法，前面的文章中咱們學了synchronized和重入鎖ReentrantLock，這2種鎖一次都只能容許一個線程訪問一個資源，而信號量能夠控制有多少個線程能夠同時訪問特定的資源。安全

Semaphore經常使用場景：限流微信

舉個例子：多線程

好比有個停車場，有5個空位，門口有個門衛，手中5把鑰匙分別對應5個車位上面的鎖，來一輛車，門衛會給司機一把鑰匙，而後進去找到對應的車位停下來，出去的時候司機將鑰匙歸還給門衛。停車場生意比較好，同時來了100兩車，門衛手中只有5把鑰匙，同時只能放5輛車進入，其餘車只能等待，等有人將鑰匙歸還給門衛以後，才能讓其餘車輛進入。併發

上面的例子中門衛就至關於Semaphore，車鑰匙就至關於許可證，車就至關於線程。異步

Semaphore主要方法

  • Semaphore(int permits)：構造方法，參數表示許可證數量，用來建立信號量分佈式

  • Semaphore(int permits,boolean fair)：構造方法，當fair等於true時，建立具備給定許可數的計數信號量並設置爲公平信號量
  • void acquire() throws InterruptedException：今後信號量獲取1個許可前線程將一直阻塞，至關於一輛車佔了一個車位，此方法會響應線程中斷，表示調用線程的interrupt方法，會使該方法拋出InterruptedException異常
  • void acquire(int permits) throws InterruptedException ：和acquire()方法相似，參數表示須要獲取許可的數量；好比一個大卡車要入停車場，因爲車比較大，須要申請3個車位才能夠停放
  • void acquireUninterruptibly(int permits) ：和acquire(int permits) 方法相似，只是不會響應線程中斷
  • boolean tryAcquire()：嘗試獲取1個許可，不論是否可以獲取成功，都當即返回，true表示獲取成功，false表示獲取失敗
  • boolean tryAcquire(int permits)：和tryAcquire()，表示嘗試獲取permits個許可
  • boolean tryAcquire(long timeout, TimeUnit unit) throws InterruptedException：嘗試在指定的時間內獲取1個許可，獲取成功返回true，指定的時間事後仍是沒法獲取許可，返回false
  • boolean tryAcquire(int permits, long timeout, TimeUnit unit) throws InterruptedException：和tryAcquire(long timeout, TimeUnit unit)相似，多了一個permits參數，表示嘗試獲取permits個許可
  • void release()：釋放一個許可，將其返回給信號量，至關於車從停車場出去時將鑰匙歸還給門衛
  • void release(int n)：釋放n個許可

  • int availablePermits()：當前可用的許可數ide

示例1：Semaphore簡單的使用

package com.itsoku.chat12;

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

/**
 * 微信公衆號：路人甲Java，專一於java技術分享（帶你玩轉 爬蟲、分佈式事務、異步消息服務、任務調度、分庫分表、大數據等），喜歡請關注！
 */
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)建立了許可數量爲2的信號量，每一個線程獲取1個許可，同時容許兩個線程獲取許可，從輸出中也能夠看出，同時有兩個線程能夠獲取許可，其餘線程須要等待已獲取許可的線程釋放許可以後才能運行。爲獲取到許可的線程會阻塞在acquire()方法上，直到獲取到許可才能繼續。工具

示例2：獲取許可以後不釋放

門衛（Semaphore）有點呆，司機進去的時候給了鑰匙，出來的時候不歸還，門衛也不會說什麼。最終結果就是其餘車輛都沒法進入了。

以下代碼：

package com.itsoku.chat12;

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

/**
 * 微信公衆號：路人甲Java，專一於java技術分享（帶你玩轉 爬蟲、分佈式事務、異步消息服務、任務調度、分庫分表、大數據等），喜歡請關注！
 */
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() + ",運行結束!");
                System.out.println(System.currentTimeMillis() + "," + thread.getName() + ",當前可用許可數量:" + 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,運行結束!
1563716606925,t-0,當前可用許可數量:0
1563716606925,t-1,運行結束!
1563716606925,t-1,當前可用許可數量:0

上面程序運行後一直沒法結束，觀察一下代碼，代碼中獲取許可後，沒有釋放許可的代碼，最終致使，可用許可數量爲0，其餘線程沒法獲取許可，會在semaphore.acquire();處等待，致使程序沒法結束。

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

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

若是獲取鎖的過程當中發生異常，致使獲取鎖失敗，最後finally裏面也釋放了許可，最終會怎麼樣，致使許可數量憑空增加了。

示例代碼：

package com.itsoku.chat12;

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

/**
 * 微信公衆號：路人甲Java，專一於java技術分享（帶你玩轉 爬蟲、分佈式事務、異步消息服務、任務調度、分庫分表、大數據等），喜歡請關注！
 */
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() + ",獲取許可,當前可用許可數量:" + semaphore.availablePermits());
                //休眠100秒
                TimeUnit.SECONDS.sleep(100);
                System.out.println(System.currentTimeMillis() + "," + thread.getName() + ",運行結束!");
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                semaphore.release();
            }
            System.out.println(System.currentTimeMillis() + "," + thread.getName() + ",當前可用許可數量:" + 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發送中斷信號
        t2.interrupt();
        t3.interrupt();
    }
}

輸出：

1563717279058,t1,獲取許可,當前可用許可數量:0
java.lang.InterruptedException
1563717281060,t2,當前可用許可數量:1
    at java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedInterruptibly(AbstractQueuedSynchronizer.java:998)
1563717281060,t3,當前可用許可數量: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)

程序中信號量許可數量爲1，建立了3個線程獲取許可，線程t1獲取成功了，而後休眠100秒。其餘兩個線程阻塞在semaphore.acquire();方法處，代碼中對線程t二、t3發送中斷信號，咱們看一下Semaphore中acquire的源碼：

public void acquire() throws InterruptedException

這個方法會響應線程中斷，主線程中對t二、t3發送中斷信號以後，acquire()方法會觸發InterruptedException異常，t二、t3最終沒有獲取到許可，可是他們都執行了finally中的釋放許可的操做，最後致使許可數量變爲了2，致使許可數量增長了。因此程序中釋放許可的方式有問題。須要改進一下，獲取許可成功纔去釋放鎖。

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

package com.itsoku.chat12;

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

/**
 * 微信公衆號：路人甲Java，專一於java技術分享（帶你玩轉 爬蟲、分佈式事務、異步消息服務、任務調度、分庫分表、大數據等），喜歡請關注！
 */
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() + ",獲取許可,當前可用許可數量:" + semaphore.availablePermits());
                //休眠100秒
                TimeUnit.SECONDS.sleep(5);
                System.out.println(System.currentTimeMillis() + "," + thread.getName() + ",運行結束!");
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                if (acquireSuccess) {
                    semaphore.release();
                }
            }
            System.out.println(System.currentTimeMillis() + "," + thread.getName() + ",當前可用許可數量:" + 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發送中斷信號
        t2.interrupt();
        t3.interrupt();
    }
}

輸出：

1563717751655,t1,獲取許可,當前可用許可數量:0
1563717753657,t3,當前可用許可數量:0
java.lang.InterruptedException
1563717753657,t2,當前可用許可數量: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,運行結束!
1563717756656,t1,當前可用許可數量:1

程序中增長了一個變量acquireSuccess用來標記獲取許但是否成功，在finally中根據這個變量是否爲true，來肯定是否釋放許可。

示例4：在規定的時間內但願獲取許可

司機來到停車場，發現停車場已經滿了，只能在外等待內部的車出來以後才能進去，可是要等多久，他本身也不知道，他但願等10分鐘，若是仍是沒法進去，就不到這裏停車了。

Semaphore內部2個方法能夠提供超時獲取許可的功能：

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

在指定的時間內去嘗試獲取許可，若是可以獲取到，返回true，獲取不到返回false。

示例代碼：

package com.itsoku.chat12;

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

/**
 * 微信公衆號：路人甲Java，專一於java技術分享（帶你玩轉 爬蟲、分佈式事務、異步消息服務、任務調度、分庫分表、大數據等），喜歡請關注！
 */
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秒內獲取許可，獲取成功返回true，不然返回false
                System.out.println(System.currentTimeMillis() + "," + thread.getName() + ",嘗試獲取許可,當前可用許可數量:" + semaphore.availablePermits());
                acquireSuccess = semaphore.tryAcquire(1, TimeUnit.SECONDS);
                //獲取成功執行業務代碼
                if (acquireSuccess) {
                    System.out.println(System.currentTimeMillis() + "," + thread.getName() + ",獲取許可成功,當前可用許可數量:" + semaphore.availablePermits());
                    //休眠5秒
                    TimeUnit.SECONDS.sleep(5);
                } else {
                    System.out.println(System.currentTimeMillis() + "," + thread.getName() + ",獲取許可失敗,當前可用許可數量:" + 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,嘗試獲取許可,當前可用許可數量:1
1563718410202,t1,獲取許可成功,當前可用許可數量:0
1563718411203,t2,嘗試獲取許可,當前可用許可數量:0
1563718412203,t3,嘗試獲取許可,當前可用許可數量:0
1563718412204,t2,獲取許可失敗,當前可用許可數量:0
1563718413204,t3,獲取許可失敗,當前可用許可數量:0

代碼中許可數量爲1，semaphore.tryAcquire(1, TimeUnit.SECONDS);：表示嘗試在1秒內獲取許可，獲取成功當即返回true，超過1秒仍是獲取不到，返回false。線程t1獲取許可成功，以後休眠了5秒，從輸出中能夠看出t2和t3都嘗試了1秒，獲取失敗。

其餘一些使用說明

  1. Semaphore默認建立的是非公平的信號量，什麼意思呢？這個涉及到公平與非公平。舉個例子：5個車位，容許5個車輛進去，來了100輛車，只能進去5輛，其餘95在外面排隊等着。裏面恰好出來了1輛，此時恰好又來了10輛車，這10輛車是直接插隊到其餘95輛前面去，仍是到95輛後面去排隊呢？排隊就表示公平，直接去插隊爭搶第一個，就表示不公平。對於停車場，排隊確定更好一些咯。不過對於信號量來講不公平的效率更高一些，因此默認是不公平的。
  2. 建議閱讀如下Semaphore的源碼，對經常使用的方法有個瞭解，不須要都記住，用的時候也方便查詢就好。
  3. 方法中帶有throws InterruptedException聲明的，表示這個方法會響應線程中斷信號，什麼意思？表示調用線程的interrupt()方法，會讓這些方法觸發InterruptedException異常，即便這些方法處於阻塞狀態，也會當即返回，並拋出InterruptedException異常，線程中斷信號也會被清除。

java高併發系列

java高併發系列連載中，總計估計會有四五十篇文章，能夠關注公衆號：javacode2018，送月薪3萬課程，獲取最新文章。

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程