Java并发工具类是构建高效、可靠多线程应用程序的关键。本文将深入解析CountDownLatch、CyclicBarrier、Semaphore、BlockingQueue、ConcurrentHashMap、CopyOnWriteArrayList、AtomicInteger、AtomicReference以及Future和CompletableFuture等核心工具类，帮助开发者掌握多线程编程的精髓。
在Java中，随着多核处理器的普及和并发需求的增加，多线程编程成为现代软件开发中不可或缺的一部分。Java提供了丰富的并发工具类，如CountDownLatch、CyclicBarrier、Semaphore、阻塞队列（BlockingQueue）、并发集合（ConcurrentHashMap和CopyOnWriteArrayList）、原子类（AtomicInteger和AtomicReference）以及Future和CompletableFuture，帮助开发者更高效地处理复杂的并发场景。本文将逐一介绍这些工具类，并探讨其使用场景和实现原理。
CountDownLatch、CyclicBarrier、Semaphore
CountDownLatch
CountDownLatch是一个同步工具类，用于协调多个线程之间的执行顺序。它允许一个或多个线程等待直到其他线程完成特定操作。CountDownLatch通过一个计数器实现，当计数器减到零时，等待的线程可以继续执行。
主要方法：
- countDown()：将计数器减1。
- await()：使当前线程等待，直到计数器为0。
使用场景： CountDownLatch通常用于“等待所有线程完成初始化后再继续执行”的场景。例如，主线程需要等待多个子线程完成数据加载后才能进行后续处理。
示例代码：
import java.util.concurrent.CountDownLatch;
public class CountDownLatchExample {
public static void main(String[] args) throws InterruptedException {
CountDownLatch latch = new CountDownLatch(3);
// 启动3个子线程
for (int i = 0; i < 3; i++) {
new Thread(() -> {
System.out.println("Thread is ready");
latch.countDown(); // 减少计数
}).start();
}
latch.await(); // 主线程等待，直到计数为0
System.out.println("All threads are ready, now main thread can proceed");
}
}
CyclicBarrier
CyclicBarrier是一个同步辅助工具类，它使得一组线程相互等待，直到所有线程都到达某个公共屏障点。与CountDownLatch不同，CyclicBarrier是可重用的，可以重复使用。
主要方法：
- await()：使当前线程等待，直到所有线程到达屏障点。
- reset()：重置屏障，允许重复使用。
使用场景： CyclicBarrier适用于多线程并行执行某些任务，然后在某个时刻等待所有线程完成某个操作后再继续执行。例如，多线程在完成各自任务后，需要在某个点汇总结果。
示例代码：
import java.util.concurrent.CyclicBarrier;
public class CyclicBarrierExample {
public static void main(String[] args) throws InterruptedException {
int threadsCount = 3;
CyclicBarrier barrier = new CyclicBarrier(threadsCount, () -> {
System.out.println("All threads are ready, let's continue execution");
});
// 启动3个子线程
for (int i = 0; i < threadsCount; i++) {
new Thread(() -> {
try {
System.out.println(Thread.currentThread().getName() + " is doing some work");
barrier.await(); // 等待其他线程
} catch (Exception e) {
e.printStackTrace();
}
}).start();
}
}
}
Semaphore
Semaphore是一个计数信号量，用于控制同时访问某个特定资源的线程数量。它可以通过限制线程数来防止系统超负荷。
主要方法：
- acquire()：获取信号量，成功获取后计数减1。
- release()：释放信号量，计数加1。
使用场景： Semaphore常用于控制对数据库、文件、网络等共享资源的并发访问。例如，限制最多3个线程同时访问某个资源。
示例代码：
import java.util.concurrent.Semaphore;
public class SemaphoreExample {
public static void main(String[] args) throws InterruptedException {
Semaphore semaphore = new Semaphore(3); // 限制最多3个线程并发访问
// 启动5个线程
for (int i = 0; i < 5; i++) {
new Thread(() -> {
try {
semaphore.acquire(); // 获取信号量
System.out.println(Thread.currentThread().getName() + " is working");
Thread.sleep(1000);
semaphore.release(); // 释放信号量
} catch (InterruptedException e) {
e.printStackTrace();
}
}).start();
}
}
}
阻塞队列 (BlockingQueue)
BlockingQueue是一个线程安全的队列，支持在队列为空时进行阻塞的读取操作，在队列满时进行阻塞的写入操作。常用的实现类有ArrayBlockingQueue、LinkedBlockingQueue、PriorityBlockingQueue和DelayQueue。
常用方法：
- put()：将元素插入队列，如果队列满则等待。
- take()：从队列取出元素，如果队列为空则等待。
- offer()：尝试插入元素，若队列已满则返回false。
- poll()：尝试取出元素，若队列为空则返回null。
使用场景： BlockingQueue适用于生产者-消费者模型，其中生产者和消费者之间通过队列进行数据交换。例如，生产者线程将数据放入队列，消费者线程从队列中取出数据进行处理。
示例代码：
import java.util.concurrent.ArrayBlockingQueue;
public class BlockingQueueExample {
public static void main(String[] args) throws InterruptedException {
ArrayBlockingQueue<Integer> queue = new ArrayBlockingQueue<>(3);
// 生产者线程
new Thread(() -> {
try {
for (int i = 0; i < 5; i++) {
queue.put(i); // 阻塞队列满时会阻塞
System.out.println("Produced: " + i);
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}).start();
// 消费者线程
new Thread(() -> {
try {
for (int i = 0; i < 5; i++) {
Integer item = queue.take(); // 阻塞队列空时会阻塞
System.out.println("Consumed: " + item);
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}).start();
}
}
ArrayBlockingQueue
ArrayBlockingQueue是一个基于数组的有界阻塞队列。它的容量在创建时固定，适用于需要严格控制队列大小的场景。
LinkedBlockingQueue
LinkedBlockingQueue是一个基于链表的无界阻塞队列，默认容量为Integer.MAX_VALUE，可以通过构造函数指定容量。适用于需要动态调整队列大小的场景。
PriorityBlockingQueue
PriorityBlockingQueue是一个支持优先级的无界阻塞队列，元素按照自然顺序或自定义的Comparator排序。适用于需要按优先级处理任务的场景。
DelayQueue
DelayQueue是一个支持延迟获取的无界阻塞队列，元素只有在经过指定延迟时间后才能被取出。适用于定时任务或缓存淘汰等场景。
并发集合 (ConcurrentHashMap, CopyOnWriteArrayList)
ConcurrentHashMap
ConcurrentHashMap是一个线程安全的哈希表，支持高效的并发访问。它通过分段锁定的方式，使得多个线程可以同时访问不同段的数据，从而提高并发性能。
示例代码：
import java.util.concurrent.ConcurrentHashMap;
public class ConcurrentHashMapExample {
public static void main(String[] args) {
ConcurrentHashMap<String, String> map = new ConcurrentHashMap<>();
map.put("key1", "value1");
map.put("key2", "value2");
map.forEach((key, value) -> System.out.println(key + ": " + value));
}
}
ConcurrentHashMap的分段锁定机制使得在多线程环境下，多个线程可以同时访问不同的段，从而提高并发性能。这种方法在Java 8之后被弃用，取而代之的是使用CAS（Compare and Swap）操作实现的无锁化设计，这进一步提升了性能。
CopyOnWriteArrayList
CopyOnWriteArrayList是一个线程安全的列表实现，采用“写时复制”的策略。当修改操作（如add()、remove()）发生时，CopyOnWriteArrayList会复制底层数组，以确保其他线程可以并发读取而不受影响。它适用于读取多、写入少的场景。
示例代码：
import java.util.concurrent.CopyOnWriteArrayList;
public class CopyOnWriteArrayListExample {
public static void main(String[] args) {
CopyOnWriteArrayList<String> list = new CopyOnWriteArrayList<>();
list.add("apple");
list.add("banana");
list.forEach(System.out::println);
}
}
CopyOnWriteArrayList通过写时复制的方式，在修改时创建新的数组副本，从而保证了读取操作的线程安全。这种方法虽然在写入时有较高的开销，但非常适合读多写少的场景，如缓存或观察者模式。
原子类 (AtomicInteger, AtomicReference)
原子类是通过底层的硬件原子操作，来确保多线程环境下对变量的操作是线程安全的。AtomicInteger和AtomicReference是常用的原子类，分别用于整数类型和对象引用类型的线程安全操作。
AtomicInteger
AtomicInteger提供了原子性操作，例如递增、递减、加减等。这些操作在多线程环境下可以保证线程安全，无需使用额外的锁。
示例代码：
import java.util.concurrent.atomic.AtomicInteger;
public class AtomicIntegerExample {
public static void main(String[] args) {
AtomicInteger atomicInt = new AtomicInteger(0);
System.out.println(atomicInt.incrementAndGet()); // 原子递增
}
}
AtomicInteger的原子操作通过CAS（Compare and Swap）实现，这使得它在多线程环境下具有较高的性能。CAS操作在硬件层面支持，无需使用锁，因此在高并发场景下表现优异。
AtomicReference
AtomicReference用于原子性地更新对象引用，确保在并发环境下对对象的操作是线程安全的。它支持对引用类型的原子操作，如设置、比较和交换等。
示例代码：
import java.util.concurrent.atomic.AtomicReference;
public class AtomicReferenceExample {
public static void main(String[] args) {
AtomicReference<String> atomicRef = new AtomicReference<>("Hello");
atomicRef.set("World");
System.out.println(atomicRef.get());
}
}
AtomicReference的原子操作同样基于CAS机制，适用于需要对对象引用进行原子更新的场景。例如，在并发环境中更新某个对象的状态，或者实现线程安全的引用类型变量。
Future 和 CompletableFuture
Future
Future是一个用于表示异步计算结果的接口，可以通过它来获取计算结果、取消任务或检查任务的完成状态。Future在Java 5中被引入，是处理异步任务的基础接口。
示例代码：
import java.util.concurrent.*;
public class FutureExample {
public static void main(String[] args) throws ExecutionException, InterruptedException {
ExecutorService executor = Executors.newSingleThreadExecutor();
Future<Integer> future = executor.submit(() -> 42);
// 获取计算结果
Integer result = future.get();
System.out.println("Result: " + result);
executor.shutdown();
}
}
Future的主要方法包括get()（获取结果）、cancel()（取消任务）和isDone()（检查任务是否完成）。它适用于简单的异步任务处理，但在处理复杂的异步流时，其功能较为有限。
CompletableFuture
CompletableFuture是Future的扩展，它提供了更多的异步操作，如thenApply()、thenAccept()等，支持更复杂的异步流控制。CompletableFuture在Java 8中被引入，是处理异步任务的高级工具。
示例代码：
import java.util.concurrent.CompletableFuture;
public class CompletableFutureExample {
public static void main(String[] args) {
CompletableFuture.supplyAsync(() -> "Hello")
.thenApply(result -> result + " World")
.thenAccept(System.out::println);
}
}
CompletableFuture支持链式调用，可以将多个异步操作组合在一起。例如，可以将一个异步任务的结果作为下一个任务的输入，实现复杂的异步流程。通过使用CompletableFuture，开发者可以更灵活地处理异步任务，提高程序的并发性能和可维护性。
总结
Java提供了丰富的并发工具类，如CountDownLatch、CyclicBarrier、Semaphore、BlockingQueue、ConcurrentHashMap、CopyOnWriteArrayList、AtomicInteger、AtomicReference以及Future和CompletableFuture。这些工具类帮助开发者简化多线程编程的复杂性，提高程序的并发性能和可扩展性。
通过合理使用这些工具类，开发者可以更高效地控制线程执行顺序、提高性能，并处理复杂的并发场景。掌握这些并发工具类，对于构建高效、可扩展的多线程应用程序至关重要。在实际开发中，根据具体需求选择合适的工具类，可以显著提升代码质量和系统性能。
关键字
Java, 并发工具类, CountDownLatch, CyclicBarrier, Semaphore, BlockingQueue, ConcurrentHashMap, CopyOnWriteArrayList, AtomicInteger, AtomicReference, Future, CompletableFuture