List并发线程安全问题
2022-08-25 本文已影响0人
AC编程
一、发现并发问题
1.1 测试代码
public class Client {
public static void main(String[] args) {
List<Integer> list = new ArrayList<>();
new Thread(() -> {
for (int i = 0; i < 10000; i++) {
list.add(1);
}
},"A").start();
new Thread(() -> {
for (int i = 0; i < 10000; i++) {
list.add(1);
}
},"B").start();
new Thread(() -> {
for (int i = 0; i < 10000; i++) {
list.add(1);
}
},"C").start();
try {
TimeUnit.SECONDS.sleep(3);
}catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("list.size() = " + list.size());
}
}
开启三个线程，每个线程向ArrayList中插入1w条数据。之后等待三秒，等到每个线程都执行完毕时再查看ArrayList中的元素个数。运行结果：
Exception in thread "A" Exception in thread "B" java.lang.ArrayIndexOutOfBoundsException: 366
at java.util.ArrayList.add(ArrayList.java:459)
at Client.lambda$main$0(Client.java:15)
at java.lang.Thread.run(Thread.java:748)
java.lang.ArrayIndexOutOfBoundsException: 1851
at java.util.ArrayList.add(ArrayList.java:459)
at Client.lambda$main$1(Client.java:21)
at java.lang.Thread.run(Thread.java:748)
list.size() = 11680
Process finished with exit code 0
1.2 问题一：ArrayIndexOutOfBoundsException
我们先来看看ArrayList.add()的源码
/**
* Appends the specified element to the end of this list.
*
* @param e element to be appended to this list
* @return <tt>true</tt> (as specified by {@link Collection#add})
*/
public boolean add(E e) {
ensureCapacityInternal(size + 1); // Increments modCount!!
elementData[size++] = e;
return true;
}
在多线程环境中时，多个线程同时进入add()方法，同时检查容量，例如当前容量为5，而已占用4。三个线程同时检查，都发现还有容量，则都同时添加元素。由此导致ArrayIndexOutOfBoundsException。
1.3 问题二：实际插入元素个数小于预期插入元素个数
从运行结果可以看出，最终list.size()只有11680 <= 30000。我们希望能够插入30000个元素，可是实际上只插入了<= 30000个元素。还是从源码入手：
public boolean add(E e) {
ensureCapacityInternal(size + 1); // Increments modCount!!
elementData[size++] = e;
return true;
}
试想一下，如果多个线程同时向size位插入元素，且都没有来得及size++，那么导致的结果就是多个元素被插入在了同一个位置，相互抵消。
二、解决并发问题
2.1 使用Vector
早期，IT前人为了解决List在并发时出现的问题，引入了Vector实现类。Vetor的实现方式与ArrayList大同小异，它的底层也是一个数组，在添加时自增长。我们先来看看Vector.add()的源码
/**
* Appends the specified element to the end of this Vector.
*
* @param e element to be appended to this Vector
* @return {@code true} (as specified by {@link Collection#add})
* @since 1.2
*/
public synchronized boolean add(E e) {
modCount++;
ensureCapacityHelper(elementCount + 1);
elementData[elementCount++] = e;
return true;
}
与ArrayList不同的是，它的add()方法带有synchronized关键字。这表明当线程调用该方法时，会自动占用锁，直到这个线程的任务完成，期间不会放弃该锁。而且当线程占有该锁时，别的线程无法进入Vetor类调用带有synchronized关键字的方法。这很好的避免了多线程竞争的现象，从而保证了并发安全
我们现在将ArrayList换成Vetor再试试：
public class Client {
public static void main(String[] args) {
Vector<Integer> list = new Vector<>();
new Thread(() -> {
for (int i = 0; i < 10000; i++) {
list.add(1);
}
},"A").start();
new Thread(() -> {
for (int i = 0; i < 10000; i++) {
list.add(1);
}
},"B").start();
new Thread(() -> {
for (int i = 0; i < 10000; i++) {
list.add(1);
}
},"C").start();
try {
TimeUnit.SECONDS.sleep(3);
}catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("list.size() = " + list.size()); // 30000
}
}
2.2 使用Collections.synchronizedList(List<T> list)
我们现在先将ArrayList换成Collections.synchronizedList(List<T> list)试试效果：
public class Client {
public static void main(String[] args) {
List<Integer> list = Collections.synchronizedList(new ArrayList<>());
new Thread(() -> {
for (int i = 0; i < 10000; i++) {
list.add(1);
}
}, "A").start();
new Thread(() -> {
for (int i = 0; i < 10000; i++) {
list.add(1);
}
}, "B").start();
new Thread(() -> {
for (int i = 0; i < 10000; i++) {
list.add(1);
}
}, "C").start();
try {
TimeUnit.SECONDS.sleep(3);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("list.size() = " + list.size()); // 30000
}
}
我们通过Collections.synchronizedList(List<T> list)源码来分析：
List<Object> list = Collections.synchronizedList(new ArrayList<>());
// Collections.synchronizedList 源码
public static <T> List<T> synchronizedList(List<T> list) {
return (list instanceof RandomAccess ? //暂且不说
new SynchronizedRandomAccessList<>(list) :
new SynchronizedList<>(list)); //创建一个 SynchronizedList 实例
}
SynchronizedList(List<E> list) {
super(list); // 调用父类构造器
this.list = list;
}
SynchronizedCollection(Collection<E> c) {
this.c = Objects.requireNonNull(c); //要求传入的 集合类实例 非空 并将这个集合赋值给 c 变量
mutex = this; // 将自己赋值给 互斥锁变量
}
public static <T> T requireNonNull(T obj) {
if (obj == null)
throw new NullPointerException(); //为空则抛出异常
return obj;
}
Collections.synchronizedList()方法会返回一个SynchronizedList类的实例，其中包含了调用该方法时传入的集合，在构造期间，将SynchronizedCollection作为互斥锁。此时，当我们再调用add()方法：
public boolean add(E e) {
synchronized (mutex) { //锁住 SynchronizedCollection 集合类
return c.add(e);
}
}
这是，当调用add()方法，SynchronizedCollection会锁住自己，从而保证线程安全。当有线程正在使用mutex互斥锁时，其他变量无法占有该锁。
2.3 使用CopyOnWriteArrayList
我们现在先将ArrayList换成CopyOnWriteArrayList试试效果：
public class Client {
public static void main(String[] args) {
List<Integer> list = new CopyOnWriteArrayList<>();
new Thread(() -> {
for (int i = 0; i < 10000; i++) {
list.add(1);
}
}, "A").start();
new Thread(() -> {
for (int i = 0; i < 10000; i++) {
list.add(1);
}
}, "B").start();
new Thread(() -> {
for (int i = 0; i < 10000; i++) {
list.add(1);
}
}, "C").start();
try {
TimeUnit.SECONDS.sleep(3);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("list.size() = " + list.size()); // 30000
}
}
我们通过CopyOnWriteArrayList源码来分析：
List<String> copyOnWriteArrayList = new CopyOnWriteArrayList<>();
final transient ReentrantLock lock = new ReentrantLock();
private transient volatile Object[] array; // volatile线程可见 底层数组
//调用构造器
public CopyOnWriteArrayList() {
setArray(new Object[0]); //设置底层 array为一个空数组
}
//将传入 array 设置给 底层 array
final void setArray(Object[] a) {
array = a;
}
// 写时
public boolean add(E e) {
final ReentrantLock lock = this.lock; // 获取锁对象
lock.lock(); //锁住
try {
Object[] elements = getArray(); //获取到原有 数组
int len = elements.length; //获取原有长度
// 创建新数组为原有长度 + 1，将原有数据拷贝到新数组里，此时新数组有一个空位
Object[] newElements = Arrays.copyOf(elements, len + 1);
// 向空位添加新元素
newElements[len] = e;
// 将新数组 替换掉 旧数组
setArray(newElements);
return true;
} finally {
lock.unlock(); //解锁
}
}
// 获取列表长度
public int size() {
// 由于底层array线程可见，所以array一旦改变 size 也会被其他线程发现
return getArray().length;
}
final Object[] getArray() {
return array;
}
一个写时复制的List，写操作时加锁，过程中创建一个新的数组长度为原来的数组+1，并将原有数组元素添加到新数组中，之后添加新元素到末尾。读时不加锁，底层数组被volatile修饰，线程可见。他的核心就是：读时不阻塞，大大提升了读的速度。
2.3.1 线程可见的重要性
由于JVM中，栈空间是线程私有的。而栈中存在一个局部变量表，用于存储运行时需要的变量。在线程被开启时，线程会将自己所需要的变量都拷贝到自己的栈内存中。在循环过程中，局部变量表中的元素是无法感知到变量的改变的。简单来说，线程在使用外部变量时，无法感知到变量的改变。
2.3.1.1 线程不可见的一个栗子
List<Integer> list = new CopyOnWriteArrayList<>();
list.add(0);
new Thread(() -> {
while (list.get(0) == 0) { //循环获取 list 的 0 位元素
}
System.out.println("结束了");
}).start();
try {
TimeUnit.SECONDS.sleep(3);
}catch (InterruptedException e) {
e.printStackTrace();
}
list.add(0,1); //三秒后修改 list 的 0 位元素
// .... 程序无法结束 因为主线程对list的修改，线程内部无法感知
跑起来后，可以发现程序无法结束。
2.3.1.2 使用CopyOnWriteArrayList线程可见的栗子
// 使用 CopyOnWriteArrayList
List<Integer> list = new CopyOnWriteArrayList<>();
list.add(0);
new Thread(() -> {
while (list.get(0) == 0) {
}
System.out.println("结束了");
}).start();
try {
TimeUnit.SECONDS.sleep(3);
}catch (InterruptedException e) {
e.printStackTrace();
}
list.add(0,1);
// .... 程序可以结束，因为CopyOnWriteArrayList底层数组被 volatile 修饰
// 所以是线程间可见的
三、3个并发集合容器性能比较
3.1 只读10并发
3.1.1 测试代码
Vector
public class RunTester {
private static List<Integer> list = null;
public static void main(String[] args) throws InterruptedException {
list = new Vector<>();
list.add(0);
int theadNum = 10;
CountDownLatch countDownLatch = new CountDownLatch(theadNum);
long start = System.currentTimeMillis();
for (int i = 0; i < theadNum; i++) {
new Thread(() -> {
for (int j = 0; j < 500000; j++) {
list.get(0);
}
countDownLatch.countDown();
}).start();
}
countDownLatch.await(); //阻塞，直到执行完毕
long end = System.currentTimeMillis();
//list.size() = 1 耗时 => 204 ms
System.out.println("list.size() = " + list.size() + " 耗时 => " + (end - start) + " ms");
}
}
SynchronizedCollection
public class RunTester {
private static List<Integer> list = null;
public static void main(String[] args) throws InterruptedException {
list = Collections.synchronizedList(new ArrayList<>());
list.add(0);
int theadNum = 10;
CountDownLatch countDownLatch = new CountDownLatch(theadNum);
long start = System.currentTimeMillis();
for (int i = 0; i < theadNum; i++) {
new Thread(() -> {
for (int j = 0; j < 500000; j++) {
list.get(0);
}
countDownLatch.countDown();
}).start();
}
countDownLatch.await(); //阻塞，直到执行完毕
long end = System.currentTimeMillis();
//list.size() = 1 耗时 => 205 ms
System.out.println("list.size() = " + list.size() + " 耗时 => " + (end - start) + " ms");
}
}
CopyOnWriteArrayList
public class RunTester {
private static List<Integer> list = null;
public static void main(String[] args) throws InterruptedException {
list = new CopyOnWriteArrayList<>();
list.add(0);
int theadNum = 10;
CountDownLatch countDownLatch = new CountDownLatch(theadNum);
long start = System.currentTimeMillis();
for (int i = 0; i < theadNum; i++) {
new Thread(() -> {
for (int j = 0; j < 500000; j++) {
list.get(0);
}
countDownLatch.countDown();
}).start();
}
countDownLatch.await(); //阻塞，直到执行完毕
long end = System.currentTimeMillis();
//list.size() = 1 耗时 => 53 ms
System.out.println("list.size() = " + list.size() + " 耗时 => " + (end - start) + " ms");
}
}
3.1.2 分析
// Vector.get()
public synchronized E get(int index) {
if (index >= elementCount)
throw new ArrayIndexOutOfBoundsException(index);
return elementData(index);
}
// Synchronized 集合 get()
public E get(int index) {
synchronized (mutex) {
return list.get(index); //调用 List.get()方法
}
}
// CopyOnWriteArrayList.get()
public E get(int index) {
return get(getArray(), index);
}
private E get(Object[] a, int index) {
return (E) a[index];
}
从运行耗时来看，CopyOnWriteArrayList的性能比其他两个好得多。从源码来看，Vector和Synchronized集合类都使用到了synchronized关键字，锁住了整个方法。这样使得多条线程并发访问get()方法时，只会同时有一个线程在真正调用get()方法。而CopyOnWriteArrayList的get()方法没有使用到锁，所以所有线程都是一起访问的，所以它的性能更好。
3.2 只写10并发
3.2.1 测试代码
Vector
public class RunTester {
private static List<Integer> list = null;
public static void main(String[] args) throws InterruptedException {
list = new Vector<>();
list.add(0);
int theadNum = 10;
CountDownLatch countDownLatch = new CountDownLatch(theadNum);
long start = System.currentTimeMillis();
for (int i = 0; i < theadNum; i++) {
new Thread(() -> {
for (int j = 0; j < 500000; j++) {
list.add(0);
}
countDownLatch.countDown();
}).start();
}
countDownLatch.await(); //阻塞，直到执行完毕
long end = System.currentTimeMillis();
//list.size() = 5000001 耗时 => 228 ms
System.out.println("list.size() = " + list.size() + " 耗时 => " + (end - start) + " ms");
}
}
SynchronizedCollection
public class RunTester {
private static List<Integer> list = null;
public static void main(String[] args) throws InterruptedException {
list = Collections.synchronizedList(new ArrayList<>());
list.add(0);
int theadNum = 10;
CountDownLatch countDownLatch = new CountDownLatch(theadNum);
long start = System.currentTimeMillis();
for (int i = 0; i < theadNum; i++) {
new Thread(() -> {
for (int j = 0; j < 500000; j++) {
list.add(0);
}
countDownLatch.countDown();
}).start();
}
countDownLatch.await(); //阻塞，直到执行完毕
long end = System.currentTimeMillis();
//list.size() = 5000001 耗时 => 260 ms
System.out.println("list.size() = " + list.size() + " 耗时 => " + (end - start) + " ms");
}
}
CopyOnWriteArrayList
public class RunTester {
private static List<Integer> list = null;
public static void main(String[] args) throws InterruptedException {
list = new CopyOnWriteArrayList<>();
list.add(0);
int theadNum = 10;
CountDownLatch countDownLatch = new CountDownLatch(theadNum);
long start = System.currentTimeMillis();
for (int i = 0; i < theadNum; i++) {
new Thread(() -> {
for (int j = 0; j < 500000; j++) {
list.add(0);
}
countDownLatch.countDown();
}).start();
}
countDownLatch.await(); //阻塞，直到执行完毕
long end = System.currentTimeMillis();
//大于10分钟 不想等了
System.out.println("list.size() = " + list.size() + " 耗时 => " + (end - start) + " ms");
}
}
3.2.2 分析
从运行结果可以看出，CopyOnWriteArrayList的写性能，实在是太差了，这是为什么呢？我们来看看源码：
// CopyOnWriteArrayList.add()
public boolean add(E e) {
final ReentrantLock lock = this.lock;
lock.lock();
try {
Object[] elements = getArray(); //获取旧数组
int len = elements.length; //长度
//创建一个原来数组长度 + 1 的新数组，复制原有内容到新数组
Object[] newElements = Arrays.copyOf(elements, len + 1);
//添加一个元素
newElements[len] = e;
//替换掉旧数组
setArray(newElements);
return true;
} finally {
lock.unlock();
}
}
CopyOnWriteArrayList使用写时复制。添加新元素时，每次仅扩容1，然后将旧数组内容拷贝到新数组中，所以它的扩容，是每次都会触发的。这也是导致写性能差最主要的原因。但是为什么要只扩容1呢？为何不像Vector一样，每次扩容两倍呢？我猜测，这应该是为了避免维护一个size变量，size表示当前实际存入元素的个数。而每次扩容1，那么数组的长度就是size，所以省去了size的并发修改。简单来说，强悍的读取性能，是牺牲写入性能换来的。所以这也表明，CopyOnWriteArrayList容器虽好，可不要滥用。在写多于读的情况下，它的性能甚至比其他两个并发容器都要差得多。
转载自：List线程安全问题
