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

Posted 潜龙在渊

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了并发容器非阻塞队列的并发容器相关的知识,希望对你有一定的参考价值。

  本文将介绍除了阻塞队列外的并发容器: 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(九)

以上是关于并发容器非阻塞队列的并发容器的主要内容,如果未能解决你的问题,请参考以下文章

并发_006 容器 队列

Java并发容器--ConcurrentLinkedQueue

并发容器 - 各种队列

并发容器阻塞队列详细介绍

深入理解java:2.3.4. 并发编程concurrent包 之容器ConcurrentLinkedQueue

并发编程: 同步容器并发容器阻塞队列双端队列