Semaphore:

A counting semaphore. Conceptually, a semaphore maintains a set of permits. Each {@link #acquire} blocks if necessary until a permit is available, and then takes it. Each {@link #release} adds a permit,potentially releasing a blocking acquirer.However, no actual permit objects are used; the {@code Semaphore} just keeps a count of the number available and acts accordingly.

翻译过来就是:

一个计数信号量。概念上来说,就是一个由信号量维护的一组许可证。每个 acquire() 都会阻塞直到许可证可用才能使用它。每个 release()添加许可证,潜在地释放阻塞获取方。但是,没有使用实际的许可证对象; Semaphore 只保留可用数量的计数,并相应地执行。信号量通常用于限制线程数,而不是访问某些(物理或逻辑)资源

Semaphore 通常用于限制只有特定数量的线程才能访问相同的资源。

Sync

Semaphore 内部实现基于公平非公平竞争的 AQS 实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
/** All mechanics via AbstractQueuedSynchronizer subclass */
private final Sync sync;

/**
* Synchronization implementation for semaphore. Uses AQS state
* to represent permits. Subclassed into fair and nonfair
* versions.
*/
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 (;;) {
int available = getState();
int remaining = available - acquires;
if (remaining < 0 ||
compareAndSetState(available, remaining))
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))
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;
}
}
}

/**
* NonFair version
*/
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);
}
}

/**
* Fair version
*/
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;
}
}
}

Semaphore 只实现了通过共享模式获取锁资源。Semaphore 同时支持公平非公平方式获取资源。

  • Semaphore 默认使用 非公平 竞争
  • 非公平竞争下,调用的是 SyncnonfairTryAcquireShared
  • 公平竞争下,会首先调用 AQShasQueuedPredecessors 判断是否当前线程是等待时间最长

核心方法

acquire

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
/**
* Acquires a permit from this semaphore, blocking until one is
* available, or the thread is {@linkplain Thread#interrupt interrupted}.
*
* <p>Acquires a permit, if one is available and returns immediately,
* reducing the number of available permits by one.
*
* <p>If no permit is available then the current thread becomes
* disabled for thread scheduling purposes and lies dormant until
* one of two things happens:
* <ul>
* <li>Some other thread invokes the {@link #release} method for this
* semaphore and the current thread is next to be assigned a permit; or
* <li>Some other thread {@linkplain Thread#interrupt interrupts}
* the current thread.
* </ul>
*
* <p>If the current thread:
* <ul>
* <li>has its interrupted status set on entry to this method; or
* <li>is {@linkplain Thread#interrupt interrupted} while waiting
* for a permit,
* </ul>
* then {@link InterruptedException} is thrown and the current thread's
* interrupted status is cleared.
*
* @throws InterruptedException if the current thread is interrupted
*/
public void acquire() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}

Semaphore 中获取一个通行证,该方法会阻塞,知道获取到通行证或线程被中断。

acquireUninterruptibly

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
/**
* Acquires a permit from this semaphore, blocking until one is
* available.
*
* <p>Acquires a permit, if one is available and returns immediately,
* reducing the number of available permits by one.
*
* <p>If no permit is available then the current thread becomes
* disabled for thread scheduling purposes and lies dormant until
* some other thread invokes the {@link #release} method for this
* semaphore and the current thread is next to be assigned a permit.
*
* <p>If the current thread is {@linkplain Thread#interrupt interrupted}
* while waiting for a permit then it will continue to wait, but the
* time at which the thread is assigned a permit may change compared to
* the time it would have received the permit had no interruption
* occurred. When the thread does return from this method its interrupt
* status will be set.
*/
public void acquireUninterruptibly() {
sync.acquireShared(1);
}

acquire,不同点是该方法会忽略线程中断操作。

tryAcquire

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
/**
* Acquires a permit from this semaphore, only if one is available at the
* time of invocation.
*
* <p>Acquires a permit, if one is available and returns immediately,
* with the value {@code true},
* reducing the number of available permits by one.
*
* <p>If no permit is available then this method will return
* immediately with the value {@code false}.
*
* <p>Even when this semaphore has been set to use a
* fair ordering policy, a call to {@code tryAcquire()} <em>will</em>
* immediately acquire a permit if one is available, whether or not
* other threads are currently waiting.
* This &quot;barging&quot; behavior can be useful in certain
* circumstances, even though it breaks fairness. If you want to honor
* the fairness setting, then use
* {@link #tryAcquire(long, TimeUnit) tryAcquire(0, TimeUnit.SECONDS) }
* which is almost equivalent (it also detects interruption).
*
* @return {@code true} if a permit was acquired and {@code false}
* otherwise
*/
public boolean tryAcquire() {
return sync.nonfairTryAcquireShared(1) >= 0;
}

/**
* Acquires a permit from this semaphore, if one becomes available
* within the given waiting time and the current thread has not
* been {@linkplain Thread#interrupt interrupted}.
*
* <p>Acquires a permit, if one is available and returns immediately,
* with the value {@code true},
* reducing the number of available permits by one.
*
* <p>If no permit is available then the current thread becomes
* disabled for thread scheduling purposes and lies dormant until
* one of three things happens:
* <ul>
* <li>Some other thread invokes the {@link #release} method for this
* semaphore and the current thread is next to be assigned a permit; or
* <li>Some other thread {@linkplain Thread#interrupt interrupts}
* the current thread; or
* <li>The specified waiting time elapses.
* </ul>
*
* <p>If a permit is acquired then the value {@code true} is returned.
*
* <p>If the current thread:
* <ul>
* <li>has its interrupted status set on entry to this method; or
* <li>is {@linkplain Thread#interrupt interrupted} while waiting
* to acquire a permit,
* </ul>
* then {@link InterruptedException} is thrown and the current thread's
* interrupted status is cleared.
*
* <p>If the specified waiting time elapses then the value {@code false}
* is returned. If the time is less than or equal to zero, the method
* will not wait at all.
*
* @param timeout the maximum time to wait for a permit
* @param unit the time unit of the {@code timeout} argument
* @return {@code true} if a permit was acquired and {@code false}
* if the waiting time elapsed before a permit was acquired
* @throws InterruptedException if the current thread is interrupted
*/
public boolean tryAcquire(long timeout, TimeUnit unit) throws InterruptedException {
return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
}

Semaphore 中获取一个通行证,该方法会立即返回是否成功获取到通信证。同时支持超时获取,若超时仍未获取到通行证会返回 false.

release

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
/**
* Releases the given number of permits, returning them to the semaphore.
*
* <p>Releases the given number of permits, increasing the number of
* available permits by that amount.
* If any threads are trying to acquire permits, then one
* is selected and given the permits that were just released.
* If the number of available permits satisfies that thread's request
* then that thread is (re)enabled for thread scheduling purposes;
* otherwise the thread will wait until sufficient permits are available.
* If there are still permits available
* after this thread's request has been satisfied, then those permits
* are assigned in turn to other threads trying to acquire permits.
*
* <p>There is no requirement that a thread that releases a permit must
* have acquired that permit by calling {@link Semaphore#acquire acquire}.
* Correct usage of a semaphore is established by programming convention
* in the application.
*
* @param permits the number of permits to release
* @throws IllegalArgumentException if {@code permits} is negative
*/
public void release(int permits) {
if (permits < 0) throw new IllegalArgumentException();
sync.releaseShared(permits);
}

release 用于释放通行证。释放令牌的过程也就是通过CASAQSstate +1

availablePermits

1
2
3
4
5
6
7
8
9
10
/**
* Returns the current number of permits available in this semaphore.
*
* <p>This method is typically used for debugging and testing purposes.
*
* @return the number of permits available in this semaphore
*/
public int availablePermits() {
return sync.getPermits();
}

该方法会返回当前 Semaphore 中可用的通行证数量

drainPermits

1
2
3
4
5
6
7
8
/**
* Acquires and returns all permits that are immediately available.
*
* @return the number of permits acquired
*/
public int drainPermits() {
return sync.drainPermits();
}

该方法会清空令牌把可用令牌数置为0,返回清空令牌的数量。

示例代码

  1. 停车场容纳总停车量10
  2. 当一辆车进入停车场后,显示牌的剩余车位数响应的减1.
  3. 每有一辆车驶出停车场后,显示牌的剩余车位数响应的加1
  4. 停车场剩余车位不足时,车辆只能在外面等待。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
public static void main(String[] args) {
// 停车场同时容纳的车辆 10
Semaphore semaphore = new Semaphore(10);
for (int i = 0; i < 100; i++) {
new Thread(() -> {
String threadName = Thread.currentThread().getName();
System.out.println(String.format("====【%s】来到停车场===", threadName));
if (semaphore.availablePermits() == 0) {
System.out.println("车位不足,请耐心等待");
}
try {
semaphore.acquire();// 获取令牌尝试进入停车场
System.out.println(String.format("【%s】成功进入停车场", threadName));
Thread.sleep(new Random().nextInt(10000));// 模拟车辆在停车场停留的时间
System.out.println(String.format("【%s】驶出停车场", threadName));
semaphore.release();//释放令牌,腾出停车场车位
} catch (InterruptedException e) {
e.printStackTrace();
}
}, String.valueOf(i + 1)).start();
}
}

参考

Semaphore 使用及原理