Java高并發(fā)編程基礎(chǔ)三大利器之Semaphore

作者： java金融 2021-03-04 07:24:24

Semaphore(信號(hào)量)是用來(lái)控制同時(shí)訪問(wèn)特定資源的線程數(shù)量，它通過(guò)協(xié)調(diào)各個(gè)線程，以保證合理的使用公共資源。

引言

最近可以進(jìn)行個(gè)稅申報(bào)了，還沒(méi)有申報(bào)的同學(xué)可以趕緊去試試哦。不過(guò)我反正是從上午到下午(3月1日)一直都沒(méi)有成功的進(jìn)行申報(bào)，一進(jìn)行申報(bào) 就返回“當(dāng)前訪問(wèn)人數(shù)過(guò)多，請(qǐng)稍后再試”。為什么有些人就能夠申報(bào)成功，有些人就直接返回失敗。這很明顯申報(bào)處理資源是有限的，只能等別人處理完了在來(lái)處理你的，你如果運(yùn)氣好可能重試幾次就輪到你了，如果運(yùn)氣不好可能重試一天也可能輪不到你。我反正已經(jīng)是放棄了，等到夜深人靜的時(shí)候再來(lái)試試。作為一個(gè)程序員我們肯定知道這是個(gè)稅申請(qǐng)app的限流操作，如果還有不懂什么是限流操作的可以參考下這個(gè)文章《高并發(fā)系統(tǒng)三大利器之限流》。比如個(gè)稅申報(bào)系統(tǒng)每臺(tái)機(jī)器只最多分別最多只能處理1000個(gè)請(qǐng)求，再多的請(qǐng)求就會(huì)把機(jī)器打掛。如果是多余的請(qǐng)求就把這些請(qǐng)求拒絕掉。直接給你返回一句溫馨提示：“當(dāng)前訪問(wèn)人數(shù)過(guò)多，請(qǐng)稍后再試”,如果要實(shí)現(xiàn)這個(gè)功能大家想想可以通過(guò)哪些方法算法來(lái)實(shí)現(xiàn)。

共享鎖、獨(dú)占鎖

學(xué)習(xí)semaphore之前我們必須要先了解下什么是共享鎖。

共享鎖：它是允許多個(gè)線程同時(shí)獲取鎖，并發(fā)的訪問(wèn)共享資源

獨(dú)占鎖：也有人把它叫做“獨(dú)享鎖”，它是是獨(dú)占的，排他的，只能被一個(gè)線程可持有，當(dāng)獨(dú)占鎖已經(jīng)被某個(gè)線程持有時(shí)，其他線程只能等待它被釋放后，才能去爭(zhēng)鎖，并且同一時(shí)刻只有一個(gè)線程能爭(zhēng)鎖成功。

什么是Semaphore

Semaphore(信號(hào)量)是用來(lái)控制同時(shí)訪問(wèn)特定資源的線程數(shù)量，它通過(guò)協(xié)調(diào)各個(gè)線程，以保證合理的使用公共資源。很多年以來(lái)，我都覺(jué)得從字面上很難理解Semaphore所表達(dá)的含義，只能把它比作是控制流量的紅綠燈，比如XX馬路要限制流量，只允許同時(shí)有一百輛車(chē)在這條路上行使，其他的都必須在路口等待，所以前一百輛車(chē)會(huì)看到綠燈，可以開(kāi)進(jìn)這條馬路，后面的車(chē)會(huì)看到紅燈，不能駛?cè)隭X馬路，但是如果前一百輛中有五輛車(chē)已經(jīng)離開(kāi)了XX馬路，那么后面就允許有5輛車(chē)駛?cè)腭R路，這個(gè)例子里說(shuō)的車(chē)就是線程，駛?cè)腭R路就表示線程在執(zhí)行，離開(kāi)馬路就表示線程執(zhí)行完成，看見(jiàn)紅燈就表示線程被阻塞，不能執(zhí)行。”

Semaphore機(jī)制是提供給線程搶占式獲取許可，所以他可以實(shí)現(xiàn)公平或者非公平，類(lèi)似于ReentrantLock。說(shuō)了這么多我們來(lái)個(gè)實(shí)際的例子看一看，比如我們?nèi)ネ＼?chē)場(chǎng)停車(chē)，停車(chē)場(chǎng)總共只有5個(gè)車(chē)位，但是現(xiàn)在有8輛汽車(chē)來(lái)停車(chē)，剩下的3輛汽車(chē)要么等其他汽車(chē)開(kāi)走后進(jìn)行停車(chē)，或者去找別的停車(chē)位?

/** 
 * @author: 公眾號(hào)【Java金融】 
 */ 
public class SemaphoreTest { 
    public static void main(String[] args) throws InterruptedException { 
         // 初始化五個(gè)車(chē)位 
        Semaphore semaphore = new Semaphore(5); 
        // 等所有車(chē)子 
        final CountDownLatch latch = new CountDownLatch(8); 
        for (int i = 0; i < 8; i++) { 
            int finalI = i; 
            if (i == 5) { 
                Thread.sleep(1000); 
                new Thread(() -> { 
                    stopCarNotWait(semaphore, finalI); 
                    latch.countDown(); 
                }).start(); 
                continue; 
            } 
            new Thread(() -> { 
                stopCarWait(semaphore, finalI); 
                latch.countDown(); 
            }).start(); 
        } 
        latch.await(); 
        log("總共還剩：" + semaphore.availablePermits() + "個(gè)車(chē)位"); 
    } 
 
    private static void stopCarWait(Semaphore semaphore, int finalI) { 
        String format = String.format("車(chē)牌號(hào)%d", finalI); 
        try { 
            semaphore.acquire(1); 
            log(format + "找到車(chē)位了，去停車(chē)了"); 
            Thread.sleep(10000); 
        } catch (Exception e) { 
            e.printStackTrace(); 
        } finally { 
            semaphore.release(1); 
            log(format + "開(kāi)走了"); 
        } 
    } 
 
    private static void stopCarNotWait(Semaphore semaphore, int finalI) { 
         String format = String.format("車(chē)牌號(hào)%d", finalI); 
        try { 
            if (semaphore.tryAcquire()) { 
                log(format + "找到車(chē)位了，去停車(chē)了"); 
                Thread.sleep(10000); 
                log(format + "開(kāi)走了"); 
                semaphore.release(); 
            } else { 
                log(format + "沒(méi)有停車(chē)位了，不在這里等了去其他地方停車(chē)去了"); 
            } 
        } catch (Exception e) { 
            e.printStackTrace(); 
        } 
 
    } 
 
    public static void log(String content) { 
        // 格式化 
        DateTimeFormatter fmTime = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"); 
        // 當(dāng)前時(shí)間 
        LocalDateTime now = LocalDateTime.now(); 
        System.out.println(now.format(fmTime) + "  "+content); 
    } 
}

2021-03-01 18:54:57  車(chē)牌號(hào)0找到車(chē)位了，去停車(chē)了 
2021-03-01 18:54:57  車(chē)牌號(hào)3找到車(chē)位了，去停車(chē)了 
2021-03-01 18:54:57  車(chē)牌號(hào)2找到車(chē)位了，去停車(chē)了 
2021-03-01 18:54:57  車(chē)牌號(hào)1找到車(chē)位了，去停車(chē)了 
2021-03-01 18:54:57  車(chē)牌號(hào)4找到車(chē)位了，去停車(chē)了 
2021-03-01 18:54:58  車(chē)牌號(hào)5沒(méi)有停車(chē)位了，不在這里等了去其他地方停車(chē)去了 
2021-03-01 18:55:07  車(chē)牌號(hào)7找到車(chē)位了，去停車(chē)了 
2021-03-01 18:55:07  車(chē)牌號(hào)6找到車(chē)位了，去停車(chē)了 
2021-03-01 18:55:07  車(chē)牌號(hào)2開(kāi)走了 
2021-03-01 18:55:07  車(chē)牌號(hào)0開(kāi)走了 
2021-03-01 18:55:07  車(chē)牌號(hào)3開(kāi)走了 
2021-03-01 18:55:07  車(chē)牌號(hào)4開(kāi)走了 
2021-03-01 18:55:07  車(chē)牌號(hào)1開(kāi)走了 
2021-03-01 18:55:17  車(chē)牌號(hào)7開(kāi)走了 
2021-03-01 18:55:17  車(chē)牌號(hào)6開(kāi)走了 
2021-03-01 18:55:17  總共還剩：5個(gè)車(chē)位

從輸出結(jié)果我們可以看到車(chē)牌號(hào)5這輛車(chē)看見(jiàn)沒(méi)有車(chē)位了，就不在這個(gè)地方傻傻的等了，而是去其他地方了，但是車(chē)牌號(hào)6和車(chē)牌號(hào)7分別需要等到車(chē)庫(kù)開(kāi)出兩輛車(chē)空出兩個(gè)車(chē)位后才停進(jìn)去。這就體現(xiàn)了Semaphore 的acquire 方法如果沒(méi)有獲取到憑證它就會(huì)阻塞，而tryAcquire方法如果沒(méi)有獲取到憑證不會(huì)阻塞的。

semaphore在dubbo中的應(yīng)用

在Dubbo中可以給Provider配置線程池大小來(lái)控制系統(tǒng)提供服務(wù)的最大并行度，默認(rèn)是200。

<dubbo:provider  threads="200"/>

比如我現(xiàn)在這個(gè)訂單系統(tǒng)有三個(gè)接口，分別為創(chuàng)單、取消訂單、修改訂單。這三個(gè)接口加起來(lái)的并發(fā)是200但是創(chuàng)單接口是核心接口，我想讓它多分點(diǎn)線程來(lái)執(zhí)行讓它可以有最大150個(gè)線程，取消訂單和修改訂單分別最大25個(gè)線程執(zhí)行就可以了。dubbo提供了executes這一屬性來(lái)實(shí)現(xiàn)這個(gè)功能

<dubbo:service interface="cn.javajr.service.CreateOrderService" executes="150"/> 
<dubbo:service interface="cn.javajr.service.CancelOrderService" executes="25"/> 
<dubbo:service interface="cn.javajr.service.EditOrderService" executes="25"/>

我們可以看看dubbo內(nèi)部是如何來(lái)executes的，具體實(shí)現(xiàn)是在ExecuteLimitFilter這個(gè)類(lèi)我們可以

 public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException { 
        URL url = invoker.getUrl(); 
        String methodName = invocation.getMethodName(); 
        Semaphore executesLimit = null; 
        boolean acquireResult = false; 
        int max = url.getMethodParameter(methodName, Constants.EXECUTES_KEY, 0); 
        if (max > 0) { 
            RpcStatus count = RpcStatus.getStatus(url, invocation.getMethodName()); 
            // 如果當(dāng)前使用的線程數(shù)量已經(jīng)大于等于設(shè)置的閾值，那么直接拋出異常 
//            if (count.getActive() >= max) { 
// throw new RpcException("Failed to invoke method " + invocation.getMethodName() + " in provider " + url + ", cause: The service // using threads greater than <dubbo:service executes=\"" + max + "\" /> limited."); 
            /** 
             * http://manzhizhen.iteye.com/blog/2386408 
             * use semaphore for concurrency control (to limit thread number) 
             */ 
              
            executesLimit = count.getSemaphore(max); 
            if(executesLimit != null && !(acquireResult = executesLimit.tryAcquire())) { 
                throw new RpcException("Failed to invoke method " + invocation.getMethodName() + " in provider " + url + ", cause: The service using threads greater than <dubbo:service executes=\"" + max + "\" /> limited."); 
            } 
        } 
        long begin = System.currentTimeMillis(); 
        boolean isSuccess = true; 
        // 計(jì)數(shù)器+1 
        RpcStatus.beginCount(url, methodName); 
        try { 
            Result result = invoker.invoke(invocation); 
            return result; 
        } catch (Throwable t) { 
            isSuccess = false; 
            if (t instanceof RuntimeException) { 
                throw (RuntimeException) t; 
            } else { 
                throw new RpcException("unexpected exception when ExecuteLimitFilter", t); 
            } 
        } finally { 
           // 計(jì)數(shù)器-1 
            RpcStatus.endCount(url, methodName, System.currentTimeMillis() - begin, isSuccess); 
            if(acquireResult) { 
                executesLimit.release(); 
            } 
        } 
    }

從上述代碼我們也可以看出早期這個(gè)是沒(méi)有采用Semaphore來(lái)實(shí)現(xiàn)的，而是直接采用被注釋的 if (count.getActive() >= max) 這個(gè)來(lái)來(lái)實(shí)現(xiàn)的，由于這個(gè)count.getActive() >= max 和這個(gè)計(jì)數(shù)加1不是原子性的,所以會(huì)有問(wèn)題，具體bug號(hào)可以看https://github.com/apache/dubbo/pull/582后面才采用上述代碼用Semaphore來(lái)修復(fù)非原子性問(wèn)題。具體更詳細(xì)的分析可以參見(jiàn)代碼的鏈接。不過(guò)現(xiàn)在最新版本(2.7.9)我看是采用采用自旋加上和CAS來(lái)實(shí)現(xiàn)的。

Semaphore

上面就是對(duì)Semaphore一個(gè)簡(jiǎn)單的使用以及dubbo中用到的例子,說(shuō)句實(shí)話Semaphore在工作中用的還是比較少的，不過(guò)面試又有可能會(huì)被問(wèn)到，所以還是有必要來(lái)一起學(xué)習(xí)一下它。我們前面《Java高并發(fā)編程基礎(chǔ)之AQS》通過(guò)ReentrantLock 一起學(xué)習(xí)了下AQS，其實(shí)Semaphore同樣也是通過(guò)AQS來(lái)是實(shí)現(xiàn)的，我們可以一起來(lái)對(duì)照下獨(dú)占鎖的方法，基本上都是有方法一一相對(duì)應(yīng)的。圖片這里有兩點(diǎn)稍微需要注意的地方：

在獨(dú)占鎖模式中，我們只有在獲取了獨(dú)占鎖的節(jié)點(diǎn)釋放鎖時(shí)，才會(huì)喚醒后繼節(jié)點(diǎn)，因?yàn)楠?dú)占鎖只能被一個(gè)線程持有，如果它還沒(méi)有被釋放，就沒(méi)有必要去喚醒它的后繼節(jié)點(diǎn)。
在共享鎖模式下，當(dāng)一個(gè)節(jié)點(diǎn)獲取到了共享鎖，我們?cè)讷@取成功后就可以喚醒后繼節(jié)點(diǎn)了，而不需要等到該節(jié)點(diǎn)釋放鎖的時(shí)候，這是因?yàn)楣蚕礞i可以被多個(gè)線程同時(shí)持有，一個(gè)鎖獲取到了，則后繼的節(jié)點(diǎn)都可以直接來(lái)獲取。因此，在共享鎖模式下，在獲取鎖和釋放鎖結(jié)束時(shí)，都會(huì)喚醒后繼節(jié)點(diǎn)。

獲取憑證

我們同樣還是通過(guò)非公平鎖的模式來(lái)獲取憑證我們可以看下acquire的核心方法

public final void acquireSharedInterruptibly(int arg) 
          throws InterruptedException { 
      if (Thread.interrupted()) 
          throw new InterruptedException(); 
      if (tryAcquireShared(arg) < 0) 
          doAcquireSharedInterruptibly(arg); 
  } 
   protected int tryAcquireShared(int acquires) { 
           return nonfairTryAcquireShared(acquires); 
  } 
 
// 主要看下這個(gè)方法，這個(gè)方法返回的值也就是tryAcquireShared返回的值，因?yàn)閠ryAcquireShared->nonfairTryAcquireShared 
   final int nonfairTryAcquireShared(int acquires) { 
         //自旋 
   for (;;) { 
        //Semaphore用AQS的state變量的值代表可用許可數(shù) 
        int available = getState(); 
        //可用許可數(shù)減去本次需要獲取的許可數(shù)即為剩余許可數(shù) 
        int remaining = available - acquires; 
        //如果剩余許可數(shù)小于0或者CAS將當(dāng)前可用許可數(shù)設(shè)置為剩余許可數(shù)成功，則返回成功許可數(shù) 
        if (remaining < 0 || 
            compareAndSetState(available, remaining)) 
            return remaining; 
    }

當(dāng)tryAcquireShared 獲取返回許可書(shū)小于0時(shí)說(shuō)明獲取許可失敗需要進(jìn)入doAcquireSharedInterruptibly這個(gè)方法去休眠。
當(dāng)tryAcquireShared 獲取返回許可書(shū)小于0時(shí)說(shuō)明獲取許可成功直接結(jié)束。

doAcquireSharedInterruptibly

private void doAcquireSharedInterruptibly(int arg) 
       throws InterruptedException { 
       // 獨(dú)占鎖的acquireQueued調(diào)用的是addWaiter(Node.EXCLUSIVE)， 
       // 而共享鎖調(diào)用的是addWaiter(Node.SHARED)，表明了該節(jié)點(diǎn)處于共享模式 
       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); 
       } 
   }

這個(gè)方法是不是跟我們上篇文章講的AQS的獨(dú)占鎖的acquireQueued很像，不過(guò)獨(dú)占鎖它是直接調(diào)用了用了setHead(node)方法，而共享鎖調(diào)用的是setHeadAndPropagate(node, r)這個(gè)方法除了調(diào)用setHead 里面還調(diào)用了doReleaseShared(喚醒后繼節(jié)點(diǎn))

private void setHeadAndPropagate(Node node, int propagate) { 
      Node h = head; // Record old head for check below 
      setHead(node); 
      if (propagate > 0 || h == null || h.waitStatus < 0 || 
          (h = head) == null || h.waitStatus < 0) { 
          Node s = node.next; 
          if (s == null || s.isShared()) 
              doReleaseShared(); 
      } 
  }

其他的方法基本上是和ReentrantLock來(lái)實(shí)現(xiàn)的獨(dú)占鎖差不多，我相信大家對(duì)源碼分析感興趣的應(yīng)該也不多，其他更多細(xì)節(jié)問(wèn)題還是需要自己親自動(dòng)手去看源碼的。

總結(jié)

當(dāng)信號(hào)量Semaphore初始化設(shè)置許可證為1 時(shí)，它也可以當(dāng)作互斥鎖使用。其中0、1就相當(dāng)于它的狀態(tài)，當(dāng)=1時(shí)表示其他線程可以獲取，當(dāng)=0時(shí)，排他，即其他線程必須要等待。

Semaphore是JUC包中的一個(gè)很簡(jiǎn)單的工具類(lèi)，用來(lái)實(shí)現(xiàn)多線程下對(duì)于資源的同一時(shí)刻的訪問(wèn)線程數(shù)限制

Semaphore中存在一個(gè)【許可】的概念，即訪問(wèn)資源之前，先要獲得許可，如果當(dāng)前許可數(shù)量為0，那么線程阻塞，直到獲得許可

Semaphore內(nèi)部使用AQS實(shí)現(xiàn)，由抽象內(nèi)部類(lèi)Sync繼承了AQS。因?yàn)镾emaphore天生就是共享的場(chǎng)景，所以其內(nèi)部實(shí)際上類(lèi)似于共享鎖的實(shí)現(xiàn)

共享鎖的調(diào)用框架和獨(dú)占鎖很相似，它們最大的不同在于獲取鎖的邏輯——共享鎖可以被多個(gè)線程同時(shí)持有，而獨(dú)占鎖同一時(shí)刻只能被一個(gè)線程持有。

由于共享鎖同一時(shí)刻可以被多個(gè)線程持有，因此當(dāng)頭節(jié)點(diǎn)獲取到共享鎖時(shí)，可以立即喚醒后繼節(jié)點(diǎn)來(lái)爭(zhēng)鎖，而不必等到釋放鎖的時(shí)候。因此，共享鎖觸發(fā)喚醒后繼節(jié)點(diǎn)的行為可能有兩處，一處在當(dāng)前節(jié)點(diǎn)成功獲得共享鎖后，一處在當(dāng)前節(jié)點(diǎn)釋放共享鎖后。

采用semaphore來(lái)進(jìn)行限流的話會(huì)產(chǎn)生突刺現(xiàn)象。

★指在一定時(shí)間內(nèi)的一小段時(shí)間內(nèi)就用完了所有資源，后大部分時(shí)間中無(wú)資源可用。比如在限流方法中的計(jì)算器算法，設(shè)置1s內(nèi)的最大請(qǐng)求數(shù)為100，在前100ms已經(jīng)有了100個(gè)請(qǐng)求，則后面900ms將無(wú)法處理請(qǐng)求，這就是突刺現(xiàn)象。

本文轉(zhuǎn)載自微信公眾號(hào)「java金融」，可以通過(guò)以下二維碼關(guān)注。轉(zhuǎn)載本文請(qǐng)聯(lián)系java金融公眾號(hào)。