本文目录导读：
- ConcurrentHashMap 并发映射
- CopyOnWriteArrayList 并发列表
- BlockingQueue 阻塞队列
- ConcurrentLinkedQueue 并发链表队列
- 完整业务场景案例：分布式任务调度系统
- 并发集合性能对比
- 选择指南
我来为你提供Java并发集合的完整案例,涵盖常用的并发容器和实际应用场景。
ConcurrentHashMap 并发映射
基本使用案例
import java.util.concurrent.*;
import java.util.concurrent.atomic.LongAdder;
public class ConcurrentHashMapDemo {
// 案例1：统计单词频率
public static void wordCountDemo() {
ConcurrentHashMap<String, Integer> wordCount = new ConcurrentHashMap<>();
// 模拟多线程写入
ExecutorService executor = Executors.newFixedThreadPool(4);
String[] words = {"Java", "Python", "Java", "Go", "Java", "Python"};
for (String word : words) {
executor.submit(() -> {
// 使用compute方法原子更新
wordCount.compute(word, (key, val) -> val == null ? 1 : val + 1);
});
}
executor.shutdown();
try {
executor.awaitTermination(5, TimeUnit.SECONDS);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("Word count: " + wordCount);
}
// 案例2：缓存示例
public static void cacheDemo() throws InterruptedException {
ConcurrentHashMap<String, String> cache = new ConcurrentHashMap<>();
// 使用putIfAbsent实现缓存
String value = cache.putIfAbsent("key1", "value1");
System.out.println("First put: " + value); // null
value = cache.putIfAbsent("key1", "newValue");
System.out.println("Second put: " + value); // value1
System.out.println("Cache value: " + cache.get("key1"));
}
public static void main(String[] args) throws InterruptedException {
wordCountDemo();
cacheDemo();
}
}
CopyOnWriteArrayList 并发列表
读写分离案例
import java.util.concurrent.*;
import java.util.concurrent.CopyOnWriteArrayList;
public class CopyOnWriteArrayListDemo {
// 读多写少场景
public static void readWriteDemo() {
CopyOnWriteArrayList<String> list = new CopyOnWriteArrayList<>();
list.add("A");
list.add("B");
list.add("C");
// 多个读线程
Runnable reader = () -> {
for (int i = 0; i < 5; i++) {
System.out.println(Thread.currentThread().getName()
+ " reading: " + list);
try {
Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
};
// 写线程
Runnable writer = () -> {
for (int i = 0; i < 3; i++) {
list.add("Element-" + i);
System.out.println(Thread.currentThread().getName()
+ " added element");
try {
Thread.sleep(200);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
};
ExecutorService pool = Executors.newFixedThreadPool(4);
// 3个读线程
for (int i = 0; i < 3; i++) {
pool.submit(reader);
}
// 1个写线程
pool.submit(writer);
pool.shutdown();
}
// 迭代器弱一致性
public static void iteratorDemo() {
CopyOnWriteArrayList<Integer> numbers =
new CopyOnWriteArrayList<>();
for (int i = 0; i < 10; i++) {
numbers.add(i);
}
// 迭代时修改
Thread iteratorThread = new Thread(() -> {
for (Integer num : numbers) {
System.out.println("Iterating: " + num);
try {
Thread.sleep(50);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
Thread modifierThread = new Thread(() -> {
for (int i = 10; i < 15; i++) {
numbers.add(i);
System.out.println("Added: " + i);
try {
Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
iteratorThread.start();
modifierThread.start();
try {
iteratorThread.join();
modifierThread.join();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
public static void main(String[] args) {
readWriteDemo();
// iteratorDemo();
}
}
BlockingQueue 阻塞队列
生产者-消费者模式
import java.util.concurrent.*;
public class BlockingQueueDemo {
// 使用ArrayBlockingQueue实现生产者消费者
public static void producerConsumerDemo() {
BlockingQueue<String> queue = new ArrayBlockingQueue<>(5);
// 生产者
Thread producer = new Thread(() -> {
try {
int item = 0;
while (item < 10) {
String product = "Product-" + item;
queue.put(product); // 队列满时阻塞
System.out.println("Produced: " + product);
item++;
Thread.sleep(200);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}, "Producer");
// 消费者
Thread consumer = new Thread(() -> {
try {
while (true) {
String product = queue.take(); // 队列空时阻塞
System.out.println("Consumed: " + product);
Thread.sleep(300);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}, "Consumer");
producer.start();
consumer.start();
try {
producer.join();
// 给消费者一些时间处理剩余任务
Thread.sleep(2000);
consumer.interrupt();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
// 延迟队列
public static void delayQueueDemo() throws InterruptedException {
DelayQueue<DelayedTask> queue = new DelayQueue<>();
// 添加延迟任务
queue.put(new DelayedTask("Task1", 2000));
queue.put(new DelayedTask("Task2", 1000));
queue.put(new DelayedTask("Task3", 3000));
System.out.println("Waiting for delayed tasks...");
while (!queue.isEmpty()) {
DelayedTask task = queue.take(); // 阻塞直到延迟到期
System.out.println("Executed: " + task.taskName
+ " at " + System.currentTimeMillis());
}
}
// 延迟任务类
static class DelayedTask implements Delayed {
private String taskName;
private long delayTime;
public DelayedTask(String taskName, long delayMillis) {
this.taskName = taskName;
this.delayTime = System.currentTimeMillis() + delayMillis;
}
@Override
public long getDelay(TimeUnit unit) {
return unit.convert(delayTime - System.currentTimeMillis(),
TimeUnit.MILLISECONDS);
}
@Override
public int compareTo(Delayed o) {
return Long.compare(this.delayTime,
((DelayedTask)o).delayTime);
}
}
public static void main(String[] args) throws InterruptedException {
producerConsumerDemo();
System.out.println("\n--- Delay Queue Demo ---");
delayQueueDemo();
}
}
ConcurrentLinkedQueue 并发链表队列
无界队列使用
import java.util.concurrent.*;
public class ConcurrentLinkedQueueDemo {
public static void main(String[] args) throws InterruptedException {
ConcurrentLinkedQueue<Integer> queue = new ConcurrentLinkedQueue<>();
// 多线程写入
Runnable producer = () -> {
for (int i = 0; i < 100; i++) {
queue.offer(Thread.currentThread().getName().hashCode() + i);
}
};
// 多线程读取
Runnable consumer = () -> {
long count = 0;
while (count < 100) {
Integer item = queue.poll();
if (item != null) {
count++;
// 处理数据
System.out.println("Processed: " + item);
}
}
};
ExecutorService pool = Executors.newFixedThreadPool(4);
// 2个生产者
pool.submit(producer);
pool.submit(producer);
// 等待生产者完成
pool.awaitTermination(2, TimeUnit.SECONDS); // 简化处理
// 2个消费者
pool.submit(consumer);
pool.submit(consumer);
pool.shutdown();
pool.awaitTermination(5, TimeUnit.SECONDS);
System.out.println("Queue size: " + queue.size());
}
}
完整业务场景案例：分布式任务调度系统
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;
public class ConcurrentCollectionCompleteDemo {
// 任务类
static class Task {
private String taskId;
private int priority;
public Task(String taskId, int priority) {
this.taskId = taskId;
this.priority = priority;
}
@Override
public String toString() {
return "Task{" + taskId + ", priority=" + priority + "}";
}
}
// 任务调度器
static class TaskScheduler {
// 高优先级任务队列
private final PriorityBlockingQueue<Task> highPriorityQueue =
new PriorityBlockingQueue<>(10, (t1, t2) ->
Integer.compare(t2.priority, t1.priority));
// 普通任务队列
private final BlockingQueue<Task> normalQueue =
new LinkedBlockingQueue<>();
// 已完成任务集合
private final ConcurrentSkipListSet<String> completedTasks =
new ConcurrentSkipListSet<>();
// 任务计数器
private final AtomicLong taskCounter = new AtomicLong();
// 任务处理结果缓存
private final ConcurrentHashMap<String, String> taskResults =
new ConcurrentHashMap<>();
// 提交高优先级任务
public void submitHighPriorityTask(Task task) {
highPriorityQueue.offer(task);
System.out.println(Thread.currentThread().getName()
+ " submitted high priority: " + task);
}
// 提交普通任务
public void submitNormalTask(Task task) {
normalQueue.offer(task);
System.out.println(Thread.currentThread().getName()
+ " submitted normal: " + task);
}
// 处理任务
public void processTasks() throws InterruptedException {
// 优先处理高优先级任务
Task highTask = null;
Task normalTask = null;
// 尝试获取高优先级任务
highTask = highPriorityQueue.poll(100, TimeUnit.MILLISECONDS);
if (highTask == null) {
// 尝试获取普通任务
normalTask = normalQueue.poll(100, TimeUnit.MILLISECONDS);
}
if (highTask != null) {
handleTask(highTask);
} else if (normalTask != null) {
handleTask(normalTask);
} else {
// 没有任务时休眠
Thread.sleep(100);
}
}
private void handleTask(Task task) {
// 检查是否已完成
if (completedTasks.contains(task.taskId)) {
return;
}
// 模拟处理
try {
Thread.sleep(100);
String result = "Processed " + task + " at "
+ System.currentTimeMillis();
taskResults.put(task.taskId, result);
completedTasks.add(task.taskId);
taskCounter.incrementAndGet();
System.out.println(Thread.currentThread().getName()
+ " completed: " + task);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
public void printStatistics() {
System.out.println("\n=== Scheduler Statistics ===");
System.out.println("Completed tasks: " + taskCounter.get());
System.out.println("Completed set size: " + completedTasks.size());
System.out.println("Results cache size: " + taskResults.size());
System.out.println("Pending high priority: " + highPriorityQueue.size());
System.out.println("Pending normal: " + normalQueue.size());
}
}
public static void main(String[] args) throws InterruptedException {
TaskScheduler scheduler = new TaskScheduler();
// 创建线程池
ExecutorService executor = Executors.newFixedThreadPool(3);
// 提交任务
executor.submit(() -> {
for (int i = 0; i < 5; i++) {
scheduler.submitHighPriorityTask(
new Task("High-" + i, 100 - i));
try {
Thread.sleep(200);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
});
executor.submit(() -> {
for (int i = 0; i < 5; i++) {
scheduler.submitNormalTask(
new Task("Normal-" + i, 0));
try {
Thread.sleep(300);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
});
// 处理线程
executor.submit(() -> {
try {
while (true) {
scheduler.processTasks();
Thread.sleep(100);
// 简单停止条件
if (Thread.currentThread().isInterrupted()) {
break;
}
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
// 运行一段时间后停止
Thread.sleep(10000);
executor.shutdownNow();
// 等待线程池关闭
executor.awaitTermination(5, TimeUnit.SECONDS);
// 打印统计信息
scheduler.printStatistics();
}
}
并发集合性能对比
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;
public class PerformanceComparison {
public static void main(String[] args) throws InterruptedException {
// HashMap vs ConcurrentHashMap
compareMaps();
// ArrayList vs CopyOnWriteArrayList
compareLists();
// synchronized vs ConcurrentLinkedQueue
compareQueues();
}
private static void compareMaps() throws InterruptedException {
int iterations = 10000;
// HashMap（非线程安全）
Map<String, Integer> hashMap = new HashMap<>();
long start = System.nanoTime();
// 使用时需要同步
Object lock = new Object();
ExecutorService pool = Executors.newFixedThreadPool(4);
// 使用同步的HashMap
Runnable syncMapTask = () -> {
Random random = new Random();
for (int i = 0; i < iterations; i++) {
synchronized (lock) {
hashMap.put("key-" + random.nextInt(1000), i);
}
}
};
for (int i = 0; i < 4; i++) {
pool.submit(syncMapTask);
}
pool.shutdown();
pool.awaitTermination(10, TimeUnit.SECONDS);
long hashMapTime = System.nanoTime() - start;
// ConcurrentHashMap
Map<String, Integer> concurrentMap = new ConcurrentHashMap<>();
start = System.nanoTime();
pool = Executors.newFixedThreadPool(4);
Runnable concurrentMapTask = () -> {
Random random = new Random();
for (int i = 0; i < iterations; i++) {
concurrentMap.put("key-" + random.nextInt(1000), i);
}
};
for (int i = 0; i < 4; i++) {
pool.submit(concurrentMapTask);
}
pool.shutdown();
pool.awaitTermination(10, TimeUnit.SECONDS);
long concurrentMapTime = System.nanoTime() - start;
System.out.println("HashMap (synchronized): "
+ hashMapTime / 1_000_000 + "ms");
System.out.println("ConcurrentHashMap: "
+ concurrentMapTime / 1_000_000 + "ms");
}
private static void compareLists() {
int elements = 10000;
// CopyOnWriteArrayList 写入
CopyOnWriteArrayList<Integer> cowList = new CopyOnWriteArrayList<>();
long start = System.nanoTime();
for (int i = 0; i < elements; i++) {
cowList.add(i);
}
long cowWriteTime = System.nanoTime() - start;
// 读取
start = System.nanoTime();
int sum = 0;
for (Integer value : cowList) {
sum += value;
}
long cowReadTime = System.nanoTime() - start;
System.out.println("\nCopyOnWriteArrayList:");
System.out.println(" Write time: " + cowWriteTime / 1_000_000 + "ms");
System.out.println(" Read time: " + cowReadTime / 1_000_000 + "ms");
}
private static void compareQueues() throws InterruptedException {
int tasks = 10000;
// ArrayBlockingQueue
BlockingQueue<Integer> arrayQueue = new ArrayBlockingQueue<>(100000);
long start = System.nanoTime();
for (int i = 0; i < tasks; i++) {
arrayQueue.offer(i);
}
for (int i = 0; i < tasks; i++) {
arrayQueue.poll();
}
long arrayQueueTime = System.nanoTime() - start;
// ConcurrentLinkedQueue
ConcurrentLinkedQueue<Integer> concurrentQueue =
new ConcurrentLinkedQueue<>();
start = System.nanoTime();
for (int i = 0; i < tasks; i++) {
concurrentQueue.offer(i);
}
for (int i = 0; i < tasks; i++) {
concurrentQueue.poll();
}
long concurrentQueueTime = System.nanoTime() - start;
System.out.println("\nQueue Performance:");
System.out.println("ArrayBlockingQueue: "
+ arrayQueueTime / 1_000_000 + "ms");
System.out.println("ConcurrentLinkedQueue: "
+ concurrentQueueTime / 1_000_000 + "ms");
}
}
选择指南
public class CollectionSelectionGuide {
public static void main(String[] args) {
System.out.println("=== 并发集合选择指南 ===");
System.out.println("""
1. ConcurrentHashMap
- 读多写少场景
- 高并发map操作
- 需要原子性操作(compute, putIfAbsent)
2. CopyOnWriteArrayList
- 读多写极少场景
- 需要弱一致性迭代
- 集合经常被遍历
3. BlockingQueue 系列
- 生产者-消费者模式
- 需要阻塞特性
- 有界队列(ArrayBlockingQueue)
- 无界队列(LinkedBlockingQueue)
4. ConcurrentLinkedQueue
- 高并发无界队列
- 不需要阻塞
- CAS操作实现
5. PriorityBlockingQueue
- 需要优先级排序
- 阻塞队列+优先级
6. DelayQueue
- 延迟任务处理
- 定时任务调度
7. ConcurrentSkipListMap/Set
- 需要排序
- 并发有序结构
- 不可重复(Set)或可重复(Map)
""");
}
}
这些案例涵盖了Java并发集合的主要使用场景和最佳实践，实际开发中，应根据具体需求选择合适的并发集合：考虑操作的读写比例、是否需要阻塞、是否需要排序、以及性能要求等因素。
