多线程下使用容器（上）

Posted 2021-04-21 我们都是小青蛙

最好使用电脑观看，投机取巧是学不到东西的，对自己诚实一点哈。往期文章参考：

普通容器的并发操作

我们之前在容器那一集介绍的各种容器，包括实现了List接口的ArrayList、LinkedList等、实现了Set接口的HashSet、TreeSet等、实现了Queue接口的LinkedList、PriorityQueue等、实现了Map接口的HashMap、TreeMap啥的，都是普通容器，换句话说就是线程不安全的，也就是说在多线程下同时调用容器的各种操作会产生安全性问题。

小贴士：

如果大家不清楚容器是什么，容器怎么用，可以出门左转去看容器那一集。

先看个例子啊：

import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.CountDownLatch;

public class LinkedListConcurrentDemo {

    static class Increament {
        private int i = 0;

        public synchronized int increaseAndGet() {
            return ++i;
        }
    }

    public static void main(String[] args) {

        CountDownLatch countDownLatch = new CountDownLatch(20);

        List<Integer> list = new LinkedList<>();
        Increament increament = new Increament();

        for (int i = 0; i < 20; i++) {
            new Thread(new Runnable() {
                @Override

                public void run() {
                    for (int i = 0; i < 10000; i++) {
                        list.add(increament.increaseAndGet());
                    }
                    countDownLatch.countDown();
                }
            }).start();
        }

        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }

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

我们创建了一个静态内部类Increament，提供了一个线程安全的increaseAndGet方法，每次获取的i的值都加1。然后我们先创建了一个LinkedList对象，然后创建了20个线程，每个线程中都执行list.add10000次，每次插入的元素都是increaseAndGet方法的返回值，也就是插入的值每次都加1。最后我们期望的执行结果当然是size的值是200000了，可是现实中的执行结果是(每次可能都不一样)：

list.size()：198464

size的值是小于200000的。这是因为add操作不是线程安全的，我们看一下它的源代码(为方便理解，只提取出它的关键代码)：

public class LinkedList<E> {
    int size = 0;     //List中元素数量
    Node<E> first;    //表示头节点
    Node<E> last;     //表示尾节点

    private static class Node<E> {  //双向链表的节点
        E item;
        Node<E> next;
        Node<E> prev;

        Node(Node<E> prev, E element, Node<E> next) {
            this.item = element;
            this.next = next;
            this.prev = prev;
        }
    }

    public boolean add(E e) {
        final Node<E> tmp = last;     //操作1
        final Node<E> newNode = new Node<>(tmp, e, null);     //操作2
        last = newNode;     //操作3
        if (tmp == null)  //链表为空时的插入
            first = newNode;
        else  //链表不为空时的插入
            tmp.next = newNode;     //操作4
        size++;     //操作5
        return true;
    }
}

我们看到，LinkedList内部其实是维护了一个双向链表，每次的add操作就是向链表尾部插入一个节点。假设此时链表不为空(你可以自己分析一下链表为空的时候的插入过程～)，我们插入一个节点的过程就是：

1. 将指向最后一个节点的引用last拷贝一个备份tmp。

2. 新创建一个节点，将它的prev指向原来的尾部节点。

3. 然后把自己设置为尾部节点。

4. 然后将原来的尾部节点的next指向新的尾部节点。

5. 将size自增1。

假设现在已经有3个节点了，现在要有一个线程要插入节点4，另一个线程要插入节点5，如图：

我们分析一下这个插入过程：

可以看到，如果碰到如图所示的执行时序，那么线程1和线程2在执行完操作1、2、3、4之后，相当于只插入了节点5，这一点导致了20个线程个执行10000次add操作后最后真实插入的节点数会比200000小。

另外，按照我们预期，每执行一次add操作，size的值就会加1，而由于size++不是原子操作，所以多线程并发执行这个操作的时候的结果比预想值要小，这一点会导致size的值并不是所有线程执行的add操作次数，也不是链表中真实的节点数，而是比真实插入的节点数要小。

小贴士：

我们上边只是输出了size的值，你有木有办法搞到`list`中真实插入的节点的数量呢？由于`LinkedList`中的`Node`类是private的，而且首尾节点`first`和`last`也是private，在类外无法访问，不过可以使用反射的各种功能哈。你可以试试，我这就不写了～

LinkedList的add方法是不安全的，是因为对于共享可变的链表数据，没有用锁把这个操作保护起来。不止LinkedList的add方法，我们之前唠叨过的所有的容器，只要它们中的某个方法在未经加锁的情况下访问了共享、可变的变量，那这个操作就是不安全的。

同步容器

所以，如果我们想让LinkedList的add操作是线程安全的咋办呢？当然是加锁喽：

synchronized (list) {
    list.add(increament.increaseAndGet());
}

让所有的add操作被锁list保护起来，这样一个线程在调用add方法的时候另一个线程就不可以同时调用add方法了。

但是这是交给客户端程序猿来加锁保护的，不同的客户端程序猿水平不同，如果在设计容器类的时候就把它设计称线程安全的岂不是更好？所以设计java的大叔们给我们做了很多同步容器，其实就是在容器类的各个操作都声明为同步方法，比如对于List，他们定义了一个叫Vector的类，我们看一下这个类的画风：

public class Vector<E> extends AbstractList<E> implements List<E>, RandomAccess, Cloneable, java.io.Serializable {

    public synchronized int size() { ... 省略具体实现 }

    public synchronized boolean isEmpty() { ... 省略具体实现 }

    public synchronized boolean add(E e) { ... 省略具体实现 }

    public synchronized int indexOf(Object o, int index) { ... 省略具体实现 }

    ... 还有好多同步方法
}

也就是说，它把各种对外开放的方法都声明为synchronized，简单暴力，不过管用。所以除了对add方法进行加锁的解决方案以外，我们还可以把原来的LinkedList换成是Vector。

类似Vector，他们还定义了Map的同步容器Hashtable，也是把各种方法都加了synchronized声明，从而保护内部共享数据。

另外，设计java的大叔为了我们方便的把普通容器转为 同步容器，也就是容器的各种操作都被锁保护着 ，在Collections工具类里提供了一系列的方法：

也就是说，我们可以这么写：

List<Integer> list = Collections.synchronizedList(new LinkedList<>());

这样list的各种操作就都被锁保护，在多线程环境下可以放心的调用了。

并发容器

由于同步容器只是简单的在方法调用时采用加锁保护，虽然简单易用解决问题，但是性能不太好。所以在JDK升级到第5个版本的时候，设计java的大叔们编写了一个java.util.concurrent包，这个包里定义了许多新的在多线程环境中使用的容器和工具，一般情况下我们把这个新定义的包中的容器称为并发容器，之前的Vector、Hashtable以及Collections.synchronizedXXX方法生成的容器称为同步容器。只是一个叫法而已，不必太纠结～ 下边我们重点看几个比较重要的并发容器。

ConcurrentHashMap

我们知道HashMap内部维护了一个数组，每个数组元素都代表了一个链表，由于采用了散列的方式存储和获取元素，所以速度会很快。而HashMap的操作是线程不安全的，Hashtable是HashMap的线程安全版本，它在每个访问底层数组的方法上都加了synchronized，确保每次只有一个线程可以访问到该共享的底层数组。

因为对所有操作一锁到底，所以Hashtable的性能不太好。除了性能问题，Hashtable也满足不了我们在多线程环境下的一些需求，比如下边这段代码：

public void method(Hashtable<String, String> map) {
    if (!map.containsKey("1")) {
            map.put("1", "xxx");
        }
}

我们需要判断某个键对应的值是否为null，如果为null则插入，不为null就什么都不做。虽然containsKey和put方法都是同步方法，但是整个操作却不是同步的，所以在多线程环境下，仍然可能出现安全性问题。如果需要保证这个操作的原子性的话，需要我们额外的加锁。如果一个容器既能解决这个问题，又能极大的提升性能该多好。

于是设计java的大叔们就设计了ConcurrentHashMap。随着一代又一代java版本的升级，为了更好的提升这个类的性能，设计java的大叔们不断优化它的实现，以至于越来越复杂，我们这里挑一个比较简单的实现版本来唠叨它的原理，如果你想看到最新最复杂的实现方案，可以到最新版本的JDK里查看源代码。

为了让大家容易理解，我只把几个重要的字段提取了出来，看一下ConcurrentHashMap的一个大致结构：

public class ConcurrentHashMap<K, V> {

    final Segment<K,V>[] segments;

    static final class HashEntry<K,V> {
        final K key;
        volatile V value;
        volatile HashEntry<K,V> next;
    }

    static final class Segment<K,V> extends ReentrantLock {
        transient volatile HashEntry<K,V>[] entrys;
    }
}

一个ConcurrentHashMap对象里维护了一个Segment[]数组，每个Segment对象里又维护了一个HashEntry[]数组，每个HashEntry对象都是链表的一个节点，一个ConcurrentHashMap对象的结构画图表示出来就是这样的：

可以看出来，每个Segment对象其实都相当于是一个小的HashMap，都是由数组和链表组成的，相当于整个ConcurrentHashMap对象由若干个小的HashMap组成。下边我们看怎么来构造一个ConcurrentHashMap对象，先看一个构造方法：

public ConcurrentHashMap(int initialCapacity, float loadFactor, int concurrencyLevel)

各个参数值的描述和默认值：

稍后我们完整的分析一下对象的创建过程，先把剩下的构造方法都认识一下：

好的，接下来我们执行new ConcurrentHashMap(33, 0.75, 15)来创建一个ConcurrentHashMap对象，也就是指定初始容量initialCapacity=33，加载因子loadFactor=0.75，当前更新线程的估计数concurrencyLevel=15。在我们介绍具体如何确定segments数组大小以及entrys数组大小之前，先看一些位运算，如果你不熟悉的话可能有些懵逼，我们知道，一个int值是有32位的，每向左移动一位，就意味着值扩大一倍：

下边再来分析一下如何确定segments数组大小(为了方便理解，只提取出了重要代码)：

int sshift = 0;     //代表左移的次数

int ssize = 1;      //segments数组的大小

while (ssize < concurrencyLevel) {  //当segments数组的大小小于concurrencyLevel时，ssize左移翻倍

    ++sshift;   //左移次数加1
    ssize <<= 1;    //ssize左移一位，值扩大一倍
}

this.segmentShift = 32 - sshif;
this.segmentMask = ssize - 1;

由于我们创建对象时concurrencyLevel取的值是15，我们画图看一下这个过程：

ssize代表segments数组的大小，它最后的值为16，sshift代表ssize的左移次数，它最后的值为4。从这个过程中我们看出：segments数组大小是比指定的concurrencyLevel大的最近的2的倍数。也就是说如果我们指定的是3，那么segments数组大小就是4，如果我们指定的是7，那么segments数组大小就是8。

segmentShift和segmentMask都是ConcurrentHashMap的一个字段，看一下它们的值：

所以，segmentShift就代表ssize在一个int变量里移动的剩余位数，一个int变量是32位，如果说ssize移动了4次，那么segmentShift就是28。segmentMask就代表ssize对应的二进制最高位之后的位都是1的数。这两个字段我们之后会用到，先理解一下它们的意思哈～

再看一下确定entrys数组大小的代码：

int c = initialCapacity / ssize;    //期望初始容量和segments数组大小比值
if (c * ssize < initialCapacity)
    ++c;
int cap = 2;    //cap代表entrys数组大小，最小为2
while (cap < c)     //保证entrys数组大小必须为2的倍数
    cap <<= 1;

变量c代表每个segment里应该存储键值对的个数，也就是HashEntry对象的个数。 我们现在指定的initialCapacity的值为33，上边计算出来的segments数组的大小是16，也就是ssize的值是16，33/16的结果为2，而2*16 < 33，所以进行自增，最后c的值就是3，也就是每个segment对象里放3个键值对的话正合适。但是cap才真正代表entrys数组大小，也就是entrys数组大小是比每个segment里应该存储键值对的个数大的最近的2的倍数，所以最后entrys数组大小为4。

至此，我们了解完了segments数组和entrys数组大小的确定过程，想必你有点不解的是：为什么segments数组和entrys数组大小都要是2的倍数呢？这还要从定位一个键值对说起。

不论是调用put方法，还是调用get方法，我们都需要通过键的哈希码来找到它在数组中的位置，以前在介绍HashMap的时候我们介绍过一种定位方式：

1. 通过hashCode/table.size()确定在数组中的位置。

2. 遍历在该数组元素处的链表，调用equals方法判断给定的键是否与对应的节点的匹配，如果匹配就意味着找着位置了，不匹配就没找着。

在ConcurrentHashMap中这种查找稍微复杂一些，它的过程是：

1. 为了保证键产生的哈希码更均匀，对给定的键的哈希码进行在此再次哈希。

   由于哈希码是一个整数值，如果在获取数组位置的时候直接取模的话，会有很多高位用不到。比方说数组的大小是10，那么不管哈希码如何复杂，只需要看整数值的个位数部分就好了，比如说11和214812319311的个位数都是1，所以最后产生的数组位置都是1，为了让最后产生的数组位置更加均匀，所以有必要再次进行一遍哈希操作，下边是一种比较复杂的哈希函数：

   private int hash(Object k) {
       h ^= k.hashCode();
       h += (h <<  15) ^ 0xffffcd7d;
       h ^= (h >>> 10);
       h += (h <<   3);
       h ^= (h >>>  6);
       h += (h <<   2) + (h << 14);
       return h ^ (h >>> 16);
   }
   

   也许你看不懂它的原理，但是它的确能产生比较均匀的哈希值～

2. 定位所属的segments数组位置。

   Segment<K, V> segmentFor(int hash) {
       int tmp = hash >>> segmentShift;
       int index = tmp & segmentMask;
       return segments[index];
   }
   

   比方说我们通过再次哈希产生了一个很复杂的哈希码，它的二进制长这样：01011000010111000110100001000101，然后需要把他向又移动segmentShift位，也就是28位，意思就是让高4位参与到求segments数组位置中，这个过程就是：

   

   再和上边产生的segmentMask求交集，这个操作的过程就是：

   

   所以最后的结果二进制就是0101，也就是十进制的5，就是说定位到segments数组的第5号元素。这个过程需要注意的是：这里采用的移位运算来获得最后的结果，比之前的取模运算更加高效，但是每次左移一位就是将原值乘2，右移一位就是将原值除以2，所以才有了segments数组大小必须是2的倍数的规定。

3. 从确定的segment中定位所属的entrys数组位置。

   这个过程和定位在segments数组中的位置差不多，就不唠叨了，随后的结论就是：entrys数组大小必须是2的倍数。

4. 从定位的entrys数组元素开始，遍历HashEntry链表，调用equals方法比较指定的键是否与该HashEntry节点匹配。

好了，定位操作说完了，回过头再来看构造方法中loadFactor的作用。我们知道，在HashMap中，只要节点数量和数组大小的比值超过了loadFactor的大小，那么就会创建一个更大的数组，并把各个Entry节点重新散列到新的数组中。但是在ConcurrentHashMap中，segmengs数组从创建ConcurrentHashMap对象后就不会再改变，每个Segment对象都代表一个小的HashMap，所以当一个Segment对象中的节点数量和本对象中entrys数组大小的比值超过加载因子后，会对本Segment对象的entrys数组进行扩容并重新散列元素，这个过程不会影响到别的 Segment 对象！这个过程不会影响到别的 Segment 对象！这个过程不会影响到别的 Segment 对象！，好了，重要的事情说了3遍。

再来唠叨一下ConcurrentHashMap提供的一些操作。

1. V put(K key, V value)操作

   put操作首先会根据指定的key定位到对应的Segment对象，因为segments数组在创建ConcurrentHashMap对象之后就不会再改动了，所以每个key对应的Segment对象是不会变的，所以这个过程是不需要加锁的。然后再在该Segment对应的entrys数组里进行插入或更新操作，这个过程需要加锁，不同的Segment的数据由不同的锁来保护。（put操作有两层含义，第一是如果指定的key在entrys中，那么就更新它的值，第二是如果指定的键不在entrys中，就插入一个新的键值对）

   也就是说整个ConcurrentHashMap的数据被拆成若干个Segment，不同的Segment的数据由不同的锁来保护，这个就是所谓的分段锁，这样就减小了锁的粒度，从而减弱了锁的竞争，达到了提高性能的目的。

   在使用时有一点需要注意，与 HashMap 不同的是，ConcurrentHashMap 的键和值都不允许为null，因为这会造成下边的get方法的歧义。

2. V get(Object key)操作

   get也需要首先根据指定的key定位到对应的Segment对象，然后再在该Segment对象对应的entrys数组里进行哈希查找操作。但是有趣的是， ConcurrentHashMap 的 get 操作并不需要加锁，只有在获取的 HashEntry 对象的 value 字段为 null 的情况下才会加锁重新读。定位Segment对象不加锁可以理解，毕竟一个key对应的Segment对象不会变化，而如果在Segment对象对应的entrys数组里进行哈希查找，不加锁的话是不是会造成安全性问题呢？我们接下来分析一波～

   在一个Segment中根据key查找节点大致是这么两步：

• 查找对应的entrys数组元素位置。

• 从该位置开始遍历链表，使用equals方法查看指定key是否与该节点匹配。

如果不加锁的话，在一个线程执行get操作的时候，另一个线程可能对该Segment进行修改、扩容、插入、移除等操作，我们一个一个分析一下。

• 修改情况

  如果在一个线程在调用get的过程中有别的线程把该节点的value给修改了咋办？让我们再看一遍HashEntry的源代码：

  static final class HashEntry<K,V> {
      final K key;
      volatile V value;
      volatile HashEntry<K,V> next;
  }
  

  擦亮你的眼睛，看到value是被volatile修饰的，所以任何线程的修改对其他线程是立即可见的！所以get操作会把修改后的值获取到，而不会拿到一个过期的值。

• 扩容情况

  假设某个Segment对象对应的entrys数组大小为2，然后只有一个节点，key是整数值1，value是字符串"a"，假设这个节点叫entry1，这个节点在0号元素所在的链表处：

  

  在此时线程t1调用get(1)来获取key为1时的值，它计算出了对应的entrys数组元素位置为0号元素，就在此时，另一个线程添加第二个元素entry2的时候需要进行扩容了，也就是entrys数组换成一个大小为4的新数组，所有节点需要进行重新散列以确定它们的位置，假设此时的第二个节点被安排到了2号元素所在的链表上，就像这样：

  

  然后线程t1才开始继续执行从0号元素开始遍历链表，结果当然是没有找到！所以在这种情况下，虽然节点在链表中，并不能立马能对get可见，这是ConcurrentHashMap不好的地方之一。

• 插入情况

  再看假设线程t1需要调用get(3)来获取一下key为3时的值，计算后对应的entrys数组元素是2号元素，它查看到2号元素的头节点为entry1。正在此时，另一个线程t2忽然插入一个新节点entry3，这个节点的key为3，value为"c"，经过计算后对应的entrys数组元素是2号元素。由于HashEntry的next字段是被final修饰的，所以一旦创建对象后，next值就不能改了，所以不能直接插到链表尾节点后边，只能从头部替换，这个过程的图示是这样的：

  

  如果线程t2执行完了插入操作，t1才开始继续执行链表遍历，虽然此时entry3节点已经插入到了链表，但是由于线程t1是从entry1开始遍历的，所以最后也每匹配不到entry3。所以插入过程中也可能产生往entrys数组中加入一个元素后，并不能立马能对get可见。

  另外，由于指令重排序的原因，在插入一个新节点的时候，有可能对该新节点已经插入到链表中了，可是该节点的value字段的赋值还没完成，所以如果调用get方法获取到了节点，还需要判断一下节点的value值是不是为null，如果为null的话，需要加锁重新获取。这也是为什么在添加键值对的时候，值不许为null的原因。

• 移除操作

  跟插入操作类似，移除操作也可能造成get方法获取的值不是最新的值，具体过程就不写了。

所以，get方法的基本过程就是：先找到key对应的Segment对象，再找到该Segment中对应的entrys数组元素，然后对该元素代表的链表进行遍历来判断是否存在key相同的节点以及获得该节点的value。但由于遍历过程中其他线程可能对链表结构做了调整，因此get方法返回的可能是过时的数据，由于这一点，ConcurrentHashMap也被称为具有弱一致性。如果要求强一致性，那么必须使用Collections.synchronizedMap()方法或者直接使用Hashtable。

3. size操作

   由于整个ConcurrentHashMap的数据被划分到多个Segment中，不同的Segment用不同的锁来保护。但是对于size操作来说，需要获取所有Segment中的Entry节点数量，我们最先想到的肯定是在执行size操作前把所有的Segment锁都获取到，把各个Segment中的Entry节点数量加起来返回之后再释放掉锁。但是设计的大叔觉得可能在执行size操作的时候并没有别的线程执行增删操作，那我们加锁不就浪费资源了么？所以他们定义了在Segment中定义了一个叫modCount的字段，每当这个Segment中有增删操作进行的时候，都把这个字段加1。

   然后在进行size操作时，先以不获取锁的方式计算所有Segment中的Entry节点数量的和，并且计算所有modCount字段的和，之后再重复进行计算一次，如果两次的modCount字段的和一致，则认为在执行方法的过程中没有发生其他线程修改ConcurrentHashMap的情况，返回获得的值。如果不一致，加锁后进行操作。

   所以我们最好避免在多线程环境下使用size方法，因为它可能获取所有Segment的锁。

另外，ConcurrentHashMap还提供了一些先判断再设置的原子操作，方便我们使用：

写入时复制

我们之前介绍过如何使用迭代器来遍历Collection容器，不论是普通容器还是同步容器，都可以使用迭代器来遍历：

Vector<String> vector = new Vector<>();
vector.add("1");
vector.add("2");
vector.add("3");

Iterator<String> iterator = vector.iterator();

while (iterator.hasNext()) {
    String s = iterator.next();
    System.out.println(s);
}

但是如果在一个线程使用迭代器的时候另一个线程改变了容器中的数据，可能对遍历过程产生意想不到的影响，所以设计java的大叔并不允许这种一边使用迭代器遍历，一边修改数据的行为，如果一个线程在遍历的过程中另一个线程修改了数据，将会抛出ConcurrentModificationException异常。。

如果我们想保证在使用迭代器进行遍历时别的线程不来干扰，只能进行加锁操作了：

synchronized (vector) {
    Iterator<String> iterator = vector.iterator();

    while (iterator.hasNext()) {
        String s = iterator.next();
        System.out.println(s);
    }
}

但是如果在遍历过程中有什么复杂的操作可能会占用很长时间，那就意味着在遍历过程中并不能进行其他的容器操作，这种代价是非常大的。

如果我们在多线程环境下遍历某个Collection容器的次数远比修改这个容器的次数多，那么我们可以尝试使用java大叔们提供的CopyOnWriteArrayList或者CopyOnWriteArraySet，它采用了一种所谓的写入时复制的技术。以CopyOnWriteArrayList举例，普通的ArrayList内部是一个数组，每次添加元素其实都是往这个数组里添加，以下边这段代码为例：

List<Integer> list = new ArrayList<>();
list.add(0);
list.add(1);
list.add(2);

（为突出重点，忽略数组扩容的情况）它的添加过程其实是这样的：

但是CopyOnWriteArrayList在每一次添加元素时，都会复制一个新的底层数组，比如现在一个CopyOnWriteArrayList中已经有两个元素，添加第三个元素的过程如图所示（为突出重点，忽略数组扩容的情况）：

这就是所谓的写入时复制，这样每次复制不是都会浪费很多时间而且还会浪费很多内存空间么，这种List有什么优势么？

优势其实就是：使用迭代器遍历数组的时候是针对当前的底层数组进行遍历的，也就是说即使在遍历过程中有新的元素插入，它会被插入的新数组中，对遍历不会影响产生影响，也就不会抛出ConcurrentModificationException异常。当然调用get方法也是针对当前的底层数组调用的，如果在调用期间有别的线程写入，get方法时不能获取到最新值的。

可以看到，这种写入时复制的容器不能保证获取到最新写入的数据，插入元素的过程也及其损耗性能，所以它主要应用于在多线程环境下遍历容器的次数远比修改这个容器的次数多的情况。

ConcurrentLinkedQueue

这种队列内部使用CAS操作实现入队和出队操作，可以保证安全性，这里就唠叨它的源代码了(有兴趣自己看吧)，不知道你看的烦不烦，反正我是写的烦了～