Java AQS信號(hào)量Semaphore的使用

更新時(shí)間：2023年02月02日 10:00:24 作者：飛奔的小付

Semaphore來自于JDK1.5的JUC包，直譯過來就是信號(hào)量，被作為一種多線程并發(fā)控制工具來使用。本文將詳解其原理與使用方法，感興趣的可以學(xué)習(xí)一下

一.什么是Semaphore

Semaphore，俗稱信號(hào)量，它是操作系統(tǒng)中PV操作的原語(yǔ)在java的實(shí)現(xiàn)，它也是基于AbstractQueuedSynchronizer實(shí)現(xiàn)的。

Semaphore的功能非常強(qiáng)大，大小為1的信號(hào)量就類似于互斥鎖，通過同時(shí)只能有一個(gè)線程獲取信號(hào)量實(shí)現(xiàn)。大小為n（n>0）的信號(hào)量可以實(shí)現(xiàn)限流的功能，它可以實(shí)現(xiàn)只能有n個(gè)線程同時(shí)獲取信號(hào)量。

PV操作是操作系統(tǒng)一種實(shí)現(xiàn)進(jìn)程互斥與同步的有效方法。PV操作與信號(hào)量（S）的處理相關(guān)，P表示通過的意思，V表示釋放的意思。用PV操作來管理共享資源時(shí)，首先要確保PV操作自身執(zhí)行的正確性。
P操作的主要?jiǎng)幼魇牵?br />①S減1；
②若S減1后仍大于或等于0，則進(jìn)程繼續(xù)執(zhí)行；
③若S減1后小于0，則該進(jìn)程被阻塞后放入等待該信號(hào)量的等待隊(duì)列中，然后轉(zhuǎn)進(jìn)程調(diào)度。
V操作的主要?jiǎng)幼魇牵?br />①S加1；
②若相加后結(jié)果大于0，則進(jìn)程繼續(xù)執(zhí)行；
③若相加后結(jié)果小于或等于0，則從該信號(hào)的等待隊(duì)列中釋放一個(gè)等待進(jìn)程，然后再返回原進(jìn)程繼續(xù)執(zhí)行或轉(zhuǎn)進(jìn)程調(diào)度。

二.Semaphore的使用

構(gòu)造器

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

permits 表示許可證的數(shù)量（資源數(shù)）

fair 表示公平性，如果這個(gè)設(shè)為 true 的話，下次執(zhí)行的線程會(huì)是等待最久的線程

常用方法

public void acquire() throws InterruptedException
public boolean tryAcquire()
public void release()
public int availablePermits()
public final int getQueueLength()
public final boolean hasQueuedThreads()
protected void reducePermits(int reduction)

acquire() 表示阻塞并獲取許可
tryAcquire() 方法在沒有許可的情況下會(huì)立即返回 false，要獲取許可的線程不會(huì)阻塞
release() 表示釋放許可
int availablePermits()：返回此信號(hào)量中當(dāng)前可用的許可證數(shù)。
int getQueueLength()：返回正在等待獲取許可證的線程數(shù)。
boolean hasQueuedThreads()：是否有線程正在等待獲取許可證。
void reducePermit（int reduction）：減少 reduction 個(gè)許可證
Collection getQueuedThreads()：返回所有等待獲取許可證的線程集合

應(yīng)用場(chǎng)景

可以用于做流量控制，特別是公用資源有限的應(yīng)用場(chǎng)景

限流

/**
     * 實(shí)現(xiàn)一個(gè)同時(shí)只能處理5個(gè)請(qǐng)求的限流器
     */
    private static Semaphore semaphore = new Semaphore(5);
    /**
     * 定義一個(gè)線程池
     */
    private static ThreadPoolExecutor executor = new ThreadPoolExecutor(10, 50, 60, TimeUnit.SECONDS, new LinkedBlockingDeque<>(200));
    public static void exec() {
        try {
            semaphore.acquire(1);
            // 模擬真實(shí)方法執(zhí)行
            System.out.println("執(zhí)行exec方法" );
            Thread.sleep(2000);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            semaphore.release(1);
        }
    }
    public static void main(String[] args) throws InterruptedException {
        {
            for (; ; ) {
                Thread.sleep(100);
                // 模擬請(qǐng)求以10個(gè)/s的速度
                executor.execute(() -> exec());
            }
        }
    }

三.Semaphore源碼分析

主要關(guān)注 Semaphore的加鎖解鎖（共享鎖）邏輯實(shí)現(xiàn)，線程競(jìng)爭(zhēng)鎖失敗入隊(duì)阻塞邏輯和獲取鎖的線程釋放鎖喚醒阻塞線程競(jìng)爭(zhēng)鎖的邏輯實(shí)現(xiàn)

	public void acquire() throws InterruptedException {
        sync.acquireSharedInterruptibly(1);
    }
    public final void acquireSharedInterruptibly(int arg)
            throws InterruptedException {
        if (Thread.interrupted())
            throw new InterruptedException();
            //嘗試獲取共享鎖，大于等于0則直接獲取鎖成功，小于0時(shí)則共享鎖阻塞
        if (tryAcquireShared(arg) < 0)
            doAcquireSharedInterruptibly(arg);
    }

tryAcquireShared的實(shí)現(xiàn)

   final int nonfairTryAcquireShared(int acquires) {
            for (;;) {
                int available = getState();
                int remaining = available - acquires;
                // 當(dāng)減一之后的值小于0 或者 
                // compareAndSetState成功，把state變?yōu)閞emaining，即將狀態(tài)減一
                if (remaining < 0 ||
                    compareAndSetState(available, remaining))
                    return remaining;
            }
        }

入隊(duì)阻塞

 private void doAcquireSharedInterruptibly(int arg)
        throws InterruptedException {
        //入隊(duì)，創(chuàng)建節(jié)點(diǎn) 使用共享模式
        final Node node = addWaiter(Node.SHARED);
        boolean failed = true;
        try {
            for (;;) {
            //獲取當(dāng)前節(jié)點(diǎn)的前軀節(jié)點(diǎn)
                final Node p = node.predecessor();
                //如果節(jié)點(diǎn)為head節(jié)點(diǎn)
                if (p == head) {
                	//阻塞動(dòng)作比較重，通常會(huì)再嘗試獲取資源，沒有獲取到返回負(fù)數(shù)
                    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);
        }
    }

入隊(duì)操作

private Node addWaiter(Node mode) {
		//構(gòu)建節(jié)點(diǎn)，模式是共享模式
        Node node = new Node(Thread.currentThread(), mode);
        Node pred = tail;
        if (pred != null) {
        	//設(shè)置前一節(jié)點(diǎn)為tail 
            node.prev = pred;
            //設(shè)置當(dāng)前節(jié)點(diǎn)為尾節(jié)點(diǎn)
            if (compareAndSetTail(pred, node)) {
            // 前一節(jié)點(diǎn)的next為當(dāng)前節(jié)點(diǎn)
                pred.next = node;
                return node;
            }
        }
        //創(chuàng)建隊(duì)列
        enq(node);
        return node;
    }

創(chuàng)建隊(duì)列，經(jīng)典的for循環(huán)創(chuàng)建雙向鏈表

 private Node enq(final Node node) {
        for (;;) {
            Node t = tail;
            if (t == null) { 
            	//節(jié)點(diǎn)為空 則new一個(gè)節(jié)點(diǎn) 設(shè)置頭節(jié)點(diǎn)
                if (compareAndSetHead(new Node()))
                //把這個(gè)節(jié)點(diǎn)的頭節(jié)點(diǎn)賦值給尾節(jié)點(diǎn)
                    tail = head;
            } else {
            // 如果尾節(jié)點(diǎn)存在 就將該節(jié)點(diǎn)的前節(jié)點(diǎn)指向tail
                node.prev = t;
                //設(shè)置當(dāng)前節(jié)點(diǎn)為tail
                if (compareAndSetTail(t, node)) {
                //前一個(gè)節(jié)點(diǎn)的next指向當(dāng)前節(jié)點(diǎn)，入隊(duì)操作就完成了
                    t.next = node;
                    return t;
                }
            }
        }
    }

設(shè)置waitStatus狀態(tài)及獲取waitStatus狀態(tài)，waitStatus為-1時(shí)可以喚醒后續(xù)節(jié)點(diǎn)

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
        int ws = pred.waitStatus;
        if (ws == Node.SIGNAL)
            /*
             * 狀態(tài)是-1 就可以喚醒后續(xù)節(jié)點(diǎn)
             *
             */
            return true;
        if (ws > 0) {
            /*
             * 前置任務(wù)已取消，刪掉節(jié)點(diǎn)
             *
             */
            do {
                node.prev = pred = pred.prev;
            } while (pred.waitStatus > 0);
            pred.next = node;
        } else {
            /*
             * cas設(shè)置waitstatus狀態(tài)，設(shè)置為-1
             * 
             */
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        }
        return false;
    }

阻塞調(diào)用LockSupport.park

 private final boolean parkAndCheckInterrupt() {
        LockSupport.park(this);
        return Thread.interrupted();
    }

釋放鎖的邏輯

	public void release() {
        sync.releaseShared(1);
    }
	public final boolean releaseShared(int arg) {
		//cas成功則進(jìn)行釋放共享鎖
        if (tryReleaseShared(arg)) {
            doReleaseShared();
            return true;
        }
        return false;
    }

state狀態(tài)+1操作，cas成功，返回true

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

private void doReleaseShared() {
        for (;;) {
            Node h = head;
            //頭節(jié)點(diǎn)不為空并且不是尾節(jié)點(diǎn)
            if (h != null && h != tail) {
                int ws = h.waitStatus;
                //waitstatus為-1
                if (ws == Node.SIGNAL) {
                //將SIGNAL置為0
                    if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                        continue;         
                    //喚醒   
                    unparkSuccessor(h);
                }
                else if (ws == 0 &&
                         !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                    continue;               
            }
            if (h == head)                  
                break;
        }
    }

喚醒操作

 private void unparkSuccessor(Node node) {
        int ws = node.waitStatus;
        //ws小于0就將其設(shè)置為0
        if (ws < 0)
            compareAndSetWaitStatus(node, ws, 0);
        Node s = node.next;
        //當(dāng)前節(jié)點(diǎn)的下一個(gè)節(jié)點(diǎn)為空或者ws大于0
        if (s == null || s.waitStatus > 0) {
            s = null;
            for (Node t = tail; t != null && t != node; t = t.prev)
                if (t.waitStatus <= 0)
                    s = t;
        }
        //s不為空 則進(jìn)行喚醒
        if (s != null)
            LockSupport.unpark(s.thread);
    }

喚醒下一個(gè)線程之后，要把上一個(gè)節(jié)點(diǎn)移除隊(duì)列

 private void doAcquireSharedInterruptibly(int arg)
        throws InterruptedException {
        final Node node = addWaiter(Node.SHARED);
        boolean failed = true;
        try {
            for (;;) {
                final Node p = node.predecessor();
                //下一個(gè)線程進(jìn)來，如果前置節(jié)點(diǎn)是頭節(jié)點(diǎn)，則將前置節(jié)點(diǎn)出隊(duì)
                if (p == head) {
                    int r = tryAcquireShared(arg);
                    //cas獲取資源成功
                    if (r >= 0) {
                    	//出隊(duì)操作
                        setHeadAndPropagate(node, r);
                        //將p.next移除
                        p.next = null;
                        failed = false;
                        return;
                    }
                }
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    throw new InterruptedException();
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

出隊(duì)操作

private void setHeadAndPropagate(Node node, int propagate) {
		//設(shè)置當(dāng)前節(jié)點(diǎn)為head節(jié)點(diǎn)，前一節(jié)點(diǎn)的head屬性被刪除
        Node h = head; 
        setHead(node);
        //如果是傳播屬性
        if (propagate > 0 || h == null || h.waitStatus < 0 ||
            (h = head) == null || h.waitStatus < 0) {
           //并且是共享模式，可以持續(xù)喚醒下一個(gè)，
           //只要資源數(shù)充足 就可以一直往下喚醒，提高并發(fā)量
            Node s = node.next;
            if (s == null || s.isShared())
                doReleaseShared();
        }
    }
 private void setHead(Node node) {
        head = node;
        node.thread = null;
        node.prev = null;
    }

至此，線程的阻塞喚醒核心邏輯就這么多，共享鎖與獨(dú)占鎖的區(qū)別是可以喚醒后續(xù)的線程，如果資源數(shù)充足的話，可以一直往下喚醒，提高了并發(fā)量。

到此這篇關(guān)于Java AQS信號(hào)量Semaphore的使用的文章就介紹到這了,更多相關(guān)Java信號(hào)量Semaphore內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: