四虎综合网,一区二区三区av,在线观看免费黄色

前言

控制并發流程的工具類，作用就是幫助我們程序員更容易的讓線程之間合作，讓線程之間相互配合來滿足業務邏輯。比如讓線程A等待線程B執行完畢后再執行等合作策略。

控制并發流程的工具類主要有：

簡介

Semaphore 信號量，許可，用于控制在一段時間內，可并發訪問執行的線程數量。它的作用是控制訪問特定資源的線程數目，底層依賴AQS的狀態State，是在生產當中比較常用的一個工具類。

關于 AQS，可以查看《并發編程之抽象隊列同步器AQS應用ReentrantLock》

一個信號量有且僅有 3 種操作，且它們全部是原子的。

初始化、增加和減少。
增加可以為一個進程解除阻塞。
減少可以讓一個進程進入阻塞。

Semaphore 管理一系列許可證。

每個 acquire() 方法阻塞，直到有一個許可證可以獲得然后拿走一個許可證。
每個 release() 方法增加一個許可證，這可能會釋放一個阻塞的 acquire() 方法。
不使用實際的許可對象，Semaphore 只對可用許可的號碼進行計數，并采取相應的行動。

Semaphore 在計數器不為 0 的時候對線程就放行，一旦達到 0，那么所有請求資源的新線程都會被阻塞，包括增加請求到許可的線程，Semaphore 是不可重入的。

每一次請求一個許可都會導致計數器減少 1，同樣每次釋放一個許可都會導致計數器增加 1，一旦達到 0，新的許可請求線程將被掛起。

Semaphore 有兩種模式，公平模式和非公平模式，默認是非公平模式。

公平模式就是調用 acquire 的順序就是獲取許可證的順序，遵循 FIFO。
非公平模式是搶占式的，也就是有可能一個新的獲取線程恰好在一個許可證釋放時得到了這個許可證，而前面還有等待的線程。

應用場景

Semaphore可以用來做流量限制，特別是公共資源有限的應用場景，比如說數據庫連接。

由于 release() 釋放許可時，未對釋放許可數做限制，所有可以通過該方法增加總的許可數量; reducePermits() 方法可以減少總的許可數量，通過這兩個方法可以到達動態調整許可的

分析：假如有一個需求，需讀取幾個萬個文件的數據，因為都是IO密集型，我們可以啟動幾十個線程并發的讀取，但是如果讀取到內存后，還需要存儲到數據庫，而數據庫的連接數只有10個，這時候我們就必須要控制只有10個線程同時獲取到數據庫連接，否則會拋出異常提示無法連接數據庫。針對這種情況，我們就可以使用Semaphore來做流量控制。

代碼如下：

package com.niuh.tools;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
/**
* <p>
* Semaphore示例
* </p>
*/
public class SemaphoreRunner {
/**
* 線程數量
*/
private static final int THREAD_COUNT = 30;
/**
* 線程池
*/
private static ExecutorService executor = Executors.newFixedThreadPool(THREAD_COUNT);
private static Semaphore semaphore = new Semaphore(10);
public static void main(String[] args) {
for (int i = 0; i < THREAD_COUNT; i++) {
executor.execute(new Runnable() {
public void run() {
try {
// 獲取一個"許可證"
semaphore.acquire();
// 模擬數據保存
TimeUnit.SECONDS.sleep(2);
System.out.println("save date...");
// 執行完后,歸還"許可證"
semaphore.release();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
}
executor.shutdown();
}
}

源碼分析

Semaphore 類圖

Semaphore 通過使用內部類 Syn 繼承 AQS 實現。

其內部主要變量和方法如下：

并發編程之Semaphore原理與應用

框架流程圖如下：

構造函數

permits 表示許可線程的數量
fair 表示公平性，如果這個設為 true 的話，下次執行的線程會是等待最久的線程

public Semaphore(int permits) {
sync = new NonfairSync(permits);
}
/**
* @param permits 總許可數
* @param fair fair=true 公平鎖 fair=false 非公平鎖
*/
public Semaphore(int permits, boolean fair) {
sync = fair ? new FairSync(permits) : new NonfairSync(permits);
}

內部類同步器

abstract static class Sync extends AbstractQueuedSynchronizer {
private static final long serialVersionUID = 1192457210091910933L;
// 賦值setState為總許可數
Sync(int permits) {
setState(permits);
}
// 剩余許可數
final int getPermits() {
return getState();
}
// 自旋 + CAS 非公平獲取
final int nonfairTryAcquireShared(int acquires) {
for (;;) {
// 剩余可用許可數
int available = getState();
// 本次獲取許可后，剩余許可
int remaining = available - acquires;
// 如果獲取后，剩余許可大于0，則CAS更新剩余許可，否則獲取更新失敗
if (remaining < 0 ||
compareAndSetState(available, remaining))
return remaining;
}
}
// 自旋 + CAS 釋放許可
// 由于未對釋放許可數做限制，所以可以通過release動態增加許可數量
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");
// CAS更新許可數量
if (compareAndSetState(current, next))
return true;
}
}
// 自旋 + CAS 減少許可數量
final void reducePermits(int reductions) {
for (;;) {
// 當前剩余許可
int current = getState();
// 更新值
int next = current - reductions;
// 如果更新值比當前剩余許可大，拋出益處
if (next > current) // underflow
throw new Error("Permit count underflow");
// CAS 更新許可數
if (compareAndSetState(current, next))
return;
}
}
// 丟棄所有許可
final int drainPermits() {
for (;;) {
int current = getState();
if (current == 0 || compareAndSetState(current, 0))
return current;
}
}
}

非公平模式

/**
* 非公平模式
*/
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);
}
}

公平模式

/**
* 公平模式
*/
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;
// 自旋 + CAS獲取許可
int available = getState();
int remaining = available - acquires;
if (remaining < 0 ||
compareAndSetState(available, remaining))
return remaining;
}
}
}

獲取許可

Semaphore 提供了兩種獲取資源的方式。

響應中斷 和 不響應中斷。

響應中斷獲取資源

兩個方法支持 Interrupt 中斷機制，可使用 acquire() 方法每次獲取一個信號量，也可以使用 acquire(int permits) 方法獲取指定數量的信號量。

從semaphore中獲取一個許可，線程會一直被阻塞直到獲取一個許可或是被中斷，獲取一個許可后立即返回，并把許可數減1，如果沒有可用的許可，當前線程會處于休眠狀態直到：

某些其他線程調用release方法，并且當前線程是下一個要被分配許可的線程
某些其他線程中斷當前線程

如果當前線程被acquire方法使得中斷狀態設置為on或者在等待許可時被中斷則拋出InterruptedException，并且清除當前線程的已中斷狀態。

acquire執行流程：

public void acquire() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}
public void acquire(int permits) throws InterruptedException {
if (permits < 0) throw new IllegalArgumentException();
sync.acquireSharedInterruptibly(permits);
}
public final void acquireSharedInterruptibly(int arg)
throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
// 獲取許可，剩余許可 >= 0,則獲取許可成功 <0 獲取許可失敗，進入排隊
if (tryAcquireShared(arg) < 0)
doAcquireSharedInterruptibly(arg);
}
/**
* 獲取許可失敗，當前線程進入同步隊列，排隊阻塞
* @param arg the acquire argument
*/
private void doAcquireSharedInterruptibly(int arg)
throws InterruptedException {
// 創建同步隊列節點，并入隊列
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);
}
}

代碼的執行步驟如下：

AQS 子類使用共享模式，需要實現 tryAcquireShared() 方法。

在公平鎖中還是與ReentrantLock中的操作一樣，先判斷同步隊列中是不是還有其他的等待線程，否則直接返回失敗。否則對 state 值進行減操作并返回剩下的信號量。
非公平鎖直接調用了父類中的 nonfairTryAcquireShared 和 ReentrantLock 一樣。

// 非公平鎖的獲取方式
protected int tryAcquireShared(int acquires) {
return nonfairTryAcquireShared(acquires);
}
final int nonfairTryAcquireShared(int acquires) {
for (;;) {
int available = getState();//獲取去中的信號量數
int remaining = available - acquires;//剩余信號量數
//1.信號量數大于0,獲取共享鎖,并設置執行compareAndSetState(available, remaining),返回剩余信號量數
//2.信號量數小于等于0,直接返回負數
if (remaining < 0 || compareAndSetState(available, remaining))
return remaining;
}
}
// 公平鎖獲取
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;
}
}

變量 state 采用 volatile 可見修飾。

/**
* The synchronization state.
*/
private volatile int state;
/**
* Returns the current value of synchronization state.
* This operation has memory semantics of a <tt>volatile</tt> read.
* @return current state value
*/
protected final int getState() {
return state;
}

不響應中斷獲取資源

兩個方法不響應 Interrupt 中斷機制，其它功能與 acquire() 方法一致。

某些其他線程調用release方法，并且當前線程是下一個要被分配許可的線程;
如果當前線程在等待許可時被中斷，那么它會接著等待，但是與沒有發生中斷相比，為線程分配許可的時間可能改變。

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

嘗試獲得信號量

嘗試獲得信號量有三個方法。

嘗試獲取信號量，如果獲取成功則返回 true，否則馬上返回 false，不會阻塞當前線程。
嘗試獲取信號量，如果在指定的時間內獲得信號量，則返回 true，否則返回 false。
嘗試獲取指定數量的信號量，如果在指定的時間內獲得信號量，則返回 true，否則返回 false。

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

釋放歸還許可

release 方法，主要作用是釋放資源，需要保證 release 的執行，否則線程退出但是資源沒有釋放。

一般代碼寫在 finally 中是最好的。
并且獲取多少資源就要釋放多少資源，否則還是資源沒被正確釋放，如果一開始執行了 acquire(10) 最后釋放的時候不能只寫一個 release() 而是 release(10) 才對。

// 嘗試釋放鎖
public final boolean release(int arg) {
// 如果釋放鎖成功喚醒同步隊列中的后繼節點
if (tryRelease(arg)) {
Node h = head;
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);
return true;
}
return false;
}
// 為了方便對比把兩個代碼放在一塊可以看到 release 中的結構完全一樣
// 區別就在于 doReleaseShared 中有更多的判斷操作
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {
doReleaseShared(); //在里面執行的 unparkSuccessor(h)
return true;
}
return false;
}

子類實現共享模式的類需要實現 tryReleaseShared() 方法判斷是否釋放成功。

這個方法是一個 CAS 自旋，原因是因為 Semaphore 是一個共享鎖，可能有多個線程同時釋放資源，因此 CAS 操作可能失敗。

// 由于未對釋放許可數做限制，所以可以通過release動態增加許可數量
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");
//CAS改變許可數量成功，返回true
if (compareAndSetState(current, next))
return true;
}
}

一旦 CAS 改變許可數量成功，就調用 doReleaseShared() 方法釋放阻塞的線程。

private void doReleaseShared() {
// 自旋，喚醒等待的第一個線程（其它線程將由第一個線程向后傳遞喚醒）
for (;;) {
Node h = head;
if (h != null && h != tail) {
int ws = h.waitStatus;
if (ws == Node.SIGNAL) {
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
continue; // loop to recheck cases
// 喚醒第一個等待線程
unparkSuccessor(h);
}
else if (ws == 0 &&
!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
continue; // loop on failed CAS
}
if (h == head) // loop if head changed
break;
}
}

其他方法

獲取當前剩余的信號量數量

該方法返回 AQS 中 state 變量的值，當前剩余的信號量個數。

public int availablePermits() {
return sync.getPermits();
}
// Sync
final int getPermits() {
return getState();
}

耗盡許可數量

獲取并返回立即可用的所有許可。
Sync 類的drainPermits()方法，獲取 1 個信號量后將可用的信號量個數置為 0。例如總共有 10 個信號量，已經使用了 5 個，再調用 drainPermits() 方法后，可以獲得一個信號量，剩余 4 個信號量就消失了，總共可用的信號量就變成 6 個了。用 CAS 自旋將剩余資源清空。

public int drainPermits() {
return sync.drainPermits();
}
// Sync
final int drainPermits() {
for (;;) {
int current = getState();
if (current == 0 || compareAndSetState(current, 0))
return current;
}
}

縮減許可數量

縮減必須是單向的，即只能減少不能增加。用 CAS 自旋在剩余共享資源上做縮減。

protected void reducePermits(int reduction) {
if (reduction < 0) throw new IllegalArgumentException();
sync.reducePermits(reduction);
}
// Sync
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;
}
}

上述兩個方法對共享資源數量的修改操作有兩點需要注意

是不可逆的
是對剩余資源的操作而不是全部資源，當剩余資源數目不足或已經為 0 時，方法就返回。
正在被占用的資源不參與。

判斷 AQS 同步隊列中是否還有 Node

public final boolean hasQueuedThreads() {
return sync.hasQueuedThreads();
}
// AbstractQueuedSynchronizer
public final boolean hasQueuedThreads() {
//頭結點不等于尾節點就說明鏈表中還有元素
return head != tail;
}

總結

Semaphore 的內部工作流程也是基于 AQS，不同于 CyclicBarrier 和 ReentrantLock，不會使用到 AQS 的條件隊列，都是在同步隊列中操作，只是當前線程會被 park。

Semaphore 是 JUC 包提供的一個典型的共享鎖，它通過自定義兩種不同的同步器(FairSync 和 NonfairSync)提供了公平和非公平兩種工作模式，兩種模式下分別提供了限時/不限時、響應中斷/不響應中斷的獲取資源的方法(限時獲取總是及時響應中斷的)，而所有的釋放資源的 release() 操作是統一的。

PS：以上代碼提交在 Github ：

https://github.com/Niuh-Study/niuh-juc-final.git