[杂谈] Java 容器的线程安全性杂谈 - 编程开发

2000000

本帖最后由 2000000 于 2023-3-18 16:29 编辑

Java 容器的线程安全性杂谈

为什么写本文?

容器在开发中使用广泛，然而在多线程环境下使用非线程安全的容器可能会引起一系列的线程安全问题，比如竞态条件、死锁、数据不一致等。
这些问题可能会导致性能下降或者数据错误，严重情况下可能会对系统的安全性产生影响。

注重容器的线程安全性是非常必要的。使用线程安全的容器可以保证多线程环境下容器的正确使用，提高应用程序的健壮性可靠性。
同时，线程安全容器的使用也可以提高应用程序的性能，减少线程竞争和锁冲突等问题。

在一些小型或个人项目中，可能并不会过多考虑线程安全的问题，而是更注重代码的简洁和易读性，忽略线程安全性问题，不出现约等于没有。

本文旨在介绍一些 Java 容器在多线程环境下的线程安全性，并比较不同容器的线程安全性和性能表现，简单分析实现原理。

笔者的话:

我水平有限，会尽量以较为简单的语言来进行说明。本文表述较为简单，有些地方可以更深入地探讨和讲解，读者可自行研究。
本文适合初学者进行阅读，对初学者来说可能会有一些帮助。
读者在阅读前应该对 并发编程有一些了解，起码需要有开发经验。

如果您在阅读中发现了任何错误欢迎指正! 大佬轻点喷。
如果本文对您有帮助，请让我知道，我很高兴听到这些信息!

关于 ArrayList

为什么 ArrayList 线程不安全?

如果多个线程同时读取、写入同一个 ArrayList，就可能会发生竞态条件 (Race Condition)。
例如，如果一个线程正在读取 ArrayList 中的某个元素，而另一个线程正在删除这个元素，那么读操作就可能读取到一个被删除的元素，导致数据不一致或抛出异常。

什么是 "竞态条件 (Race Condition)" ?

当多个线程访问同一个资源时，由于线程执行顺序不确定，可能导致数据出现不确定性。在这种情况下，一般就说发生了 "竞态条件"。

这一段代码线程安全吗?

public static void main(String[] args) {
Thread thread = new Thread(() -> {
int j = 0;
// 运行 15 次
while (j != 15) {
// 休眠 500 毫秒
try {
Thread.sleep(500);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
ArrayList<Integer> list = new ArrayList<>();
// 线程 A 向列表中添加元素
Thread threadA = new Thread(() -> {
for (int i = 0; i < 1000; i++) {
list.add(i);
}
});
// 线程 B 若列表不为空则从列表中删除元素
Thread threadB = new Thread(() -> {
while (!list.isEmpty()) {
list.remove(0);
}
});
// 启动线程
threadA.start();
threadB.start();
// 等待线程执行完毕
try {
threadA.join();
threadB.join();
} catch (InterruptedException e) {
e.printStackTrace();
}
// 输出列表长度
System.out.println("List size: " + list.size() + ", 运行第 " + j + " 次。");
j ++;
}
});
thread.start();
}

复制代码

它不是线程安全的! 这段代码会出现竞态条件，即使两个线程操作不同的部分，也会出现问题!在这个例子中，线程A 向列表中添加元素，线程B 从列表中删除元素，因此线程A和线程B都会修改列表的内容。

当线程A和线程B同时访问列表时，可能会发生以下情况：

线程A 正在添加一个元素，线程B 在同时尝试删除第一个元素，这可能会导致删除操作抛出 IndexOutOfBoundsException 异常，因为该元素已被线程A移动到列表的末尾。
线程B 在删除列表中的第一个元素时，线程A 同一时间向列表添加元素，这可能会导致删除操作跳过一个或多个元素，因为列表中的元素已经被线程A添加到列表的末尾，从而导致数据不一致。

运行十五次，最终结果如下:

List size: 0, 运行第 0 次。
List size: 951, 运行第 1 次。
List size: 0, 运行第 2 次。
List size: 928, 运行第 3 次。
List size: 0, 运行第 4 次。
List size: 827, 运行第 5 次。
List size: 0, 运行第 6 次。
List size: 0, 运行第 7 次。
List size: 0, 运行第 8 次。
List size: 0, 运行第 9 次。
List size: 0, 运行第 10 次。
List size: 0, 运行第 11 次。
List size: 0, 运行第 12 次。
List size: 1000, 运行第 13 次。
List size: 0, 运行第 14 次。
进程已结束,退出代码0

复制代码

如果代码是线程安全的，那么最终输出的 list.size 应为 0。
因为线程B 不停地从列表中删除元素，当线程A 将所有元素添加到列表中后，线程B 会删除所有的元素，使得列表变为空。
在运行第 1、3、5、13 次很明显出现了数据不一致的问题。

使用 Vector?

使用 Vector 是很常见的解决方法之一，但是这真的推荐吗...?

Vector 是线程安全的，这没错，它的方法都使用了 synchronized 关键字来保证同步。因此多个线程可以同时访问 Vector，而不需要担心线程安全问题。
但是，在高并发的情况下，这种同步机制会带来较大的性能开销，导致 Vector 的性能比较差。这真的不推荐。让我们来看看 add、remove、get 方法源码。

public class Vector<E>
extends AbstractList<E>
implements List<E>, RandomAccess, Cloneable, java.io.Serializable
{
public boolean remove(Object o) {
return removeElement(o);
}
public synchronized boolean add(E e) {
modCount++;
ensureCapacityHelper(elementCount + 1);
elementData[elementCount++] = e;
return true;
}
public synchronized boolean removeElement(Object obj) {
modCount++;
int i = indexOf(obj);
if (i >= 0) {
removeElementAt(i);
return true;
}
return false;
}
public synchronized E get(int index) {
if (index >= elementCount)
throw new ArrayIndexOutOfBoundsException(index);
return elementData(index);
}
public synchronized E set(int index, E element) {
if (index >= elementCount)
throw new ArrayIndexOutOfBoundsException(index);
E oldValue = elementData(index);
elementData[index] = element;
return oldValue;
}
}

复制代码

我们可以看到 Vector 使用 synchronized 关键字来保证线程安全，这是一种独占锁的机制，即在同一时间只能有一个线程获得该锁，其他线程需要等待。
在高并发的情况下，由于多个线程需要对 Vector 进行操作，这会导致大量的线程阻塞等待，从而影响程序的性能。

另外，Vector 内部是通过数组来存储元素的，当数组已经满了之后，需要对数组进行扩容。
在扩容时，需要新建一个数组，并将原数组中的元素复制到新数组中。这个操作非常消耗性能，特别是在数据量比较大的情况下，会导致严重的性能问题。
而且，由于 Vector 进行同步处理，这会导致多个线程需要等待其他线程完成扩容操作才能进行后续的操作。

使用 CopyOnWriteArrayList?

这也是常见的解决方法之一，该类是线程安全的，支持高并发的读写操作。
但是，CopyOnWriteArrayList 的性能开销可能会比较高，因为每次写操作都会创建一个新的数组来保存数据，这可能会对内存使用产生一定的影响。
而且，由于写操作会进行复制，所以写操作的速度也会比较慢。接下来让我们分析一下 CopyOnWriteArrayList set 方法源码。

public class CopyOnWriteArrayList<E>
implements List<E>, RandomAccess, Cloneable, java.io.Serializable {
// The array, accessed only via getArray/setArray.
private transient volatile Object[] array;
// 获取当前数组
final Object[] getArray() {
return array;
}
// 设置新数组
final void setArray(Object[] a) {
array = a;
}
public E set(int index, E element) {
final ReentrantLock lock = this.lock;
lock.lock(); // 获取锁
try {
Object[] elements = getArray(); // 获取当前数组
E oldValue = get(elements, index);
if (oldValue != element) {
int len = elements.length;
Object[] newElements = Arrays.copyOf(elements, len);
newElements[index] = element; // 对复制后的数组进行写操作
setArray(newElements); // 将原数组替换为复制后的数组
} else {
// Not quite a no-op; ensures volatile write semantics
setArray(elements);
}
return oldValue;
} finally {
lock.unlock(); // 释放锁
}
}
// 其他方法的实现和 set 方法类似都需要先获取锁然后获取当前数组对复制后的数组进行写操作最后替换
}

复制代码

可以看到，每个写操作都需要获取一个独占锁，然后将当前数组复制一份进行修改，最后将原数组替换为复制后的数组。
由于每次写操作都会复制一份数组，所以在写操作较频繁的情况下，会带来一定的性能开销和内存消耗。
因此，在 读多写少 的情况下，CopyOnWriteArrayList 的性能表现会比较好，但在读多写多的情况下，绝对不是最佳选择。

在线程安全的情况下获得更好的性能? 解决方法?

接下来让我们分别分析 ConcurrentLinkedQueue 与 ConcurrentLinkedDeque。
ConcurrentLinkedQueue 是高效的、线程安全、无|界、非阻塞队列。它的高效主要体现在以下几个方面:

无锁并发: ConcurrentLinkedQueue 使用 CAS (Compare-And-Swap) 操作来实现无锁并发，避免了使用锁带来的各种问题。
多独立队列: ConcurrentLinkedQueue 中的每个线程在入队时都会被分配到一个独立的队列中，每个独立的队列都是非阻塞的，避免多个线程之间的竞争，大大提高效率。
避免不必要数据复制: ConcurrentLinkedQueue 先写入后读取，即在入队时直接将数据插入队列的尾部，而不是像 CopyOnWriteArrayList 那样在每次修改时都复制一份数据，避免不必要的数据复制开销。

下面是简要的 ConcurrentLinkedQueue 部分源码分析:

public class ConcurrentLinkedQueue<E> extends AbstractQueue<E> implements Queue<E>, Serializable {
public boolean add(E e) {
return offer(e);
}
public boolean offer(E e) {
checkNotNull(e);
final Node<E> newNode = new Node<E>(e);
for (Node<E> t = tail, p = t;;) {
Node<E> q = p.next;
if (q == null) {
// p is last node
if (p.casNext(null, newNode)) {
// Successful CAS is the linearization point
// for e to become an element of this queue,
// and for newNode to become "live".
if (p != t) // hop two nodes at a time
casTail(t, newNode); // Failure is OK.
return true;
}
// Lost CAS race to another thread; re-read next
}
else if (p == q)
// We have fallen off list. If tail is unchanged, it
// will also be off-list, in which case we need to
// jump to head, from which all live nodes are always
// reachable. Else the new tail is a better bet.
p = (t != (t = tail)) ? t : head;
else
// Check for tail updates after two hops.
p = (p != t && t != (t = tail)) ? t : q;
}
}
public E poll() {
restartFromHead:
for (;;) {
for (Node<E> h = head, p = h, q;;) {
E item = p.item;
if (item != null && p.casItem(item, null)) {
// Successful CAS is the linearization point
// for item to be removed from this queue.
if (p != h) // hop two nodes at a time
updateHead(h, ((q = p.next) != null) ? q : p);
return item;
}
else if ((q = p.next) == null) {
updateHead(h, p);
return null;
}
else if (p == q)
continue restartFromHead;
else
p = q;
}
}
}
public boolean remove(Object o) {
if (o != null) {
Node<E> next, pred = null;
for (Node<E> p = first(); p != null; pred = p, p = next) {
boolean removed = false;
E item = p.item;
if (item != null) {
if (!o.equals(item)) {
next = succ(p);
continue;
}
removed = p.casItem(item, null);
}
next = succ(p);
if (pred != null && next != null) // unlink
pred.casNext(p, next);
if (removed)
return true;
}
}
return false;
}
}

复制代码

我们逐个分析方法:

add
add 方法的实现是基于 offer 方法的调用，并返回 offer 方法的返回值。

offer
先创建一个新的节点，节点包含要添加的元素。
从队列的尾部 (tail) 开始往后遍历，找到最后一个节点 p。
如果p的下一个节点 q 为空，则将新节点添加到 p 的下一个节点上。
如果添加成功，返回 true。如果添加失败，重新读取 tail 的值，并重新执行步骤 2。
这里使用了 CAS 操作来保证安全。如果多个线程同时调用 offer 方法，只有一个线程能够成功地将节点添加到队列中。

poll
从队列的头部 (head) 开始往后遍历，找到第一个非空的节点 p。
如果 p 的元素 item 不为空，使用 CAS 操作将 p 的元素 item 设为 null。如果成功，返回item。
如果 p 的元素 item 为空，则说明该节点已经被其他线程删除，需要重新遍历队列。
如果遍历完整个队列都没有找到可用的元素，返回 null。
这里使用了 CAS 操作来保证安全。如果多个线程同时调用 poll 方法，只有一个线程能够成功地从队列中取出元素。

remove
从队列的头部 (head) 开始往后遍历，找到第一个包含指定元素的节点p。
如果找到了包含指定元素的节点 p，使用 CAS 操作将 p 的元素 item 设为null。如果成功，返回true。
如果 CAS 操作失败，则说明该元素已经被其他线程删除，需要重新遍历队列。
如果遍历完整个队列都没有找到指定的元素，则返回false。

相比之下，CopyOnWriteArrayList 为了支持元素快速查找和删除，使用了可重入锁 (ReentrantLock)，保证修改操作互斥，从而保证了线程安全性。频繁的加锁和解锁会降低性能，导致低效。
ConcurrentLinkedQueue 中的插入和删除操作都不需要锁，所以能够快速处理高并发 Qin (神奇的屏蔽雾) 况，使用单向链表实现，减少对共享数据的访问，提高性能。

与 ConcurrentLinkedQueue 类不同，ConcurrentLinkedDeque 类实现双向队列，因此它支持在队列头部和尾部插入和删除元素。
ConcurrentLinkedDeque 的高效体现在每个节点可以同时保存它的前驱节点和后继节点。

ConcurrentLinkedDeque 节点都保存了前驱节点和后继节点的引用，在执行插入和删除操作时，更新前驱节点和后继节点的引用即可。
而 ConcurrentLinkedQueue 节点只保存了后继节点的引用，在执行插入和删除操作时，需更新队头和队尾节点的引用，需使用 CAS 操作，相对更加耗时。

ConcurrentLinkedDeque 比 ConcurrentLinkedQueue在一些场景下更加高效。在此便不看源码了，如有有兴趣的开发者可以自行了解。
总而言之: 如果您需要双向队列来支持在头部和尾部插入和删除元素，可以选择使用 ConcurrentLinkedDeque。如果只需要在队列尾部插入和删除元素，可以选择使用 ConcurrentLinkedQueue。

关于 HashMap

为什么 HashMap 线程不安全?

因为它的操作不是原子性的，多线程同时进行 put 或 remove 操作可能会导致数据的不一致性或死锁。

这一段代码线程安全吗?

public static void main(String[] args) {
Thread thread = new Thread(() -> {
int j = 0;
// 运行 15 次
while (j != 15) {
// 休眠 500 毫秒
try {
Thread.sleep(500);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
HashMap<Integer, String> map = new HashMap<>();
// 线程 A 向 Map 中插入键值对
Thread threadA = new Thread(() -> {
for (int i = 0; i < 10000; i++) {
map.put(i, "value" + i);
}
});
// 线程 B 也向 Map 中插入键值对
Thread threadB = new Thread(() -> {
for (int i = 0; i < 10000; i++) {
map.put(i, "value" + i);
}
});
// 启动线程
threadA.start();
threadB.start();
// 等待线程执行完毕
try {
threadA.join();
threadB.join();
} catch (InterruptedException e) {
e.printStackTrace();
}
// 输出 Map 长度
System.out.println("Map size: " + map.size() + ", 运行第 " + j + " 次。");
j ++;
}
});
thread.start();
}

复制代码

这个例子创建了两个线程，每个线程都往 HashMap 中插入了 10000 个键值对。
在这个例子中，若线程安全，则最终 HashMap 的大小为 10000。让我们运行它，最终结果如下:

Map size: 10543, 运行第 0 次。
Map size: 12156, 运行第 1 次。
Map size: 10810, 运行第 2 次。
Map size: 9975, 运行第 3 次。
Map size: 10103, 运行第 4 次。
Map size: 10088, 运行第 5 次。
Map size: 10209, 运行第 6 次。
Map size: 10028, 运行第 7 次。
Map size: 10211, 运行第 8 次。
Map size: 10054, 运行第 9 次。
Map size: 10237, 运行第 10 次。
Map size: 10208, 运行第 11 次。
Map size: 9984, 运行第 12 次。
Map size: 10014, 运行第 13 次。
Map size: 10011, 运行第 14 次。
进程已结束,退出代码0

复制代码

稳定发挥，这是由于两个线程可能会同时插入具有相同的键的值，这会导致其中一个线程的插入操作被覆盖。

ConcurrentHashMap!

常见的解决方法之一 ConcurrentHashMap，它是线程安全的。

Segment? 不，它只是早期的实现方式，现在的实现方式已经不再使用 Segment 了。

当前实现使用了一种基于分段锁的技术，将哈希表分为多个桶，并在每个桶上使用独立的锁来实现线程安全性。
这种分段锁的设计可以保证在并发环境下，每个线程只需要获取相应的桶的锁，而不是对整个哈希表进行加锁，提高了并发性能

以下是 ConcurrentHashMap putVal 方法的源代码:

final V putVal(K key, V value, boolean onlyIfAbsent) {
if (key == null || value == null) throw new NullPointerException();
int hash = spread(key.hashCode());
int binCount = 0;
for (Node<K,V>[] tab = table;;) {
Node<K,V> f; int n, i, fh;
if (tab == null || (n = tab.length) == 0)
tab = initTable();
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
if (casTabAt(tab, i, null,
new Node<K,V>(hash, key, value, null)))
break; // no lock when adding to empty bin
}
else if ((fh = f.hash) == MOVED)
tab = helpTransfer(tab, f);
else {
V oldVal = null;
synchronized (f) {
if (tabAt(tab, i) == f) {
if (fh >= 0) {
binCount = 1;
for (Node<K,V> e = f;; ++binCount) {
K ek;
if (e.hash == hash &&
((ek = e.key) == key ||
(ek != null && key.equals(ek)))) {
oldVal = e.val;
if (!onlyIfAbsent)
e.val = value;
break;
}
Node<K,V> pred = e;
if ((e = e.next) == null) {
pred.next = new Node<K,V>(hash, key,
value, null);
break;
}
}
}
else if (f instanceof TreeBin) {
Node<K,V> p;
binCount = 2;
if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
value)) != null) {
oldVal = p.val;
if (!onlyIfAbsent)
p.val = value;
}
}
}
}
if (binCount != 0) {
if (binCount >= TREEIFY_THRESHOLD)
treeifyBin(tab, i);
if (oldVal != null)
return oldVal;
break;
}
}
}
addCount(1L, binCount);
return null;
}

复制代码

可以看出，ConcurrentHashMap 的插入操作使用了非常复杂的逻辑，并且涉及到了多线程之间的协作，非常复杂，在此便不分析了。

通过合理地利用哈希算法和并发控制机制，ConcurrentHashMap 可以高效的支持并发访问。因此，ConcurrentHashMap 是一种非常高效的并发哈希表实现。

ConcurrentHashMap 的轻量级实现 - ConcurrentSkipListMap?

ConcurrentSkipListMap 是一个基于跳表实现的线程安全的Map，比 ConcurrentHashMap 更适合于需要高效有序访问的并发场景，例如并发的数据范围查询。
如果您的业|务需要高效的有序访问并且不在乎内存消耗，那么 ConcurrentSkipListMap 可能更适合您。而如果您的业|务对内存消耗比较敏感，同时对有序性要求不是特别高，那么 ConcurrentHashMap 可能更合适。

Caffeine!

Caffeine 是一个基于 Java8+ 的高性能本地缓存库，其支持异步加载、过期自动回收等高级功能。
且提供了更多的缓存策略和配置选项，Caffeine 采用了一些新的技术，如缓存统计和突发模式等，以实现更好的性能和稳定性。
在高并发、大规模数据场景下，其性能表现更加优异。Caffeine 相比于 ConcurrentHashMap 具有更多的功能和优势。

针对 Caffeine 的使用可参考我写的简易工具类，此类将构建一个最简单的 Caffeine 缓存对象

public class CaffeineUtils {
/**
* 构建一个具有默认参数的 Caffeine 缓存。
*
* @param <K> 缓存键的类型
* @param <V> 缓存值的类型
* @return 一个配置好的 Caffeine 缓存
*/
public static <K, V> Cache<K, V> buildCaffeineCache() {
return Caffeine.newBuilder().build();
}
}

复制代码

在使用此方法时，需要使用泛型类型指定缓存键和值的类型。例如 Cache<String, Integer> cache = buildCaffeineCache(); 将构建一个缓存，其键和值的类型分别为 String 和 Integer。

您还可以通过 Caffeine 构建器的不同方法来配置缓存的行为。以下是 Caffeine 构建器常用方法及其含义:

initialCapacity(int initialCapacity): 设置缓存的初始容量。
maximumSize(long maximumSize): 设置缓存的最大容量，当缓存项数量超过该值时，缓存将尝试清除不常用的缓存项。
expireAfterAccess(Duration duration): 设置缓存项在最后一次访问之后过期的时间。
expireAfterWrite(Duration duration): 设置缓存项在创建或最后一次写入之后过期的时间。
refreshAfterWrite(Duration duration): 设置缓存项在创建或最后一次写入之后多久自动刷新。
weakKeys(): 设置缓存键为弱引用，当没有其他引用时，键将被垃圾回收。
weakValues(): 设置缓存值为弱引用，当没有其他引用时，值将被垃圾回收。
softValues(): 设置缓存值为软引用，当内存不足时，值将被垃圾回收。
recordStats(): 启用缓存统计信息收集。

这些方法中的大多数都接受一个 Duration 参数，该参数指定了时间段。例如，Duration.ofMinutes(10) 指定了 10 分钟的时间段。

此外，还有一些其他的方法用于更细粒度的配置，如 maximumWeight(long maximumWeight) 用于设置缓存的最大权重，weigher(Weigher<? super K, ? super V> weigher) 用于指定计算缓存项权重的方法。

在构建完缓存对象后，可以使用 Cache 接口提供的方法来操作缓存。例如，get(K key) 方法用于获取缓存中给定键的值，put(K key, V value) 方法用于将给定键值对添加到缓存中。

Caffeine: https://github.com/ben-manes/caffeine
温馨提示: 对于 Java 11 或更高版本，使用 3.x 否则使用 2.x.

结语:

wow，很感谢您看到了这里，这只是一个杂谈，我很高兴如果这篇杂谈对您有帮助！:)关于其余容器可能会在我有空的时候随便写写，也许? 我很建议您阅读下面的 参考文献!

参考文献:

结冰的离季

本帖最后由结冰的离季于 2023-2-22 21:50 编辑

新手: 能跑就行
老手: 能跑就行

2000000

本帖最后由 2000000 于 2023-2-22 22:10 编辑

结冰的离季发表于 2023-2-22 21:42
新手: 能跑就行
老手: 能跑就行

感谢建议!
笔者的话中有写到 "读者在阅读前应该对并发编程有一些了解，起码需要有开发经验。"
我并没有想把这个帖子写成类似于教程，更多的是分析 (也许)，并在分析过后给予一些建议。
这篇杂谈的表述已经尽可能较为简单一些，很多地方没有进行深入探讨
我应该会过段时间加入更多的实例问题分析，感谢建议~