本篇文章给大家分享的是有关java多线程中的并发工具类CountDownLatch,CyclicBarrier和Semaphore该怎么理解，小编觉得挺实用的，因此分享给大家学习，希望大家阅读完这篇文章后可以有所收获，话不多说，跟着小编一起来看看吧。
CountDownLatch
CountDownLatch允许一个或多个线程等待其他线程完成操作。
假设一个Excel文件有多个sheet，我们需要去记录每个sheet有多少行数据，
这时我们就可以使用CountDownLatch实现主线程等待所有sheet线程完成sheet的解析操作后，再继续执行自己的任务。
public class CountDownLatchTest { private static class WorkThread extends Thread { private CountDownLatch cdl; public WorkThread(String name, CountDownLatch cdl) { super(name); this.cdl = cdl; } public void run() { System.out.println(this.getName() + "启动了，时间为" + System.currentTimeMillis()); System.out.println(this.getName() + "我要统计每个sheet的行数"); try { cdl.await(); Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println(this.getName() + "执行完了，时间为" + System.currentTimeMillis()); } } private static class sheetThread extends Thread { private CountDownLatch cdl; public sheetThread(String name, CountDownLatch cdl) { super(name); this.cdl = cdl; } public void run() { try { System.out.println(this.getName() + "启动了，时间为" + System.currentTimeMillis()); Thread.sleep(1000); //模拟任务执行耗时 cdl.countDown(); System.out.println(this.getName() + "执行完了，时间为" + System.currentTimeMillis() + " sheet的行数为：" + (int) (Math.random()*100)); } catch (InterruptedException e) { e.printStackTrace(); } } } public static void main(String[] args) throws Exception { CountDownLatch cdl = new CountDownLatch(2); WorkThread wt0 = new WorkThread("WorkThread", cdl ); wt0.start(); sheetThread dt0 = new sheetThread("sheetThread1", cdl); sheetThread dt1 = new sheetThread("sheetThread2", cdl); dt0.start(); dt1.start(); }}执行结果：
WorkThread启动了，时间为1640054503027
WorkThread我要统计每个sheet的行数
sheetThread1启动了，时间为1640054503028
sheetThread2启动了，时间为1640054503029
sheetThread2执行完了，时间为1640054504031 sheet的行数为：6
sheetThread1执行完了，时间为1640054504031 sheet的行数为：44
WorkThread执行完了，时间为1640054505036
可以看到，首先WorkThread执行await后开始等待，WorkThread在等待sheetThread1和sheetThread2都执行完自己的任务后，WorkThread立刻继续执行后面的代码。
CountDownLatch的构造函数接收一个int类型的参数作为计数器，如果你想等待N个点完成，这里就传入N。
当我们调用CountDownLatch的countDown方法时，N就会减1，CountDownLatch的await方法会阻塞当前线程，直到N变成零。
由于countDown方法可以用在任何地方，所以这里说的N个点，可以是N个线程，也可以是1个线程里的N个执行步骤。
用在多个线程时，只需要把这个CountDownLatch的引用传递到线程里即可。
我们继续根据上面的测试案例流程，一步一步的分析CountDownLatch 源码。
第一步看CountDownLatch的构造方法，传入一个不能小于0的int类型的参数作为计数器
public CountDownLatch(int count) { if (count < 0) throw new IllegalArgumentException("count < 0"); this.sync = new Sync(count); } private static final class Sync extends AbstractQueuedSynchronizer { private static final long serialVersionUID = 4982264981922014374L; Sync(int count) { setState(count); } int getCount() { return getState(); } protected int tryAcquireShared(int acquires) { return (getState() == 0) ? 1 : -1; } protected boolean tryReleaseShared(int releases) { // Decrement count; signal when transition to zero for (;;) { int c = getState(); if (c == 0) return false; int nextc = c-1; if (compareAndSetState(c, nextc)) return nextc == 0; } } }看它的注释，说的非常清楚，Sync就是CountDownLatch的同步控制器了，而它也是继承了AQS，并且第3行注释说到使用了AQS的state去代表count值。
第二步就是工作线程调用await()方法
public void await() 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); }如果线程中断，抛出异常，否则开始调用tryAcquireShared(1)，其内部类Sync的实现也非常简单，就是判断state也就是CountDownLatch的计数是否等于0，
如果等于0，则该方法返回1，第5行的if判断不成立，否则该方法返回-1，第5行的if判断成立，继续执行doAcquireSharedInterruptibly(1)。
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); } }这个方法其实就是去获取共享模式下的锁，获取失败就park住。正如我们测试案例中的WorkThread线程应该次数就被park住了，那么它又是何时被唤醒的呢？
下面就到countDown()方法了
public void countDown() { sync.releaseShared(1); }public final boolean releaseShared(int arg) { if (tryReleaseShared(arg)) { doReleaseShared(); return true; } return false; }tryReleaseShared(1)方法尝试去释放共享锁
protected boolean tryReleaseShared(int releases) { // Decrement count; signal when transition to zero for (;;) { int c = getState(); if (c == 0) return false; int nextc = c-1; if (compareAndSetState(c, nextc)) return nextc == 0; } }在for循环中，先获取CountDownLatch的计数也就是当前state，如果等于0返回false，否则将state更新为state-1，并返回最新的state是否等于0。
因此在我们的测试案例中，我们需要调用两次countDown方法，才会将全局的state更新为0，然后继续执行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; } } private void unparkSuccessor(Node node) { int ws = node.waitStatus; if (ws < 0) compareAndSetWaitStatus(node, ws, 0); Node s = node.next; if (s == null || s.waitStatus > 0) { s = null; for (Node t = tail; t != null && t != node; t = t.prev) if (t.waitStatus <= 0) s = t; } if (s != null) LockSupport.unpark(s.thread); }LockSupport.unpark(s.thread)，唤醒线程的方法被调用后，WorkThread线程就可以继续执行了。
至此我们简单分析了整个测试案例中CountDownLatch的代码流程。
Semaphore
Semaphore（信号量）是用来控制同时访问特定资源的线程数量，相当于一个并发控制器，构造的时候传入可供管理的信号量的数值，这个数值就是用来控制并发数量的，
每个线程执行前先通过acquire方法获取信号，执行后通过release归还信号 。每次acquire返回成功后，Semaphore可用的信号量就会减少一个，如果没有可用的信号，
acquire调用就会阻塞，等待有release调用释放信号后，acquire才会得到信号并返回。
下面我们看个测试案例
public class SemaphoreTest { public static void main(String[] args) { final Semaphore semaphore = new Semaphore(5); Runnable runnable = () -> { try { semaphore.acquire(); System.out.println(Thread.currentThread().getName() + "获得了信号量>>>>>,时间为" + System.currentTimeMillis()); Thread.sleep(1000); System.out.println(Thread.currentThread().getName() + "释放了信号量<<<<<,时间为" + System.currentTimeMillis()); } catch (InterruptedException e) { e.printStackTrace(); } finally { semaphore.release(); } }; Thread[] threads = new Thread[10]; for (int i = 0; i < threads.length; i++) threads[i] = new Thread(runnable); for (int i = 0; i < threads.length; i++) threads[i].start(); }}执行结果：
Thread-0获得了信号量>>>>>,时间为1640058647604
Thread-1获得了信号量>>>>>,时间为1640058647604
Thread-2获得了信号量>>>>>,时间为1640058647604
Thread-3获得了信号量>>>>>,时间为1640058647605
Thread-4获得了信号量>>>>>,时间为1640058647605
Thread-0释放了信号量<<<<<,时间为1640058648606
Thread-1释放了信号量<<<<<,时间为1640058648606
Thread-5获得了信号量>>>>>,时间为1640058648607
Thread-4释放了信号量<<<<<,时间为1640058648607
Thread-3释放了信号量<<<<<,时间为1640058648607
Thread-7获得了信号量>>>>>,时间为1640058648607
Thread-8获得了信号量>>>>>,时间为1640058648607
Thread-2释放了信号量<<<<<,时间为1640058648606
Thread-6获得了信号量>>>>>,时间为1640058648607
Thread-9获得了信号量>>>>>,时间为1640058648607
Thread-7释放了信号量<<<<<,时间为1640058649607
Thread-6释放了信号量<<<<<,时间为1640058649607
Thread-8释放了信号量<<<<<,时间为1640058649607
Thread-9释放了信号量<<<<<,时间为1640058649608
Thread-5释放了信号量<<<<<,时间为1640058649607
我们使用for循环同时创建10个线程，首先是线程 0 1 2 3 4获得了信号量，再后面的10行打印结果中，线程1到5分别释放信号量，相同线程间隔也是1000毫秒，然后线程5 6 7 8 9才能继续获得信号量，而且保持最大获取信号量的线程数小于等于5。
看下Semaphore的构造方法
public Semaphore(int permits) { sync = new NonfairSync(permits); }public Semaphore(int permits, boolean fair) { sync = fair ? new FairSync(permits) : new NonfairSync(permits); }它支持传入一个int类型的permits，一个布尔类型的fair，因此Semaphore也有公平模式与非公平模式。
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; } } }第9行代码可见Semaphore也是通过AQS的state来作为信号量的计数的
第12行 getPermits() 方法获取当前的可用的信号量，即还有多少线程可以同时获得信号量
第15行nonfairTryAcquireShared方法尝试获取共享锁，逻辑就是直接将可用信号量减去该方法请求获取的数量，更新state并返回该值。
第24行tryReleaseShared 方法尝试释放共享锁，逻辑就是直接将可用信号量加上该方法请求释放的数量，更新state并返回。
再看下Semaphore的公平锁
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; } } }看尝试获取共享锁的方法中，多了个 if (hasQueuedPredecessors) 的判断，在java多线程6：ReentrantLock，
分析过hasQueuedPredecessors其实就是判断当前等待队列中是否存在等待线程，并判断第一个等待的线程(head.next)是否是当前线程。
CyclicBarrier
CyclicBarrier的字面意思是可循环使用（Cyclic）的屏障（Barrier）。它要做的事情是，让一组线程到达一个屏障（也可以叫同步点）时被阻塞，直到最后一个线程到达屏障时，屏障才会开门，所有被屏障拦截的线程才会继续运行。
一组线程同时被唤醒，让我们想到了ReentrantLock的Condition，它的signalAll方法可以唤醒await在同一个condition的所有线程。
下面我们还是从一个简单的测试案例先了解下CyclicBarrier的用法
public class CyclicBarrierTest extends Thread { private CyclicBarrier cb; private int sleepSecond; public CyclicBarrierTest(CyclicBarrier cb, int sleepSecond) { this.cb = cb; this.sleepSecond = sleepSecond; } public void run() { try { System.out.println(this.getName() + "开始, 时间为" + System.currentTimeMillis()); Thread.sleep(sleepSecond * 1000); cb.await(); System.out.println(this.getName() + "结束, 时间为" + System.currentTimeMillis()); } catch (Exception e) { e.printStackTrace(); } } public static void main(String[] args) { Runnable runnable = new Runnable() { public void run() { System.out.println("CyclicBarrier的barrierAction开始运行, 时间为" + System.currentTimeMillis()); } }; CyclicBarrier cb = new CyclicBarrier(2, runnable); CyclicBarrierTest cbt0 = new CyclicBarrierTest(cb, 3); CyclicBarrierTest cbt1 = new CyclicBarrierTest(cb, 6); cbt0.start(); cbt1.start(); }}执行结果：
Thread-1开始, 时间为1640069673534
Thread-0开始, 时间为1640069673534
CyclicBarrier的barrierAction开始运行, 时间为1640069679536
Thread-1结束, 时间为1640069679536
Thread-0结束, 时间为1640069679536
可以看到Thread-0和Thread-1同时运行，而自定义的线程barrierAction是在6000毫秒后开始执行，说明Thread-0在await之后，等待了3000毫秒，和Thread-1一起继续执行的。
看下CyclicBarrier 的一个更高级的构造函数
public CyclicBarrier(int parties, Runnable barrierAction) { if (parties <= 0) throw new IllegalArgumentException(); this.parties = parties; this.count = parties; this.barrierCommand = barrierAction; }parties就是设定需要多少线程在屏障前等待，只有调用await方法的线程数达到才能唤醒所有的线程，还有注意因为使用CyclicBarrier的线程都会阻塞在await方法上，所以在线程池中使用CyclicBarrier时要特别小心，如果线程池的线程过少，那么就会发生死锁。
Runnable barrierAction用于在线程到达屏障时，优先执行barrierAction，方便处理更复杂的业务场景。
private int dowait(boolean timed, long nanos) throws InterruptedException, BrokenBarrierException, TimeoutException { final ReentrantLock lock = this.lock; lock.lock(); try { final Generation g = generation; if (g.broken) throw new BrokenBarrierException(); if (Thread.interrupted()) { breakBarrier(); throw new InterruptedException(); } int index = --count; if (index == 0) { // tripped boolean ranAction = false; try { final Runnable command = barrierCommand; if (command != null) command.run(); ranAction = true; nextGeneration(); return 0; } finally { if (!ranAction) breakBarrier(); } } // loop until tripped, broken, interrupted, or timed out for (;;) { try { if (!timed) trip.await(); else if (nanos > 0L) nanos = trip.awaitNanos(nanos); } catch (InterruptedException ie) { if (g == generation && ! g.broken) { breakBarrier(); throw ie; } else { // We're about to finish waiting even if we had not // been interrupted, so this interrupt is deemed to // "belong" to subsequent execution. Thread.currentThread().interrupt(); } } if (g.broken) throw new BrokenBarrierException(); if (g != generation) return index; if (timed && nanos <= 0L) { breakBarrier(); throw new TimeoutException(); } } } finally { lock.unlock(); } }首先是ReentrantLock加锁，全局的count值-1，然后判断count是否等于0，如果不等于0，则循环，condition执行await等待，直到触发、中断、中断或超时，如果count值等于0，先执行barrierAction线程，然后condition开始唤醒所有等待的线程。
简单是使用之后，有人会觉得CyclicBarrier和CountDownLatch有点像，其实它们两者有些细微的差别：
CountDownLatch是在多个线程都进行了latch.countDown()后才会触发事件，唤醒await()在latch上的线程，而执行countDown()的线程，是不会阻塞的；
CyclicBarrier是一个栅栏，用于同步所有调用await()方法的线程，线程执行了await()方法之后并不会执行之后的代码，而只有当执行await()方法的线程数等于指定的parties之后，这些执行了await()方法的线程才会同时运行。
CountDownLatch不能循环使用，计数器减为0就减为0了，不能被重置；CyclicBarrier本是就是支持循环使用parties，而且提供了reset()方法，可以重置计数器。
以上就是java多线程中的并发工具类CountDownLatch,CyclicBarrier和Semaphore该怎么理解，小编相信有部分知识点可能是我们日常工作会见到或用到的。希望你能通过这篇文章学到更多知识。更多详情敬请关注编程网行业资讯频道。
