Java并发包的CountDownLatch、CyclicBarrier、Semaphore

2025-01-16 技术教程

1.CountDownLatch
CountDownLatch.wait()会使当前线程阻塞，直到CountDownLatch中的计数器递减完毕后继续执行。

public class CountDownLatchDemo {private final static Logger log = Logger.getLogger(AtomicIntegerCyclicBarrierDemo.class);public static void main(String[] args) { final CountDownLatch latch = new CountDownLatch(2); new Thread(){ public void run() { log.info(Thread.currentThread().getName() + " running"); latch.countDown(); }; }.start(); new Thread(){ public void run() { try { log.info(Thread.currentThread().getName() + " running"); Thread.sleep(5000); latch.countDown(); } catch (InterruptedException e) { e.printStackTrace(); } }; }.start(); try { latch.await(); log.info("all finished"); } catch (InterruptedException e) { e.printStackTrace(); } }}

运行日志：
2019-10-24 09:23:24,069-[HL][17] INFO Thread-0 AtomicIntegerCyclicBarrierDemo - Thread-0 running
2019-10-24 09:23:24,075-[HL][24] INFO Thread-1 AtomicIntegerCyclicBarrierDemo - Thread-1 running
2019-10-24 09:23:29,076-[HL][34] INFO main AtomicIntegerCyclicBarrierDemo - all finished
见第三条09:23:29，主线程一直阻塞直到线程2完成后才继续向下执行

2.CyclicBarrier
CyclicBarrier可以比做一个阻拦器，当同时阻塞的线程数达到某个个数时，才会将所有线程放行，这里结合同样是并发包中的AtomicInteger实现了一个案例，AtomicInteger时基于CAS实现的乐观锁，也就是说当修改值与预期值不一致时，AtomicInteger会修改失败并重试，也就是说当我们用多个线程同时累加，在所有线程执行完毕之后，结果是安全的

public class AtomicIntegerCyclicBarrierDemo {private final static Logger log = Logger.getLogger(AtomicIntegerCyclicBarrierDemo.class);private final static AtomicInteger count = new AtomicInteger(0);private final static int THREAD_COUNT = 100; public static void main(String[] args) { CyclicBarrier barrier = new CyclicBarrier(THREAD_COUNT, new Runnable() { public void run() { log.info(String.format("result = %s", count.get())); } }); for (int i = 0; i < THREAD_COUNT; i++) { new Task(barrier, count).start(); } }}class Task extends Thread {private CyclicBarrier cyclicBarrier;private AtomicInteger count;public Task(CyclicBarrier cyclicBarrier, AtomicInteger count) { this.cyclicBarrier = cyclicBarrier; this.count = count;}@Overridepublic void run() { super.run(); try { count.addAndGet(2); cyclicBarrier.await(); } catch (InterruptedException e) { e.printStackTrace(); } catch (BrokenBarrierException e) { e.printStackTrace(); }}}

运行日志：
2019-10-24 09:40:11,961-[HL][22] INFO Thread-99 AtomicIntegerCyclicBarrierDemo - result = 200
CyclicBarrier帮助我们在所有线程执行完之后，进行结果的打印

3.Semaphore
Semaphore用来控制可同时运行的线程数，比如一个案例，我们有一些模型转换任务，为了控制计算资源占用，只允许同时3个线程同时运行，每当一个线程运行完才能空出一个槽位给新的线程运行，下面见代码

public class SemaphoreDemo {private final static int THREAD_COUNT = 8 ;public static void main(String[] args) { Semaphore semaphore = new Semaphore(3); for(int i = 0; i < THREAD_COUNT ; i++) { try { Thread.sleep(2000); new Task2(i, semaphore).start(); } catch (InterruptedException e) { e.printStackTrace(); } }}}class Task2 extends Thread{private final static Logger log = Logger.getLogger(Task2.class);private int num;private Semaphore semaphore;public Task2(int num,Semaphore semaphore){ this.num = num; this.semaphore = semaphore;}@Overridepublic void run() { try { semaphore.acquire(); log.info(String.format("task(id = %s) opeater", this.num)); Thread.sleep(10000); log.info(String.format("task(id = %s) finished", this.num)); semaphore.release(); } catch (InterruptedException e) { e.printStackTrace(); }}}

运行日志：
2019-10-24 09:51:49,180-[HL][39] INFO Thread-0 Task2 - task(id = 0) opeater
2019-10-24 09:51:51,150-[HL][39] INFO Thread-1 Task2 - task(id = 1) opeater
2019-10-24 09:51:53,151-[HL][39] INFO Thread-2 Task2 - task(id = 2) opeater
2019-10-24 09:51:59,190-[HL][41] INFO Thread-0 Task2 - task(id = 0) finished
2019-10-24 09:51:59,190-[HL][39] INFO Thread-3 Task2 - task(id = 3) opeater
2019-10-24 09:52:01,151-[HL][41] INFO Thread-1 Task2 - task(id = 1) finished
2019-10-24 09:52:01,152-[HL][39] INFO Thread-4 Task2 - task(id = 4) opeater
2019-10-24 09:52:03,151-[HL][41] INFO Thread-2 Task2 - task(id = 2) finished
2019-10-24 09:52:03,159-[HL][39] INFO Thread-5 Task2 - task(id = 5) opeater
这里用Executors.newFixedThreadPool也能实现同样效果