Java并发集合案例

wen java案例 2026-08-09 164

本文目录导读：

Java并发集合案例

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并发集合的主要使用场景和最佳实践，实际开发中，应根据具体需求选择合适的并发集合：考虑操作的读写比例、是否需要阻塞、是否需要排序、以及性能要求等因素。

本文地址： https://bmjup.com/post/42225.html