MK
摩柯社区 - 一个极简的技术知识社区
AI 面试

Java Collections 的线程安全策略

2022-10-037.5k 阅读

Java Collections 线程安全问题概述

在多线程编程环境中，Java Collections 框架中的集合类如果使用不当，很容易引发线程安全问题。例如，当多个线程同时对一个非线程安全的集合进行读写操作时，可能会导致数据不一致、并发修改异常等情况。以 ArrayList 为例，它是一个常用的动态数组实现类，但并非线程安全。

import java.util.ArrayList;
import java.util.List;

public class ArrayListUnsafeExample {
    private static List<Integer> list = new ArrayList<>();

    public static void main(String[] args) {
        Thread thread1 = new Thread(() -> {
            for (int i = 0; i < 1000; i++) {
                list.add(i);
            }
        });

        Thread thread2 = new Thread(() -> {
            for (int i = 1000; i < 2000; i++) {
                list.add(i);
            }
        });

        thread1.start();
        thread2.start();

        try {
            thread1.join();
            thread2.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        System.out.println("List size: " + list.size());
    }
}

在上述代码中，两个线程同时向 ArrayList 中添加元素。由于 ArrayList 不是线程安全的，在并发环境下运行时，可能会出现 IndexOutOfBoundsException 或者最终集合的大小并非预期的 2000。这是因为多个线程同时修改集合的内部结构，导致数据不一致。

同步包装器（Synchronized Wrapper）

为了使非线程安全的集合类在多线程环境中安全使用，Java 提供了同步包装器。通过 Collections.synchronizedXxx 方法，可以将非线程安全的集合转换为线程安全的集合。例如，对于 List，可以使用 Collections.synchronizedList 方法。

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

public class SynchronizedListExample {
    private static List<Integer> list = Collections.synchronizedList(new ArrayList<>());

    public static void main(String[] args) {
        Thread thread1 = new Thread(() -> {
            for (int i = 0; i < 1000; i++) {
                synchronized (list) {
                    list.add(i);
                }
            }
        });

        Thread thread2 = new Thread(() -> {
            for (int i = 1000; i < 2000; i++) {
                synchronized (list) {
                    list.add(i);
                }
            }
        });

        thread1.start();
        thread2.start();

        try {
            thread1.join();
            thread2.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        synchronized (list) {
            System.out.println("List size: " + list.size());
        }
    }
}

在这段代码中，首先通过 Collections.synchronizedList 将 ArrayList 包装成线程安全的 List。在对集合进行操作时，需要手动同步访问，这里通过 synchronized 块来实现。这样可以确保在同一时间只有一个线程能够访问集合，从而避免线程安全问题。

同步包装器虽然能解决基本的线程安全问题，但它存在一些局限性。例如，迭代器遍历集合时，如果在遍历过程中有其他线程修改了集合，仍然会抛出 ConcurrentModificationException。这是因为迭代器本身并没有进行同步保护。

import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;

public class SynchronizedListIteratorProblem {
    private static List<Integer> list = Collections.synchronizedList(new ArrayList<>());

    public static void main(String[] args) {
        for (int i = 0; i < 10; i++) {
            list.add(i);
        }

        Thread thread1 = new Thread(() -> {
            synchronized (list) {
                Iterator<Integer> iterator = list.iterator();
                while (iterator.hasNext()) {
                    System.out.println(iterator.next());
                    try {
                        Thread.sleep(100);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        });

        Thread thread2 = new Thread(() -> {
            synchronized (list) {
                try {
                    Thread.sleep(200);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                list.add(10);
            }
        });

        thread1.start();
        thread2.start();

        try {
            thread1.join();
            thread2.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

在上述代码中，thread1 正在遍历集合，而 thread2 在 thread1 遍历过程中修改了集合，这就会导致 ConcurrentModificationException 异常。要解决这个问题，在遍历集合时需要对整个集合进行同步锁定，确保在遍历过程中没有其他线程修改集合。

CopyOnWrite 集合

CopyOnWrite 集合是 Java 提供的另一种线程安全的集合实现方式。它的核心思想是，当对集合进行修改（如添加、删除元素）时，会创建一个新的底层数组，将修改操作应用到新数组上，而读取操作则始终在旧数组上进行。这样，读取操作不会被修改操作所干扰，从而实现线程安全。

以 CopyOnWriteArrayList 为例：

import java.util.Iterator;
import java.util.concurrent.CopyOnWriteArrayList;

public class CopyOnWriteArrayListExample {
    private static CopyOnWriteArrayList<Integer> list = new CopyOnWriteArrayList<>();

    public static void main(String[] args) {
        Thread thread1 = new Thread(() -> {
            for (int i = 0; i < 1000; i++) {
                list.add(i);
            }
        });

        Thread thread2 = new Thread(() -> {
            Iterator<Integer> iterator = list.iterator();
            while (iterator.hasNext()) {
                System.out.println(iterator.next());
            }
        });

        thread1.start();
        thread2.start();

        try {
            thread1.join();
            thread2.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

在这个例子中，thread1 向 CopyOnWriteArrayList 中添加元素，thread2 同时遍历该集合。由于 CopyOnWriteArrayList 的特性，读取操作不受写入操作的影响，不会抛出 ConcurrentModificationException。

CopyOnWrite 集合适用于读多写少的场景。因为每次修改操作都要创建新的数组，开销较大，所以写操作性能较低。而读操作由于不需要加锁，性能相对较高。例如，在一个日志记录系统中，可能会有多个线程读取日志信息（读操作频繁），而只有少数线程会添加新的日志记录（写操作较少），这种情况下 CopyOnWriteArrayList 是一个不错的选择。

并发集合框架（Concurrent Collections Framework）

Java 的并发集合框架提供了一系列高性能、线程安全的集合类。这些集合类针对多线程环境进行了优化，采用了更细粒度的锁机制或者无锁算法，以提高并发性能。

ConcurrentHashMap

ConcurrentHashMap 是一个线程安全的哈希表实现。它采用了分段锁（Segment）机制，在 JDK 1.8 之后，改为使用 CAS（Compare - And - Swap）操作和 synchronized 关键字来实现线程安全。

import java.util.concurrent.ConcurrentHashMap;

public class ConcurrentHashMapExample {
    private static ConcurrentHashMap<String, Integer> map = new ConcurrentHashMap<>();

    public static void main(String[] args) {
        Thread thread1 = new Thread(() -> {
            for (int i = 0; i < 1000; i++) {
                map.put("key" + i, i);
            }
        });

        Thread thread2 = new Thread(() -> {
            for (int i = 1000; i < 2000; i++) {
                map.put("key" + i, i);
            }
        });

        thread1.start();
        thread2.start();

        try {
            thread1.join();
            thread2.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        System.out.println("Map size: " + map.size());
    }
}

在上述代码中，两个线程同时向 ConcurrentHashMap 中插入键值对。ConcurrentHashMap 允许并发的读写操作，不会像 HashMap 在多线程环境下那样出现死循环或者数据丢失等问题。在 JDK 1.8 之前，ConcurrentHashMap 通过分段锁来实现并发控制，不同的线程可以同时访问不同的段，从而提高并发性能。在 JDK 1.8 之后，虽然不再使用分段锁，但通过 CAS 操作和 synchronized 关键字，同样实现了高效的并发访问。

ConcurrentLinkedQueue

ConcurrentLinkedQueue 是一个基于链表的无界线程安全队列。它采用无锁算法实现，通过 CAS 操作来保证线程安全。

import java.util.concurrent.ConcurrentLinkedQueue;

public class ConcurrentLinkedQueueExample {
    private static ConcurrentLinkedQueue<Integer> queue = new ConcurrentLinkedQueue<>();

    public static void main(String[] args) {
        Thread thread1 = new Thread(() -> {
            for (int i = 0; i < 1000; i++) {
                queue.add(i);
            }
        });

        Thread thread2 = new Thread(() -> {
            for (int i = 1000; i < 2000; i++) {
                queue.add(i);
            }
        });

        thread1.start();
        thread2.start();

        try {
            thread1.join();
            thread2.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        System.out.println("Queue size: " + queue.size());
    }
}

在这个例子中，两个线程同时向 ConcurrentLinkedQueue 中添加元素。由于它采用无锁算法，在高并发场景下，性能优于传统的同步队列。ConcurrentLinkedQueue 适用于需要高效并发插入和删除元素的场景，例如生产者 - 消费者模型中的消息队列。

BlockingQueue 及其实现类

BlockingQueue 是一个接口，它提供了在多线程环境下阻塞的队列操作。当队列满时，插入操作会被阻塞；当队列空时，取出操作会被阻塞。常见的实现类有 ArrayBlockingQueue、LinkedBlockingQueue 等。

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

public class ArrayBlockingQueueExample {
    private static BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(10);

    public static void main(String[] args) {
        Thread producer = new Thread(() -> {
            for (int i = 0; i < 20; i++) {
                try {
                    queue.put(i);
                    System.out.println("Produced: " + i);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });

        Thread consumer = new Thread(() -> {
            while (true) {
                try {
                    Integer item = queue.take();
                    System.out.println("Consumed: " + item);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                    break;
                }
            }
        });

        producer.start();
        consumer.start();

        try {
            producer.join();
            Thread.sleep(1000);
            consumer.interrupt();
            consumer.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

在上述代码中，producer 线程向 ArrayBlockingQueue 中插入元素，consumer 线程从队列中取出元素。当队列满时，producer 线程会被阻塞；当队列空时，consumer 线程会被阻塞。这种阻塞机制在多线程协作场景中非常有用，例如实现生产者 - 消费者模型时，可以有效地控制数据的生产和消费速度，避免数据丢失或溢出。

选择合适的线程安全策略

在实际应用中，选择合适的线程安全策略至关重要。以下是一些选择的依据：

  1. 读写比例：如果读操作远远多于写操作，CopyOnWrite 集合或者 ConcurrentHashMap 等并发集合框架中的类可能是较好的选择。例如，在缓存系统中，数据的读取频率通常高于写入频率，ConcurrentHashMap 可以提供高效的并发读取性能。而如果写操作频繁，同步包装器或者 BlockingQueue 等更适合，因为 CopyOnWrite 集合写操作开销较大。
  2. 性能要求：对于高性能要求的场景，并发集合框架中的类由于采用了更细粒度的锁机制或者无锁算法，通常能提供更好的并发性能。例如，在高并发的网络服务器中，处理大量请求时，ConcurrentLinkedQueue 可以高效地处理请求队列。而同步包装器虽然简单易用，但由于采用粗粒度的锁，在高并发下性能可能较差。
  3. 功能需求：如果需要阻塞操作，如在生产者 - 消费者模型中，BlockingQueue 及其实现类是必不可少的。如果需要遍历集合且在遍历过程中不希望受到其他线程修改的影响，CopyOnWrite 集合是一个不错的选择。

自定义线程安全集合

有时候，现有的线程安全集合不能满足特定的需求，这就需要自定义线程安全集合。自定义线程安全集合可以通过多种方式实现，例如使用锁机制或者无锁算法。

以下是一个简单的自定义线程安全的固定大小队列的示例，使用 ReentrantLock 来实现线程安全：

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

public class CustomThreadSafeQueue<T> {
    private final T[] queue;
    private int head = 0;
    private int tail = 0;
    private int size = 0;
    private final ReentrantLock lock = new ReentrantLock();
    private final Condition notFull = lock.newCondition();
    private final Condition notEmpty = lock.newCondition();

    public CustomThreadSafeQueue(int capacity) {
        this.queue = (T[]) new Object[capacity];
    }

    public void put(T item) throws InterruptedException {
        lock.lock();
        try {
            while (size == queue.length) {
                notFull.await();
            }
            queue[tail] = item;
            tail = (tail + 1) % queue.length;
            size++;
            notEmpty.signal();
        } finally {
            lock.unlock();
        }
    }

    public T take() throws InterruptedException {
        lock.lock();
        try {
            while (size == 0) {
                notEmpty.await();
            }
            T item = queue[head];
            head = (head + 1) % queue.length;
            size--;
            notFull.signal();
            return item;
        } finally {
            lock.unlock();
        }
    }
}

在上述代码中，通过 ReentrantLock 和 Condition 实现了一个线程安全的固定大小队列。put 方法在队列满时会等待，take 方法在队列空时会等待，从而保证了线程安全和正确的队列操作。

自定义线程安全集合需要深入理解多线程编程和并发控制机制，确保在高并发环境下的正确性和性能。同时，要对锁的粒度、竞争情况等进行仔细的考虑和优化，以避免性能瓶颈。

总结

在多线程编程中，选择合适的 Java Collections 线程安全策略对于程序的正确性和性能至关重要。同步包装器提供了简单的线程安全解决方案，但在高并发场景下存在局限性。CopyOnWrite 集合适用于读多写少的场景，能提供高效的读操作性能。并发集合框架中的类，如 ConcurrentHashMap、ConcurrentLinkedQueue 和 BlockingQueue 及其实现类，针对多线程环境进行了优化，在不同的场景下能提供高性能的并发操作。在特定需求下，还可以自定义线程安全集合。通过深入理解这些线程安全策略，并根据实际应用场景进行选择和优化，可以编写出高效、健壮的多线程程序。

在实际项目中，要综合考虑读写比例、性能要求和功能需求等因素，选择最合适的线程安全集合。同时，要注意在使用过程中遵循相应的规则，如正确处理迭代器遍历、合理使用锁等，以确保程序的线程安全性和稳定性。