问题背景

需要将10万条数据通过io写入初始链表中，再将十万条数据进行反序列、解析等操作写入es数据库

采用的线程池为自定义拒接策略的阻塞方式，

问题

写入的时候发现总是会少数据，即线程不安全。少掉的数量为大约1/5

解决路径

一开始很明确为线程安全问题，先通过加锁的方式，问题还是会出现

countlantch也用上，还是一样

后来注意到，之所以采用线程池的方式写入，是因为run的过程中有耗时的计算逻辑，

原来是中间的逻辑出现了反序列失败，

但是很奇怪，线程池提交线程之后，如果运行出错，不会报错吗？

我采用的线程接口是runnable，出错会无感知吗？

问题解决

list为什么线程不安全

先来看下list.add的源码

1
2
3
4
5
6

public boolean add(E e) {
    ensureCapacityInternal(size + 1);  // 检查容量，必要时扩容
    elementData[size++] = e;  //扩容后 添加元素
    return true;
}

可能的问题一，

ArrayIndexOutOfBoundsException

在多线程环境中时，多个线程同时进入add()方法，同时检查容量，例如当前容量为5，而已占用4。

若三个线程同时检查，都发现还有容量，则都同时添加元素。

由此导致ArrayIndexOutOfBoundsException。

问题二，

实际插入元素个数小于预期插入元素个数

list.size()小于总数即能说明问题，多个线程同时插入的问题

Vector保证线程安全

1
2
3
4
5
6
7

public synchronized boolean add(E e) {
    modCount++;
    ensureCapacityHelper(elementCount + 1);   //检查容量 必要时增长
    elementData[elementCount++] = e;
    return true;
}

与ArrayList不同的是，它的add()方法带有synchronized关键字。

这表明当线程调用该方法时，会自动占用锁，直到该线程的任务完成，期间不会放弃该锁

而且当线程占用该锁时，别的线程无法进入该实例对象调用带有synchronized关键字的方法。

这很好的避免了多线程竞争的现象，从而保证了并发安全。

Collections.synchronizedList(List list)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

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方法

1
2
3
4
5
6

public boolean add(E e) {
    synchronized (mutex) {  //锁住 SynchronizedCollection 集合类
        return c.add(e);
    }
}

当然，比较推荐的方法还是手工加锁，比较灵活，也比较容易加深理解

threadPoolExecutor虽然传入的runnable接口，但貌似会自动转为future会捕获异常，但是不会打印出来

这里要注意两个区别

1. execute方法和submit方法的区别

execute可以添加一个runnable任务，submit（）不仅可以添加runnable任务还可以添加callable任何，即有返回值

execute没有返回值，而submit在添加callable任务时，会有返回值，可以通过返回值来查看线程执行的情况

如果发生异常submit可以捕获异常，在任务结束时finally才会执行，不会当时处理。而execute会终止这个线程

2. list要注意是非线程安全的

要保证线程安全，可以对list加锁，或者使用一个比较古老的容器vector，线程安全的list效率比较低

或者使用arrayblockingArr或者linkedblockingArr阻塞队列