CopyOnWriteArrayList 的故事--一起看看java原生的读写分离

CopyOnWriteArrayList 是JUC中提供的，为了实现高并发的而提供的list容器之一。
对于大部分的业务场景，都是读多写少，并发度也基本卡在了读的位置。
通常支持并发的容器在解决并发时，采用是：
（1）数据分割，每个线程只操作属于当前线程自己的数据，如ThreadLocal （感兴趣的同学可以看我前文 《Java内存模型及Java关键字 volatile的作用和使用说明》  https://www.cnblogs.com/jilodream/p/9452391.html）
（2）元数据被操作时，通过锁来进行并发控制。（如Hashtable 、concurrentHashMap 等）
今天要说的CopyOnWriteArrayList的思想是读写分离：
如图，他的思路大概是这样的，

 
（0）提供一个数组来作为队列：
（1）所有的读操作直接读取数组，由于只有读操作，因此不会存在数据安全问题。
（2）所有的写操作是会将原始数据（数组）+改动操作（增删数据） 直接体现到另外一个新的数组中。(防盗连接：本文首发自http://www.cnblogs.com/jilodream/ )在写操作完毕之前，替换为原有的，外部读可以访问的数组。
（3）由于写操作才会有并发的问题，因此所有的写操作是通过锁来隔离的。而读操作都是建立在一个完整的，不被改动的数组中的，因此读也就不再需要锁了。
先来看下命名  
Copy  复制/拷贝
On  在...时间点
Write 写
Array  数组
List  队列
结合起来就是一个在写数据时进行拷贝的数组队列。一语道破该并发容器的核心。
怎么使用CopyOnWriteArrayList就不说了，他的常规使用与其他List并没有太大的区别。
我们直接来看源码（我这里采用的是JDK17）：
首先来看定义

1. public class CopyOnWriteArrayList<E>
2.     implements List<E>, RandomAccess, Cloneable, java.io.Serializable

复制代码

定义中首先指定该容器依次实现了
List, ---> 队列接口
RandomAccess, ---> 标记接口，支持随机访问
Cloneable, ---> 标记访问，支持拷贝
java.io.Serializable --->  标记接口，支持可序列化
至于什么是标记接口，可以看我的这篇文章（为什么这些java接口没有抽象方法？浅谈Java标记接口  https://www.cnblogs.com/jilodream/p/5986519.html）。
（ps. 不知道大家发现没有代表能力的接口jdk往往喜欢加-able，表示可以的。这在代码简洁之道这本书中也是推荐的命名方式之一。）
我们接下来看下它的核心源码。
首先是元数据部分：

1.     /**
2.      * The lock protecting all mutators.  (We have a mild preference
3.      * for builtin monitors over ReentrantLock when either will do.)
4.      */
5.     final transient Object lock = new Object();
7.     /** The array, accessed only via getArray/setArray. */
8.     private transient volatile Object[] array;

复制代码

定义了一个Object lock作为全局锁。用final transient 来修饰。
final 是为了防止锁被乱改，导致出现安全问题。
transient 则是表示在自动序列化时，不要主动来序列化该字段。这里也是了为了数据安全考虑的，防止正反序列化后，实例中的lock对象暴露出去，导致有安全问题。
注意在JDK1.8中，锁的实现使用的是ReentrantLock。但是在后续的版本中由于synchronized锁的优化，又使用了syncronized锁。源码中的注释也有这样写：

1.     /**
2.      * The lock protecting all mutators.  (We have a mild preference
3.      * for builtin monitors over ReentrantLock when either will do.)
4.      */<strong>
5.      当内置锁和 ReentrantLock 都能用的时候，我们略微倾向于使用内置锁。</strong>

复制代码

定义了用来存放Object[] array;使用了transient volatile来修饰。
使用transient  是为了解决序列化的问题。这里就会有一个疑问，为什么支持序列化，(防盗连接：本文首发自http://www.cnblogs.com/jilodream/ )并且又要在array 前边修饰transient，防止它序列化？这是为何，我们序列化不就是为了转存数据么？
使用volatile修饰是为了保证array 发生变化时，所有直接使用的线程可以及时的感知到。这里需要对volatile有一定的认知才能明白，不太明白的同学请看我的这篇文章
首先是构造方法：

1.     public CopyOnWriteArrayList() {
2.         setArray(new Object[0]);
3.     }

复制代码

注意看CopyOnWriteArrayList是将一个len=0的数组初始化到容器中的。而并不像ArrayList一样预留一个初始化大小的数组。这是为什么呢？
然后是核心的读方法：

1.     public E get(int index) {
2.         return elementAt(getArray(), index);
3.     }
4.    
5.         static <E> E elementAt(Object[] a, int index) {
6.         return (E) a[index];
7.     }

复制代码

很简单，从指定的数组中直接读出对应位置的元素，返回
核心的四个写方法：
（1）队列添加全新元素

1.     public boolean add(E e) {
2.         synchronized (lock) {
3.             Object[] es = getArray();
4.             int len = es.length;
5.             es = Arrays.copyOf(es, len + 1);
6.             es[len] = e;
7.             setArray(es);
8.             return true;
9.         }
10.     }

复制代码

1、写方法直接加了一把全局锁。
2、然后在同步的代码块中，将元数据加入到一个（比当前数组长度+1长度的）新数组中。
3、接着将新元素添加到队列的末尾中。
4、最后将新数组覆盖到原数组中。
5、由于原数组在定义时，使用了volatile关键字修饰。因此下次的读操作就会立刻感知到该元素，
（2）队列指定位置添加全新元素

1.     public void add(int index, E element) {
2.         synchronized (lock) {
3.             Object[] es = getArray();
4.             int len = es.length;
5.             if (index > len || index < 0)
6.                 throw new IndexOutOfBoundsException(outOfBounds(index, len));
7.             Object[] newElements;
8.             int numMoved = len - index;
9.             if (numMoved == 0)
10.                 newElements = Arrays.copyOf(es, len + 1);
11.             else {
12.                 newElements = new Object[len + 1];
13.                 System.arraycopy(es, 0, newElements, 0, index);
14.                 System.arraycopy(es, index, newElements, index + 1,
15.                                  numMoved);
16.             }
17.             newElements[index] = element;
18.             setArray(newElements);
19.         }
20.     }

复制代码

1、写方法直接加了一把全局锁。
2、然后判断当前添加位置的索引是否合法超出了len的合理范围，如果是，那么就抛出异常。
3、然后判断当前添加位置是不是数组的最后位置：
如果是的话，就把原数组的所有元素拷贝到新数组的中。新数组的长度等于老数组+1。
如果不是的话，就直接new 一个长度为len+1的数组。然后分别把老数组的前后两部分拷贝到新数组中。
4、将添加位置的元素设置为新元素。
（3）移出指定位置的元素

1.     public E remove(int index) {
2.         synchronized (lock) {
3.             Object[] es = getArray();
4.             int len = es.length;
5.             E oldValue = elementAt(es, index);
6.             int numMoved = len - index - 1;
7.             Object[] newElements;
8.             if (numMoved == 0)
9.                 newElements = Arrays.copyOf(es, len - 1);
10.             else {
11.                 newElements = new Object[len - 1];
12.                 System.arraycopy(es, 0, newElements, 0, index);
13.                 System.arraycopy(es, index + 1, newElements, index,
14.                                  numMoved);
15.             }
16.             setArray(newElements);
17.             return oldValue;
18.         }
19.     }

复制代码

1、移出方法首先加入一把全局锁
2、获取指定位置的元素
3、判断移出元素是不是队列的末尾：
如果是的话，(防盗连接：本文首发自http://www.cnblogs.com/jilodream/ )就把原数组的去除末尾元素的部分拷贝到新数组的中。新数组的长度等于老数组-1。
如果不是的话，就直接new 一个长度为len-1的数组。然后分别把老数组的前后两部分拷贝到新数组中。
4、返回查询到的指定元素
（4）移出指定元素

1.    public boolean remove(Object o) {
2.         Object[] snapshot = getArray();
3.         int index = indexOfRange(o, snapshot, 0, snapshot.length);
4.         return index >= 0 && remove(o, snapshot, index);
5.     }
7.     private boolean remove(Object o, Object[] snapshot, int index) {
8.         synchronized (lock) {
9.             Object[] current = getArray();
10.             int len = current.length;
11.             if (snapshot != current) findIndex: {
12.                 int prefix = Math.min(index, len);
13.                 for (int i = 0; i < prefix; i++) {
14.                     if (current[i] != snapshot[i]
15.                         && Objects.equals(o, current[i])) {
16.                         index = i;
17.                         break findIndex;
18.                     }
19.                 }
20.                 if (index >= len)
21.                     return false;
22.                 if (current[index] == o)
23.                     break findIndex;
24.                 index = indexOfRange(o, current, index, len);
25.                 if (index < 0)
26.                     return false;
27.             }
28.             Object[] newElements = new Object[len - 1];
29.             System.arraycopy(current, 0, newElements, 0, index);
30.             System.arraycopy(current, index + 1,
31.                              newElements, index,
32.                              len - index - 1);
33.             setArray(newElements);
34.             return true;
35.         }
36.     }
37.    

复制代码

1、先获取当前元素的位置
2、获取元素对应的位置（注意如果有多个相同的元素这里只取第一个找到的元素o），我们记为index
3、然后跳转到另外一个remove 重载的方法中。
4、在重载方法中加入全局锁
5、取出最新的数组current
6、判断新数组的current和之前查找时用的数组是不是一个数组：
   如果不是，说明有其他线程并发写操作了，此时开始重新找。
   首先在刚才index之前的元素都找一遍，如果找到，说明前移了，或者在index的更前方又插入该元素，此时这个新位置就是要移除的值。
   如果没找到又判断index>=新数组长度，那就说明是没有这个元素o了，直接就返回false了（因为上一步找的是index的位置，如果index都>=新数组长度，那么其实就已经遍历了）。
   如果还没有，那么判断index的位置是否仍等于元素o，如果是那么也判断找到了。
   如果还没有就判断index后边的位置是否有元素等于o
   这里其实相当于分了3部分，如果发生了写操作就判断 index之前，index的位置，index之后，这三块分别有没有o出现。
7、直接new 一个长度为len-1的数组。然后分别把老数组的前后两部分拷贝到新数组中。
最后这个remove方法其实看的人觉得很繁琐，为什么他没有直接进来一把全局锁，然后遍历查找最后生成新数组即可。其实我觉得是可以的，jdk1.7也的确是这样的。但是在后续的版本中将预查找的部分移出了锁的范围。
锁的粒度更小。对写的操作更友好。
以上就是CopyOnWriteArrayList的内部核心实现了。
现在又回到最初的问题。(防盗连接：本文首发自http://www.cnblogs.com/jilodream/ )为什么元数组array的len始终和当前存储的元素数保持一致。而不是像arraylist等其他容器，给一定的预留空间方便，方便写呢？
我认为主要基于以下考虑：
CopyOnWriteArrayList本质是为了读写分离更准确的说是为了读多写少的场景设计的。尽可能的压缩数组也为数组拷贝提供了优势。
array是及时变量，每次使用都是全新new一个出来，并不是在原有基础上写，没有必要为了性能预留空间。（核心原因）
另外的一个疑问为什么支持序列化，又要在array 前边修饰transient，防止它序列化？这是为何，我们序列化不就是为了转存数据么？
这主要是由于CopyOnWriteArrayList是使用写单独拷贝的方式来处理并发的。而array前边又有volatile修饰，也就是说array一旦发生变化，那么所有使用的线程在下次使用时就会立刻感知到。倘若我们序列化进行中，突然发生了写操作，并已经完成，我是应该接着序列化（此时就有错误），还是重新序列化（性能太差）？
因此又单独实现了该方法配合序列化用：

1.    private void writeObject(java.io.ObjectOutputStream s)
2.         throws java.io.IOException {
4.         s.defaultWriteObject();
6.         Object[] es = getArray();
7.         // Write out array length
8.         s.writeInt(es.length);
10.         // Write out all elements in the proper order.
11.         for (Object element : es)
12.             s.writeObject(element);
13.     }

复制代码

也就是我单独取出一份当前的数组放到es中，后边不管容器怎么写，我只序列化es中数据。
除此之外，队列的迭代也会有类似的问题。
我们知道队列的迭代，本质上就是使用迭代器依次的迭代元数据。而在CopyOnWriteArrayList中，和序列化相似，在迭代时，是将当前的元数组的引用指向到snapshot变量中。后续的迭代操作都是对该snapshot来操作的。
也就是说，在并发迭代时，我们迭代的数据是当前缓存的历史快照数据。如下：

1.     public Iterator<E> iterator() {
2.         return new COWIterator<E>(getArray(), 0);
3.     }
4.    
5.     static final class COWIterator<E> implements ListIterator<E> {
6.         /** Snapshot of the array */
7.         private final Object[] snapshot;
8.         /** Index of element to be returned by subsequent call to next.  */
9.         private int cursor;
11.         COWIterator(Object[] es, int initialCursor) {
12.             cursor = initialCursor;
13.             snapshot = es;
14.         }
15.     ....
16.     }

复制代码

可以看这个例子：

1. public class CopyOnWriteArrayListLearn {
2.     public static void main(String[] args) {
3.         CopyOnWriteArrayList<Integer> list = new CopyOnWriteArrayList();
4.         list.add(1);
5.         list.add(2);
6.         list.add(3);
7.         list.add(4);
8.         for (Integer item : list) {
9.             list.add(item * 10);
10.             System.out.println("item:" + item);
11.         }
12.     }
13. }

复制代码

输出如下，只输出了迭代开启一瞬间的容器中的元素：

1. Connected to the target VM, address: '127.0.0.1:61377', transport: 'socket'
2. item:1
3. item:2
4. item:3
5. item:4
6. Disconnected from the target VM, address: '127.0.0.1:61377', transport: 'socket'

复制代码

同时，直接使用迭代器进行写操作时，(防盗连接：本文首发自http://www.cnblogs.com/jilodream/ )就会因为是快照数据而没有意义，所以源码中直接将迭代器的写操作抛出异常了：

1.         public void remove() {
2.             throw new UnsupportedOperationException();
3.         }
5.         /**
6.          * Not supported. Always throws UnsupportedOperationException.
7.          * @throws UnsupportedOperationException always; {@code set}
8.          *         is not supported by this iterator.
9.          */
10.         public void set(E e) {
11.             throw new UnsupportedOperationException();
12.         }
14.         /**
15.          * Not supported. Always throws UnsupportedOperationException.
16.          * @throws UnsupportedOperationException always; {@code add}
17.          *         is not supported by this iterator.
18.          */
19.         public void add(E e) {
20.             throw new UnsupportedOperationException();
21.         }

复制代码

 

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！