CountDownLatch
是一个同步辅助类,它允许一个或多个线程等待,直到其他线程完成一组操作。CountDownLatch
有一个计数器,当计数器减为0时,等待的线程将被唤醒。计数器只能减少,不能增加。
示例:使用CountDownLatch等待所有线程完成任务
假设我们有一个任务需要三个子任务完成,我们可以使用CountDownLatch
来等待所有子任务完成。
import java.util.concurrent.CountDownLatch; public class CountDownLatchExample { public static void main(String[] args) throws InterruptedException { CountDownLatch latch = new CountDownLatch(3); for (int i = 1; i <= 3; i++) { final int taskNumber = i; new Thread(() -> { System.out.println("Task " + taskNumber + " started"); try { Thread.sleep(1000 * taskNumber); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("Task " + taskNumber + " completed"); latch.countDown(); }).start(); } System.out.println("Waiting for all tasks to complete..."); latch.await(); System.out.println("All tasks completed"); } }
在这个示例中,我们创建了一个CountDownLatch
并设置初始计数器为3。每个子任务完成后,调用latch.countDown()
减少计数器。主线程调用latch.await()
等待所有子任务完成。
CyclicBarrier
是一个同步辅助类,它允许一组线程相互等待,直到所有线程都准备好继续执行。当所有线程都到达屏障点时,屏障将打开。CyclicBarrier
可以重复使用。
示例:使用CyclicBarrier同步多个线程
假设我们有三个线程需要在某个点同步,我们可以使用CyclicBarrier
实现这个目的。
import java.util.concurrent.BrokenBarrierException; import java.util.concurrent.CyclicBarrier; public class CyclicBarrierExample { public static void main(String[] args) { CyclicBarrier barrier = new CyclicBarrier(3, () -> System.out.println("All threads are ready to proceed")); for (int i = 1; i <= 3; i++) { final int taskNumber = i; new Thread(() -> { System.out.println("Task " + taskNumber + " is ready"); try { barrier.await(); } catch (InterruptedException | BrokenBarrierException e) { e.printStackTrace(); } System.out.println("Task " + taskNumber + " is proceeding"); }).start(); } } }
在这个示例中,我们创建了一个CyclicBarrier
并设置参与者数量为3。每个线程在准备好继续执行之前调用barrier.await()
。当所有线程都准备好时,屏障将打开,所有线程将继续执行。
Semaphore
是一个计数信号量,它维护了一个许可集。线程可以请求许可,如果有可用的许可,线程将获得许可并继续执行。否则,线程将阻塞,直到有可用的许可。许可可以由任何线程释放。Semaphore
可用于实现资源池、限制并发访问等。
示例:使用Semaphore限制并发访问
假设我们有一个只能同时处理三个请求的服务器,我们可以使用Semaphore
来实现并发访问限制。
import java.util.concurrent.Semaphore; public class SemaphoreExample { public static void main(String[] args) { Semaphore semaphore = new Semaphore(3); for (int i = 1; i <= 10; i++) final int clientNumber = i; new Thread(() -> { try { System.out.println("Client " + clientNumber + " is trying to connect"); semaphore.acquire(); System.out.println("Client " + clientNumber + " is connected"); Thread.sleep(2000); System.out.println("Client " + clientNumber + " is disconnected"); } catch (InterruptedException e) { e.printStackTrace(); } finally { semaphore.release(); } }).start(); } } }
在这个示例中,我们创建了一个Semaphore
并设置初始许可数量为3。每个客户端线程在连接服务器之前调用semaphore.acquire()
请求许可。当许可可用时,线程将获得许可并继续执行。线程完成后,调用semaphore.release()
释放许可。
Exchanger
是一个同步辅助类,它允许两个线程在一个临界点交换数据。当两个线程都到达交换点时,它们将交换数据。Exchanger
可以用于遗传算法、管道设计等。
示例:使用Exchanger交换数据
假设我们有两个线程,一个生成数据,另一个处理数据。我们可以使用Exchanger
在这两个线程之间交换数据。
import java.util.concurrent.Exchanger; public class ExchangerExample { public static void main(String[] args) { Exchanger<String> exchanger = new Exchanger<>(); new Thread(() -> { try { String data = "Data from producer"; System.out.println("Producer is sending: " + data); String receivedData = exchanger.exchange(data); System.out.println("Producer received: " + receivedData); } catch (InterruptedException e) { e.printStackTrace(); } }).start(); new Thread(() -> { try { String data = "Data from consumer"; System.out.println("Consumer is sending: " + data); String receivedData = exchanger.exchange(data); System.out.println("Consumer received: " + receivedData); } catch (InterruptedException e) { e.printStackTrace(); } }).start(); } }
在这个示例中,我们创建了一个Exchanger
。生产者和消费者线程在交换数据前调用exchanger.exchange(data)
。当两个线程都到达交换点时,它们将交换数据。
Phaser
是一个灵活的同步辅助类,它允许一组线程相互等待,直到所有线程都准备好继续执行。与CyclicBarrier
类似,但Phaser
更加灵活,可以动态调整参与者数量和支持多个阶段。
示例:使用Phaser同步多个线程
假设我们有三个线程需要在某个点同步,我们可以使用Phaser
实现这个目的。
import java.util.concurrent.Phaser; public class PhaserExample { public static void main(String[] args) { Phaser phaser = new Phaser(3); for (int i = 1; i <= 3; i++) { final int taskNumber = i; new Thread(() -> { System.out.println("Task " + taskNumber + " is ready"); phaser.arriveAndAwaitAdvance(); System.out.println("Task " + taskNumber + " is proceeding"); }).start(); } } }
在这个示例中,我们创建了一个Phaser
并设置参与者数量为3。每个线程在准备好继续执行之前调用phaser.arriveAndAwaitAdvance()
。当所有线程都准备好时,屏障将打开,所有线程将继续执行。
这些并发工具类为Java多线程编程提供了强大的支持,帮助我们更轻松地实现各种同步和并发场景。希望这些示例能帮助你理解并掌握这些工具类的用法。
推荐阅读:
https://mp.weixin.qq.com/s/dV2JzXfgjDdCmWRmE0glDA
https://mp.weixin.qq.com/s/an83QZOWXHqll3SGPYTL5g