0%

CyclicBarrier、CountDownLatch 和 Semaphore 的比较

Java 提供了很多同步工具,最近看Java锁的时候,看到了 CountDownLatch 和 Semaphore,同时了解到还有 CyclicBarrier,就想对它们做一个一定的了解。

CountDownLatch

CountDownLatch 是在完成一组正在其他线程中执行的操作之前,它允许一个或多个线程一直等待。使用一个计数器进行实践,计数器初始值为线程的数量。每当一个线程完成后,计数器的值就会减一。当计数器的值为0时,便是所有线程都完成了任务,然后等待的线程就可以恢复执行任务了。

用给定的计数 初始化 CountDownLatch。由于调用了 countDown() 方法,所以在当前计数到达零之前,await 方法会一直受阻塞。之后,会释放所有等待的线程,await 的所有后续调用都将立即返回。这种现象只出现一次——计数无法被重置。如果需要重置计数,请考虑使用 CyclicBarrier。

初始化

1
CountDownLatch countDownLatch = new CountDownLatch(1);

CountDownLatch 是一次性的,计数值在构建的时候就已经初始化完成了。

await

提供两种方法是当前线程处于等待状态

1
2
3
4
5
6
7
public void await() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}
public boolean await(long timeout, TimeUnit unit)
throws InterruptedException {
return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
}

countDown

当调用 countDown() 方法时,CountDownLatch 计数器就会减去一,而当计数器为0的时候,等待的线程就会被唤醒并执行线程。

CyclicBarrier

CyclicBarrier 是一个同步工具,允许一组线程互相等待,直到到达某个公共屏障点。与CountDownLatch不同的是该barrier在释放等待线程后可以重用,所以称它为循环(Cyclic)的屏障(Barrier)。

初始化

1
2
3
4
5
6
7
8
9
10
//parties表示屏障拦截的线程数量,当屏障撤销时,先执行barrierAction,然后在释放所有线程
public CyclicBarrier(int parties, Runnable barrierAction) {
if (parties <= 0) throw new IllegalArgumentException();
this.parties = parties;
this.count = parties;
this.barrierCommand = barrierAction;
}
public CyclicBarrier(int parties) {
this(parties, null);
}

await()

1
2
3
4
5
6
7
8
9
10
11
12
13
public int await() throws InterruptedException, BrokenBarrierException {
try {
return dowait(false, 0L);
} catch (TimeoutException toe) {
throw new Error(toe); // cannot happen
}
}
public int await(long timeout, TimeUnit unit)
throws InterruptedException,
BrokenBarrierException,
TimeoutException {
return dowait(true, unit.toNanos(timeout));
}

reset()

将屏障重置为其初始状态。如果所有参与者目前都在屏障处等待,则它们将返回,同时抛出一个BrokenBarrierException。

Semaphore

Semaphore 是Java中提供的信号量,它是一个计数信号量,必须由获取线程释放。

初始化

1
2
3
4
5
6
7
8
9
//默认构建非公平锁  
public Semaphore(int permits) {
sync = new NonfairSync(permits);
}

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

获取许可: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
32
33
34
35
public void acquire() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}
public final void acquireSharedInterruptibly(int arg)
throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
if (tryAcquireShared(arg) < 0)
doAcquireSharedInterruptibly(arg);
}
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);
}
}

释放许可: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
27
28
29
30
31
32
33
34
35
36
37
38
39
public void release() {
sync.releaseShared(1);
}
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {
doReleaseShared();
return true;
}
return false;
}
private void doReleaseShared() {
/*
* Ensure that a release propagates, even if there are other
* in-progress acquires/releases. This proceeds in the usual
* way of trying to unparkSuccessor of head if it needs
* signal. But if it does not, status is set to PROPAGATE to
* ensure that upon release, propagation continues.
* Additionally, we must loop in case a new node is added
* while we are doing this. Also, unlike other uses of
* unparkSuccessor, we need to know if CAS to reset status
* fails, if so rechecking.
*/
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;
}
}

总结

  • CountDownLatch:允许一个或多个线程等待,直到在某些线程中执行的一组操作完成
  • CyclicBarrier:一个辅助性的同步结构,允许多个线程等待到达某个屏障
  • Semaphore:Java 版本的信号量实现
客官,赏一杯coffee嘛~~~~