并发容器非阻塞队列的并发容器

Posted 2020-10-26 潜龙在渊

  本文将介绍除了阻塞队列外的并发容器： ConcurrentHashMap、CopyOnWriteArrayList、CopyOnWriteArraySet、ConcurrentSkipListMap、ConcurrentSkipListSet、ConcurrentLinkedQueue；

1. CopyOnWriteArrayList

• 是 ArrayList 的线程安全的实现，同时也可用于代替 Vector 。底层实现是一个数组，其中所有可变操作（add、set 等等）都是通过对底层数组进行一次新的复制来实现的。这就是 “写时复制”。
• 可变操作一般需要很大的开销，但是当遍历操作的数量大大超过可变操作的数量时，这种方法可能比其他替代方法更 有效。在不能或不想进行同步遍历，但又需要从并发线程中排除冲突时，它也很有用。
• 迭代器在创建时，使用了数组状态的快照，此数组快照在迭代期间是不会改变的，因此也不会发生冲突，并且迭代器保证不会抛出 ConcurrentModificationException。
• 迭代器使用的是数组的快照，所以迭代器是无法反映列表的添加、移除或者更改。
• 在迭代器上进行的元素更改操作（remove、set 和 add）不受支持。这些方法将抛出 UnsupportedOperationException。

看一下add()方法的源码：add()方法加了锁，添加一个元素，就是将数组的元素复制到新的数组中（新数组的大小=旧数组大小+1），再把新元素放到新数组中。remove方法也是如此。还提供原子操作 addIfAbsent() 方法。

 public void add(int index, E element) {
        final ReentrantLock lock = this.lock;
        //加锁
        lock.lock();
        try {
            //获取当前的数组
            Object[] elements = getArray();
            int len = elements.length;
            if (index > len || index < 0)
                throw new IndexOutOfBoundsException("Index: "+index+ ", Size: "+len);
            
            Object[] newElements;
            int numMoved = len - index;
            if (numMoved == 0)
                newElements = Arrays.copyOf(elements, len + 1);//复制创建新数组
            else {
                newElements = new Object[len + 1];
                System.arraycopy(elements, 0, newElements, 0, index);
                System.arraycopy(elements, index, newElements, index + 1,
                                 numMoved);
            }
            newElements[index] = element;//添加新元素到新数组中去
            setArray(newElements);//将新数组设为 当前对象的底层数组
        } finally {
            lock.unlock();
        }
    }

2. CopyOnWriteArraySet

  CopyOnWriteArraySet 是 HashSet 线程安全的一个实现。CopyOnWriteArraySet 的实现是基于 CopyOnWriteArrayList，其内部维护着一个 CopyOnWriteArrayList。其特性可参考 CopyOnWriteArrayList。

private final CopyOnWriteArrayList<E> al;

/**
     * Creates an empty set.构造方法
     */
    public CopyOnWriteArraySet() {
        al = new CopyOnWriteArrayList<E>();
    }

    public int size() {
        return al.size();
    }

    public boolean isEmpty() {
        return al.isEmpty();
    }

    public boolean contains(Object o) {
        return al.contains(o);
    }

    public boolean add(E e) {
        return al.addIfAbsent(e);
    }
    //........

get()方法使用的也是数组的快照，没有加锁阻塞，这就意味着get()方法返回的值不是很精确。

 public E get(int index) {
        return get(getArray(), index);
    }

    @SuppressWarnings("unchecked")
    private E get(Object[] a, int index) {
        return (E) a[index];
    }

3. ConcurrentLinkedQueue

  一个基于链接节点的 无界线程安全队列 。此队列按照 FIFO（先进先出）原则对元素进行排序。队列的头部 是队列中时间最长的元素。队列的尾部 是队列中时间最短的元素。新的元素插入到队列的尾部，队列获取操作从队列头部获得元素。
  ConcurrentLinkedQueue 采用了非阻塞的CAS算法，在高并发的环境下，性能非常好。其源码分析可参考。

4. ConcurrentHashMap

  ConcurrentHashMap 是 HashMap 线程安全的实现，同时也用于 代替 HashTable。(此类可以通过程序完全与 Hashtable 进行互操作，这取决于其线程安全，而与其同步细节无关。)。不同于HashTable(一张hash表只用一把锁)，在ConcurrentHashMap中，会将hash表的数据分成若干段，每段维护一个锁，粒度更细，以达到高效的并发访问；

  ConcurrentHashMap 与 其他并发容器一样，在迭代的过程不需要加锁，迭代器具有弱一致性，迭代期间不会抛出ConcurrentModificationException异常，并非“立即失败”；所谓 弱一致性 ，就是返回的元素将反映迭代器创建时或创建后某一时刻的映射状态。同时，需要在整个Map上进行计算的方法，如 size()、isEmpty()，这些方法的语义被略微减弱，以反映并发的特性，换句话说，这些方法的值是一个估计值，并不是很精确。事实上，这些方法在并发环境下用处很小，因为在并发的情况下，它们的返回值总是在变化。如果需要强一致性，那么就得考虑加锁。同步容器类便是强一致性的

  由于 ConcurrentHashMap 不能被加锁来执行独占访问，因此无法通过加锁来创建新的原子操作。不过，ConcurrentHashMap 提供了以下几个原子操作（由其父接口 ConcurrentMap 提供），基本满足需求了：

//如果指定键已经不再与某个值相关联，则将它与给定值关联。
V putIfAbsent(K key, V value);

//只有目前将键的条目映射到给定值时，才移除该键的条目。
boolean remove(Object key, Object value);

//只有目前将键的条目映射到某一值时，才替换该键的条目。
V replace(K key, V value);

//只有目前将键的条目映射到给定值时，才替换该键的条目。
boolean replace(K key,V oldValue, V newValue);

5. ConcurrentSkipListMap

  ConcurrentSkipListMap 是 TreeMap 的线程安全的实现。与上面的并发容器一样，迭代器是是弱一致性，返回的元素将反映迭代器创建时或创建后某一时刻的映射状态。它们不 抛出 ConcurrentModificationException，可以并发处理其他操作。size()操作返回的值也不是精确的。此外，批量操作 putAll、equals 和 clear 并不保证能以原子方式 (atomically) 执行 。例如，与 putAll 操作一起并发操作的迭代器只能查看某些附加元素。

6. ConcurrentSkipListSet

   ConcurrentSkipListSet 是 TreeSet 的线程安全的实现。ConcurrentSkipListSet 是基于 ConcurrentSkipListMap 实现的，就是将所有Map的Key的值所对应的 value 值为Boolean.TRUE。其特性参考 ConcurrentSkipListMap

 private final ConcurrentNavigableMap<E,Object> m;

 public boolean add(E e) {
        return m.putIfAbsent(e, Boolean.TRUE) == null;
    }

7. 最后总结几点：

  并发容器的性能一般都要比同步容器的性能更高。同步容器的所有公开的方法都用 synchronized 加了锁，所以同一时间只能一个线程访问同步容器。并发容器类则是锁的粒度更小（多个线程就可以并发地访问方法里面的非临界区代码，提高效率），还有的采用了 非阻塞的算法CAS(如 ConcurrentLinkedQueue)，锁分段等技术。当然，这并不是说同步容器就没有用了，如希望程序的 HashSet 线程安全,可以采用 CopyOnWriteArraySet,但如果写操作多于读操作的话，那么就应该采用 Collections.synchronizedSet(Set)，这种情况下，同步容器的效率会更高。

  并发容器是“弱一致性”，因此在迭代器创建后，不能反映容器的增、删、改的情况，同时size、isEmpty等方法只能得到一个估值，不是很精确。“弱一致性” 虽然在一些地方做出牺牲（就是上面所说的），但也极大提高了并发容器的其他方面性能，特别是迭代，迭代是可以并发进行，不需要额外的同步，也不会抛出 ConcurrentModificationException。

同步容器是“强一致性”,所以同步容器的迭代操作都需要加锁来保证原子性操作。

对于除了迭代操作外的复合操作，并发容器中某些类提供了常用的复合操作的原子性方法（如：ConcurrentHashMap.putIfAbsent(K key, V value) ）。对于复合操作，要谨慎，特别是同步容器，记得加锁来保证原子性。

---

下面是源码分析的好文章，值得一看

1. 【JUC】JUC集合框架综述
2. 【JUC】JDK1.8源码分析之ConcurrentHashMap（一）
3. 【JUC】JDK1.8源码分析之ConcurrentSkipListMap（二）
4. 【JUC】JDK1.8源码分析之ArrayBlockingQueue（三）
5. 【JUC】JDK1.8源码分析之LinkedBlockingQueue（四）
6. 【JUC】JDK1.8源码分析之ConcurrentLinkedQueue（五）
7. 【JUC】JDK1.8源码分析之CopyOnWriteArrayList（六）
8. 【JUC】JDK1.8源码分析之CopyOnWriteArraySet（七）
9.【JUC】JDK1.8源码分析之ConcurrentSkipListSet（八）
10.【JUC】JDK1.8源码分析之SynchronousQueue（九）