Java的信號量semaphore講解

更新時間：2023年12月22日 09:47:39 作者：正經人z.

這篇文章主要介紹了Java的信號量semaphore講解,Semaphore底層是基于AbstractQueuedSynchronizer來實現的,Semaphore稱為計數信號量,它允許n個任務同時訪問某個資源,需要的朋友可以參考下

一、簡介

1、Semaphore底層是基于AbstractQueuedSynchronizer來實現的。Semaphore稱為計數信號量，它允許n個任務同時訪問某個資源，可以將信號量看做是在向外分發(fā)使用資源的許可證，只有成功獲取許可證，才能使用資源。

2、它也是通過內部類Sync繼承了AQS，它的方法幾乎都是用的AQS的，少部分重寫了AQS的方法，同時，它的信號量機制是通過AQS的state屬性來操作的，變更state屬性使用到了CAS操作保證變量的數據一致性。

3、它的方法幾乎是通過Sync類，也就是AQS實現的。

4、該類如果單獨使用，不涉及到條件隊列condition，使用到條件隊列需要調用await方法、notify方法，notifyAll方法等一系列線程通信的方法，但是，semaphore類中不涉及到這些方法。

5、注意其中一個方法：release，這個方法是釋放信號量到semaphore中，默認是釋放一個。源碼中作者這么說的：不要求一個線程在調用release方法釋放許可前必須通過acquire方法獲取到許可。正確的使用semaphore建立在應用程序里面，對許可證的使用，由程序員掌握！semaphore只是提供了一個信號量的機制供其使用。

6、核心方法都在Sync類中，外部方法都是調用的AQS和Sync中重寫的AQS的方法

7、注意其中一個方法：acquire，這個方法是獲取信號量從semaphore中，默認是獲取一個。源碼中作者這么說的：如果沒有可用的許可證，則當前線程變?yōu)槌鲇诰€程調度目的而禁用，并處于休眠狀態(tài)，直到發(fā)生兩件事之一：

a、一些其他的線程調用了release方法釋放了許可，同時semaphore中的許可滿足當前線程，且當前線程是下一個被分配許可證的線程

b、一些其他的線程打斷了當前線程且當前線程是通過acquire方式獲取許可的

二、源碼分析

三個內部類

Sync

abstract static class Sync extends AbstractQueuedSynchronizer {
        private static final long serialVersionUID = 1192457210091910933L;
    //構造方法設置許可數
        Sync(int permits) {
            setState(permits);
        }
    //獲取許可
        final int getPermits() {
            return getState();
        }
    //共享模式下非公平策略獲取許可數
        final int nonfairTryAcquireShared(int acquires) {
            for (;;) {   //無限循環(huán)
                int available = getState();
                int remaining = available - acquires;
                if (remaining < 0 ||
                    compareAndSetState(available, remaining))  //通過CAS修改許可數
                    return remaining;
            }
        }
    //共享模式下的公平策略釋放許可
        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))   //通過CAS修改許可數
                    return true;
            }
        }
    //減去許可數
        final void reducePermits(int reductions) {
            for (;;) {
                int current = getState();
                int next = current - reductions;
                if (next > current) // underflow
                    throw new Error("Permit count underflow");
                if (compareAndSetState(current, next))
                    return;
            }
        }
// 獲取并返回立即可用的所有許可
        final int drainPermits() {
            for (;;) {
                int current = getState();
                if (current == 0 || compareAndSetState(current, 0))
                    return current;
            }
        }
    }

Sync類繼承了AQS類，重寫了nonfairTryAcquireShared、tryReleaseShared方法

NonfairSync和FairSync

//非公平策略	
static final class NonfairSync extends Sync {
        private static final long serialVersionUID = -2694183684443567898L;
        NonfairSync(int permits) {
            super(permits);
        }
        protected int tryAcquireShared(int acquires) {
            return nonfairTryAcquireShared(acquires);  //調用的父類Sync的方法
        }
    }
//公平策略
    static final class FairSync extends Sync {
        private static final long serialVersionUID = 2014338818796000944L;
        FairSync(int permits) {
            super(permits);
        }
        //公平策略下子類重寫了方法
        protected int tryAcquireShared(int acquires) {
            for (;;) {
                if (hasQueuedPredecessors())
                    return -1;
                int available = getState();
                int remaining = available - acquires;
                if (remaining < 0 ||
                    compareAndSetState(available, remaining))
                    return remaining;
            }
        }
    }

構造方法

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

常用方法

//默認獲取一個許可
public void acquire() throws InterruptedException {
        sync.acquireSharedInterruptibly(1);
    }
//默認釋放一個許可
public void release() {sync.releaseShared(1); }
//獲取指定的許可數    
public void acquire(int permits) throws InterruptedException {
        if (permits < 0) throw new IllegalArgumentException();
        sync.acquireSharedInterruptibly(permits);
    }
//嘗試獲取指定的許可數    
public boolean tryAcquire(int permits) {
        if (permits < 0) throw new IllegalArgumentException();
        return sync.nonfairTryAcquireShared(permits) >= 0;
    }
//釋放指定的許可數    
public void release(int permits) {
        if (permits < 0) throw new IllegalArgumentException();
        sync.releaseShared(permits);
    }

沒啥好說的，都是套娃子模式，真正的核心方法在Sync類中以及AQS類

示例

package com.hust.grid.leesf.semaphore;
import java.util.concurrent.Semaphore;
class MyThread extends Thread {
    private Semaphore semaphore;
    public MyThread(String name, Semaphore semaphore) {
        super(name);
        this.semaphore = semaphore;
    }
    public void run() {        
        int count = 3;
        System.out.println(Thread.currentThread().getName() + " trying to acquire");
        try {
            semaphore.acquire(count);
            System.out.println(Thread.currentThread().getName() + " acquire successfully");
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            semaphore.release(count);
            System.out.println(Thread.currentThread().getName() + " release successfully");
        }
    }
}
public class SemaphoreDemo {
    public final static int SEM_SIZE = 10;
    public static void main(String[] args) {
        Semaphore semaphore = new Semaphore(SEM_SIZE);
        MyThread t1 = new MyThread("t1", semaphore);
        MyThread t2 = new MyThread("t2", semaphore);
        t1.start();
        t2.start();
        int permits = 5;
        System.out.println(Thread.currentThread().getName() + " trying to acquire");
        try {
            semaphore.acquire(permits);
            System.out.println(Thread.currentThread().getName() + " acquire successfully");
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            semaphore.release();
            System.out.println(Thread.currentThread().getName() + " release successfully");
        }
    }
}

說明：首先，生成一個信號量，信號量有10個許可，然后，main，t1，t2三個線程獲取許可運行

首先，main線程執(zhí)行acquire操作，并且成功獲得許可，之后t1線程執(zhí)行acquire操作，成功獲得許可，之后t2執(zhí)行acquire操作，由于此時許可數量不夠，t2線程將會阻塞，直到許可可用。之后t1線程釋放許可，main線程釋放許可，此時的許可數量可以滿足t2線程的要求，所以，此時t2線程會成功獲得許可運行，t2運行完成后釋放許可。

三、總結

1、semaphore類的核心是Sync內部類，它繼承了AQS類，適當重寫了一些方法，其他的方法都調用的這個Sync中的方法，包括Sync類的兩個子類：FairSync和NonFairSync。

2、semaphore類中實現了公平鎖FairSync和非公平鎖NonFairSync。默認使用的是非公平鎖。

3、對于公平鎖和非公平鎖，主要體現在獲取鎖上面是否公平。非公平鎖不會判斷當前線程是否為同步隊列中的第一個節(jié)點，而是直接操作state屬性；公平鎖會判斷當前線程是否為同步隊列中的第一個節(jié)點，具體方法是hasQueuedPredecessors，如果不是，則返回-1，如果是，則執(zhí)行下面步驟。

4、semaphore的信號量機制使用的是AQS類的state屬性，默認每次獲取或釋放信號量都是1，除非你指定要使用的信號量或釋放的信號量數。

5、對state屬性的添加和釋放都必須保證是原子性的，所以，semaphore類中使用的是unsafe類的compareAndSetState方法配合for(; ;)無限循環(huán)，語義為CAS自旋，來保證對state屬性的操作是原子性的。

到此這篇關于Java的信號量semaphore講解的文章就介紹到這了,更多相關信號量semaphore內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: