Java并发集合详解：线程安全的集合类

概述

Java提供了多种线程安全的集合类，用于在并发环境下安全地操作数据。本教程介绍常用的并发集合类及其使用场景。

1. ConcurrentHashMap

import java.util.concurrent.*;
import java.util.*;

public class ConcurrentHashMapExample {
    public static void main(String[] args) throws InterruptedException {
        ConcurrentHashMap<String, Integer> map = new ConcurrentHashMap<>();
        
        // 基本操作
        map.put("Alice", 25);
        map.put("Bob", 30);
        map.put("Charlie", 35);
        
        System.out.println("Map: " + map);
        
        // 原子操作
        map.putIfAbsent("David", 28);
        System.out.println("putIfAbsent: " + map);
        
        map.computeIfAbsent("Eve", k -> 22);
        System.out.println("computeIfAbsent: " + map);
        
        map.merge("Alice", 5, Integer::sum);
        System.out.println("merge: " + map);
        
        // 并发测试
        ConcurrentHashMap<String, Integer> concurrentMap = new ConcurrentHashMap<>();
        int threadCount = 10;
        int incrementsPerThread = 1000;
        
        Thread[] threads = new Thread[threadCount];
        for (int i = 0; i < threadCount; i++) {
            threads[i] = new Thread(() -> {
                for (int j = 0; j < incrementsPerThread; j++) {
                    concurrentMap.merge("counter", 1, Integer::sum);
                }
            });
            threads[i].start();
        }
        
        for (Thread thread : threads) {
            thread.join();
        }
        
        System.out.println("并发计数: " + concurrentMap.get("counter"));
        System.out.println("期望值: " + (threadCount * incrementsPerThread));
    }
}

2. CopyOnWriteArrayList

import java.util.concurrent.*;
import java.util.*;

public class CopyOnWriteArrayListExample {
    public static void main(String[] args) {
        // CopyOnWriteArrayList：写时复制，读无锁
        CopyOnWriteArrayList<String> list = new CopyOnWriteArrayList<>();
        
        // 添加元素
        list.add("Alice");
        list.add("Bob");
        list.add("Charlie");
        
        System.out.println("List: " + list);
        
        // 读取元素（无锁，高效）
        String first = list.get(0);
        System.out.println("第一个元素: " + first);
        
        // 遍历（快照，不会抛出ConcurrentModificationException）
        for (String name : list) {
            System.out.println("遍历: " + name);
        }
        
        // 写入操作（复制底层数组，开销较大）
        list.add("David");
        list.remove("Bob");
        
        System.out.println("修改后: " + list);
        
        // 适用场景：读多写少
        // 优点：读操作无锁，性能高
        // 缺点：写操作需要复制数组，内存开销大
    }
}

3. CopyOnWriteArraySet

import java.util.concurrent.*;
import java.util.*;

public class CopyOnWriteArraySetExample {
    public static void main(String[] args) {
        CopyOnWriteArraySet<String> set = new CopyOnWriteArraySet<>();
        
        set.add("Alice");
        set.add("Bob");
        set.add("Charlie");
        set.add("Alice");  // 重复元素不会被添加
        
        System.out.println("Set: " + set);
        System.out.println("大小: " + set.size());
        
        // 遍历
        for (String name : set) {
            System.out.println("遍历: " + name);
        }
    }
}

4. ConcurrentLinkedQueue

import java.util.concurrent.*;

public class ConcurrentLinkedQueueExample {
    public static void main(String[] args) {
        ConcurrentLinkedQueue<String> queue = new ConcurrentLinkedQueue<>();
        
        // 添加元素
        queue.offer("Alice");
        queue.offer("Bob");
        queue.offer("Charlie");
        
        System.out.println("Queue: " + queue);
        
        // 获取并移除元素
        String first = queue.poll();
        System.out.println("取出: " + first);
        
        // 查看但不移除
        String peek = queue.peek();
        System.out.println("查看: " + peek);
        
        // 检查是否为空
        System.out.println("是否为空: " + queue.isEmpty());
        System.out.println("大小: " + queue.size());
    }
}

5. BlockingQueue

import java.util.concurrent.*;

public class BlockingQueueExample {
    public static void main(String[] args) {
        // ArrayBlockingQueue：有界阻塞队列
        BlockingQueue<String> arrayQueue = new ArrayBlockingQueue<>(10);
        
        // LinkedBlockingQueue：可选有界阻塞队列
        BlockingQueue<String> linkedQueue = new LinkedBlockingQueue<>();
        
        // PriorityBlockingQueue：优先级阻塞队列
        BlockingQueue<Integer> priorityQueue = new PriorityBlockingQueue<>();
        
        // SynchronousQueue：同步队列
        BlockingQueue<String> syncQueue = new SynchronousQueue<>();
        
        // 阻塞操作
        try {
            arrayQueue.put("Alice");  // 队列满时阻塞
            String item = arrayQueue.take();  // 队列空时阻塞
            
            // 超时操作
            boolean added = arrayQueue.offer("Bob", 1, TimeUnit.SECONDS);
            String polled = arrayQueue.poll(1, TimeUnit.SECONDS);
            
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

6. 实际应用示例

生产者-消费者模式

import java.util.concurrent.*;

public class ProducerConsumer {
    private final BlockingQueue<String> queue;
    
    public ProducerConsumer(int capacity) {
        this.queue = new ArrayBlockingQueue<>(capacity);
    }
    
    public void produce(String item) throws InterruptedException {
        queue.put(item);
        System.out.println("生产: " + item);
    }
    
    public String consume() throws InterruptedException {
        String item = queue.take();
        System.out.println("消费: " + item);
        return item;
    }
    
    public static void main(String[] args) {
        ProducerConsumer pc = new ProducerConsumer(5);
        
        // 生产者线程
        Thread producer = new Thread(() -> {
            for (int i = 0; i < 10; i++) {
                try {
                    pc.produce("Item-" + i);
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
        
        // 消费者线程
        Thread consumer = new Thread(() -> {
            for (int i = 0; i < 10; i++) {
                try {
                    pc.consume();
                    Thread.sleep(150);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
        
        producer.start();
        consumer.start();
    }
}

7. 最佳实践

1. 选择合适的并发集合：
   • ConcurrentHashMap：高并发读写
   • CopyOnWriteArrayList：读多写少
   • BlockingQueue：生产者-消费者模式
2. 使用原子操作：putIfAbsent、merge、computeIfAbsent等
3. 避免长时间锁定：使用无锁数据结构
4. 合理设置容量：有界队列设置合适的容量
5. 监控集合状态：定期检查集合大小和性能

总结

Java并发集合提供了线程安全的数据结构，适用于不同的并发场景。掌握这些集合的使用，可以编写高效的并发程序。在实际编程中，要根据需求选择合适的并发集合。