Java併發包源碼學習系列：同步組件Semaphore源碼解析

時間  2021-02-22
標籤 java 編程 併發 ide 工具 學習 .net 線程 code 欄目 Java 简体版
原文   原文鏈接

目錄

Semaphore概述及案例學習

Semaphore信號量用來控制同時訪問特定資源的線程數量，它經過協調各個線程，以保證合理地使用公共資源。java

public class SemaphoreTest {

    private static final int THREAD_COUNT = 30;
    private static ExecutorService threadPool = Executors.newFixedThreadPool(THREAD_COUNT);

    private static Semaphore s = new Semaphore(10); //10個許可證數量,最大併發數爲10

    public static void main(String[] args) {
        for(int i = 0; i < THREAD_COUNT; i ++){ //執行30個線程
            threadPool.execute(new Runnable() {
                @Override
                public void run() {
                    s.tryAcquire(); //嘗試獲取一個許可證
                    System.out.println("save data");
                    s.release(); //使用完以後歸還許可證
                }
            });
        }
        threadPool.shutdown();
    }
}
  • 建立一個大小爲30的線程池，可是信號量規定在10，保證許可證數量爲10。
  • 每次線程調用tryAcquire()或者acquire()方法都會原子性的遞減許可證的數量，release()會原子性遞增許可證數量。

類圖結構及重要字段

public class Semaphore implements java.io.Serializable {
    private static final long serialVersionUID = -3222578661600680210L;
    /** All mechanics via AbstractQueuedSynchronizer subclass */
    private final Sync sync;
    
    abstract static class Sync extends AbstractQueuedSynchronizer {
        // permits指定初始化信號量個數
        Sync(int permits) {
            setState(permits);
        }
        // ...
    }
    
    static final class NonfairSync extends Sync {...}
    
    static final class FairSync extends Sync {...}
    
    // 默認採用非公平策略
    public Semaphore(int permits) {
        sync = new NonfairSync(permits);
    }
	
    // 能夠指定公平策略
    public Semaphore(int permits, boolean fair) {
        sync = fair ? new FairSync(permits) : new NonfairSync(permits);
    }
    
    //...
}
  • 基於AQS，相似於ReentrantLock，Sync繼承自AQS，有公平策略和非公平策略兩種實現。
  • 相似於CountDownLatch，state在這裏也是經過構造器指定，表示初始化信號量的個數。

本篇文章閱讀須要創建在必定的AQS基礎之上，這邊推薦幾篇前置文章，能夠瞅一眼：編程

void acquire()

調用該方法時，表示但願獲取一個信號量資源，至關於acquire(1)。併發

若是當前信號量個數大於0，CAS將當前信號量值減1，成功後直接返回。ide

若是當前信號量個數等於0，則當前線程將被置入AQS的阻塞隊列。工具

該方法是響應中斷的，其餘線程調用了該線程的interrupt()方法，將會拋出中斷異常返回。學習

// Semaphore.java
	public void acquire() throws InterruptedException {
        // 傳遞的 arg 爲 1 ， 獲取1個信號量資源
        sync.acquireSharedInterruptibly(1);
    }
	// AQS.java
    public final void acquireSharedInterruptibly(int arg)
            throws InterruptedException {
        // 線程被 中斷， 拋出中斷異常
        if (Thread.interrupted())
            throw new InterruptedException();
        // 子類實現， 公平和非公平兩種策略
        if (tryAcquireShared(arg) < 0)
            // 若是獲取失敗， 則置入阻塞隊列， 
            // 再次進行嘗試， 嘗試失敗則掛起當前線程
            doAcquireSharedInterruptibly(arg);
    }

非公平

static final class NonfairSync extends Sync {
        private static final long serialVersionUID = -2694183684443567898L;

        NonfairSync(int permits) {
            super(permits);
        }

        protected int tryAcquireShared(int acquires) {
            // 這裏直接調用Sync定義的 非公平共享模式獲取方法
            return nonfairTryAcquireShared(acquires);
        }
    }

    abstract static class Sync extends AbstractQueuedSynchronizer {
        
        final int nonfairTryAcquireShared(int acquires) {
            for (;;) {
                // 獲取當前信號量的值
                int available = getState();
                // 減去須要獲取的值， 獲得剩餘的信號量個數
                int remaining = available - acquires;
                // 不剩了，表示當前信號量個數不能知足需求， 返回負數， 線程置入AQS阻塞
                // 還有的剩， CAS設置當前信號量值爲剩餘值， 並返回剩餘值
                if (remaining < 0 ||
                    compareAndSetState(available, remaining))
                    return remaining;
            }
        }
    }

你會發現，非公平策略是沒法保證【AQS隊列中阻塞的線程】和【當前線程】獲取的順序的，當前線程是有可能在排隊的線程以前就拿到資源，產生插隊現象。ui

公平策略就不同了，它會經過hasQueuedPredecessors()方法看看隊列中是否存在前驅節點，以保證公平性。.net

公平策略

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

void acquire(int permits)

在acquire()的基礎上，指定了獲取信號量的數量permits。線程

public void acquire(int permits) throws InterruptedException {
        if (permits < 0) throw new IllegalArgumentException();
        sync.acquireSharedInterruptibly(permits);
    }

void acquireUninterruptibly()

該方法與acquire()相似，可是不響應中斷。code

public void acquireUninterruptibly() {
        sync.acquireShared(1);
    }
	
    public final void acquireShared(int arg) {
        if (tryAcquireShared(arg) < 0)
            doAcquireShared(arg);
    }

void acquireUninterruptibly(int permits)

該方法與acquire(permits)相似，可是不響應中斷。

public void acquireUninterruptibly(int permits) {
        if (permits < 0) throw new IllegalArgumentException();
        sync.acquireShared(permits);
    }

boolean tryAcquire()

tryAcquire和acquire非公平策略公用一個邏輯，可是區別在於，若是獲取信號量失敗，或者CAS失敗，將會直接返回false，而不會置入阻塞隊列中。

通常try開頭的方法的特色就是這樣，嘗試一下，成功是最好，失敗也不至於被阻塞，而是馬上返回false。

public boolean tryAcquire() {
        return sync.nonfairTryAcquireShared(1) >= 0;
    }
    abstract static class Sync extends AbstractQueuedSynchronizer {
        final int nonfairTryAcquireShared(int acquires) {
            for (;;) {
                int available = getState();
                int remaining = available - acquires;
                if (remaining < 0 ||
                    compareAndSetState(available, remaining))
                    return remaining;
            }
        }
    }

boolean tryAcquire(int permits)

相比於普通的tryAcquire()，指定了permits的值。

public boolean tryAcquire(int permits) {
        if (permits < 0) throw new IllegalArgumentException();
        return sync.nonfairTryAcquireShared(permits) >= 0;
    }

boolean tryAcquire(int permits, long timeout, TimeUnit unit)

相比於tryAcquire(int permits)，增長了超時控制。

public boolean tryAcquire(int permits, long timeout, TimeUnit unit)
        throws InterruptedException {
        if (permits < 0) throw new IllegalArgumentException();
        return sync.tryAcquireSharedNanos(permits, unit.toNanos(timeout));
    }

void release()

將信號量值加1，若是有線程由於調用acquire方法而被阻塞在AQS阻塞隊列中，將根據公平策略選擇一個信號量個數知足需求的線程喚醒，線程喚醒後也會嘗試獲取新增的信號量。

參考文章：Java併發包源碼學習系列：AQS共享模式獲取與釋放資源

// Semaphore.java
	public void release() {
        sync.releaseShared(1);
    }
	// AQS.java
    public final boolean releaseShared(int arg) {
        // 嘗試釋放鎖
        if (tryReleaseShared(arg)) {
            // 釋放成功， 喚醒AQS隊列裏面最早掛起的線程
            // https://blog.csdn.net/Sky_QiaoBa_Sum/article/details/112386838
            doReleaseShared();
            return true;
        }
        return false;
    }
	// Semaphore#Sync.java
    abstract static class Sync extends AbstractQueuedSynchronizer {
        protected final boolean tryReleaseShared(int releases) {
            for (;;) {
                // 獲取當前信號量
                int current = getState();
                // 指望加上releases
                int next = current + releases;
                if (next < current) // overflow
                    throw new Error("Maximum permit count exceeded");
                // CAS操做，更新
                if (compareAndSetState(current, next))
                    return true;
            }
        }
    }

void release(int permits)

和release()相比指定了permits的值。

public void release(int permits) {
        if (permits < 0) throw new IllegalArgumentException();
        sync.releaseShared(permits);
    }

其餘方法

Semaphore還提供其餘一些方法，實現比較簡單，這邊就簡單寫一下吧：

// 返回此信號量中當前可用的許可證數量， 其實就是獲得當前的 state值  getState()
	public int availablePermits() {
        return sync.getPermits();
    }

    // 將state更新爲0， 返回0
    public int drainPermits() {
        return sync.drainPermits();
    }

	// 減小reduction個許可證
    protected void reducePermits(int reduction) {
        if (reduction < 0) throw new IllegalArgumentException();
        sync.reducePermits(reduction);
    }

    // 判斷公平策略
    public boolean isFair() {
        return sync instanceof FairSync;
    }

	// 判斷是否有線程證在等待獲取許可證
    public final boolean hasQueuedThreads() {
        return sync.hasQueuedThreads();
    }

	// 返回正在等待獲取許可證的線程數
    public final int getQueueLength() {
        return sync.getQueueLength();
    }

	// 返回全部等待獲取許可證的線程集合
    protected Collection<Thread> getQueuedThreads() {
        return sync.getQueuedThreads();
    }

總結

Semaphore信號量用來控制同時訪問特定資源的線程數量，它經過協調各個線程，以保證合理地使用公共資源。

  • 基於AQS，相似於ReentrantLock，Sync繼承自AQS，有公平策略和非公平策略兩種實現。
  • 相似於CountDownLatch，state在這裏也是經過構造器指定，表示初始化信號量的個數。

每次線程調用tryAcquire()或者acquire()方法都會原子性的遞減許可證的數量，release()會原子性遞增許可證數量，只要有許可證就能夠重複使用。

參考閱讀

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