0%

线程安全的容器

概述

如果你要使用线程安全的集合的话, java.util.concurrent 包中提供了很多并发容器供你使用:

  1. 线程安全的Map:ConcurrentHashMap: 可以看作是线程安全的 HashMap
  2. 线程安全的ListCopyOnWriteArrayList:可以看作是线程安全的 List,在读多写少的场合性能非常好,远远好于 Vector.
  3. 线程安全的队列
    1. 非阻塞队列ConcurrentLinkedQueue:高效的并发队列,使用链表实现。可以看做一个线程安全的 LinkedList,这是一个非阻塞队列
    2. 阻塞队列BlockingQueue: 这是一个接口,直接继承自Queue这个接口,JDK 内部通过链表、数组等方式实现了这个接口。表示阻塞队列,非常适合用于作为数据共享的通道
      1. 关于阻塞队列与非阻塞队列的区别
        1. 大致的理解就是,当队列满了或为空时,进行添加或删除操作时,阻塞队列会阻塞操作(一般也会同时提供超时阻塞版本以及非阻塞版本,线程池中三种类型的方法都有使用)或者报异常,并使用线程锁来实现线程安全,适用于生产者与消费者的模型,非阻塞队列则不会阻塞,直接返回操作结果,并且使用CAS保证线程安全,性能比较好
        2. 参考此篇文章
      2. ArrayBlockingQueue
      3. LinkedBlockingQueue
      4. PriorityBlockingQueue
      5. SynchronousQueue
      6. DelayedWorkQueue
  4. 线程安全的跳表:ConcurrentSkipListMap :跳表的实现。这是一个Map,使用跳表的数据结构进行快速查找。
    1. 维护元素的顺序

CopyOnWriteArrayList

概述

  • 类似于ReentrantReadWriteLock,允许多线程读,但是与ReenrantReadWriteLock不同的是,后者是读读共享、写写互斥、读写互斥、写读互斥;而CopyOnWriteArrayList 类比相比于在读写锁的思想又更进一步。为了将读取的性能发挥到极致,CopyOnWriteArrayList 读取是完全不用加锁的:写入也不会阻塞读取操作。只有写入和写入之间需要进行同步等待

CopyOnWriteArrayList 实现

  • 如何实现写不影响读:从名字就能看出来,这个线程安全容器的所有可变操作(add,set 等等)都是通过创建底层数组的新副本来实现的。当 List 需要被修改的时候,并不修改原有内容,而是对原有数据进行一次复制,将修改的内容写入副本。写完之后,再将修改完的副本替换原来的数据,这样就可以保证写操作不会影响读操作了
  • 已知的一个缺陷就是遍历的时候因为使用的是快照版本的数组,如果并发的修改是不能在遍历中可见的
  • CopyOnWrite是一个计算机专业术语
    • CopyOnWrite 也就是说:在计算机,如果你想要对一块内存进行修改时,我们不在原有内存块中进行写操作,而是将内存拷贝一份,在新的内存中进行写操作,写完之后,就将指向原来内存指针指向新的内存,原来的内存就可以被回收掉了

源码分析

  • 读操作

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    private transient volatile Object[] array;
    public E get(int index) {
    return get(getArray(), index);
    }

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

    final Object[] getArray() {
    return array;
    }
    • 可以发现是无拘无束的读数组,其原因在于对于读操作来说,数组是完全不会变的,因为任何变动都只会发生在数组的副本上,除此之外,内部存储数据的数组是被volatile修饰的,不会有内存可见性的问题
  • 写操作

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    public boolean add(E e) {
    final ReentrantLock lock = this.lock;
    // 加锁
    lock.lock();
    try {
    // 获取旧数组
    Object[] elements = getArray();
    int len = elements.length;
    // 获取旧数组拷贝
    Object[] newElements = Arrays.copyOf(elements, len + 1);
    // 对新数组添加新的值
    newElements[len] = e;
    // 直接更改数组引用
    setArray(newElements);
    return true;
    } finally {
    // 释放锁
    lock.unlock();
    }
    }
    • CopyOnWriteArrayList内部使用了ReentrantLock(可重入锁)进行加锁
    • 写操作很粗暴就是以下几个步骤
      • 加锁
      • 拷贝旧数组
      • 在新数组中写入
      • 新数组替换旧数组
      • 解锁
    • 拷贝过程使用到了Arrays.copyOf函数,其内部使用的是System.arraycopy函数,ArrayList扩容时使用的也是这个方法

ConcurrentLinkedQueue

  • Java 提供的线程安全的 Queue 可以分为阻塞队列非阻塞队列,其中阻塞队列的典型例子是 BlockingQueue,非阻塞队列的典型例子是 ConcurrentLinkedQueue,在实际应用中要根据实际需要选用阻塞队列或者非阻塞队列。 阻塞队列可以通过加锁来实现,非阻塞队列可以通过 CAS 操作实现
  • 从名字可以看出,ConcurrentLinkedQueue这个队列使用链表作为其数据结构。ConcurrentLinkedQueue 应该算是在高并发环境中性能最好的队列了。它之所有能有很好的性能,是因为其内部复杂的实现
  • 入队和出队操作均利用CAS(compare and set)更新,这样允许多个线程并发执行,并且不会因为加锁而阻塞线程,使得并发性能更好

ConcurrentHashMap

  • 有专门的的一篇文章去分析其功能与源码

BlockingQueue

  • 在线程池的创建中,大量的使用了线程安全的阻塞队列

ArrayBlockingQueue

  • 使用数组实现的有界队列,一旦创建则容量不能更改

  • 使用ReentrantLock执行加锁操作

  • 并发控制使用可重入锁实现,不管是插入操作还是读取操作,都需要获取到锁才能进行操作。当队列容量满时,尝试将元素放入队列将导致操作阻塞;尝试从一个空队列中取一个元素也会同样阻塞

  • ArrayBlockingQueue 默认情况下不能保证线程访问队列的公平性,在构造函数中指定公平性为true可以保证线程访问队列的公平性(即支持公平锁与非公平锁

    所谓公平性是指严格按照线程等待的绝对时间顺序,即最先等待的线程能够最先获得锁访问到队列。而非公平性则是指访问队列的顺序不是遵守严格的时间顺序,有可能存在,当队列可以被访问时,长时间阻塞的线程依然无法访问到 ArrayBlockingQueue,而最近尝试加锁的线程竟可以获得锁。如果保证公平性,通常会降低吞吐量,但是非公平锁也容易导致大量的线程饥饿状态

    访问队列的公平性实际上就是锁(ReentrantLock)的公平性,其本质可在AQS的源码中看到

  • ArrayBlockingQueue在生产者放入数据和消费者获取数据,都是共用同一个锁对象,由此也意味着两者无法真正并行运行

LinkedBlockingQueue

  • 基于单向链表实现的队列,因为用链表实现,所以可以被当做是无界队列或者有界队列使用
  • 使用ReentrantLock执行加锁操作(仅支持默认的非公平锁,其目的应该就在于提高吞吐量)
  • 与 ArrayBlockingQueue 相比起来具有更高的吞吐量,为了防止 LinkedBlockingQueue 容量迅速增,损耗大量内存。通常在创建 LinkedBlockingQueue 对象时,会指定其大小,如果未指定,容量等于 Integer.MAX_VALUE,直接使用的话是比较危险的
  • ArrayBlockingQueue的区别就在于在生产者端和消费者端使用了独立的锁来进行控制,提高并发度
  • Executors工厂类的newFixedThreadPool方法与newSingleThreadExecutor方法使用的是此队列做任务队列

LinkedBlockingDeque

  • 链表实现的线程安全的双向阻塞队列
  • ForkJoinPool中使用的是双端队列,但是不清楚使用的是否是这个双端队列

PriorityBlockingQueue

  • 从名字就可以看出此队列是支持优先级的队列,所谓支持优先级就是能够对队列中的数据进行排序,默认情况下是自然排序,也可以使用成员的compareTo方法或者是在构造函数中传入比较器来指定排序规则
  • 真正意义的无界队列,只需要指定初始容量,后续容量不够时会自动执行扩容(内部存储结构是一个数组)
  • 使用ReentrantLock实现并发控制
  • 可以将本类理解为PriorityQueue的线程安全类,不能插入null值,插入队列的对象必须是可比较大小的(comparable),否则报 ClassCastException 异常。它的插入操作 put 方法不会 block,因为它是无界队列会自动扩容(take 方法在队列为空的时候会阻塞)
  • 不能保证同样优先级的对象的顺序

SynchronousQueue

  • 从名字可以看出,这是一个**同步队列**,也就是内部是没有任何数据结构来缓存队列的数据,也可以理解为是缓存空间为1的队列—–在某次元素添加后,必须等待其他线程取走该元素后才能继续添加
  • 在线程池创建的newCachedThreadPool中就是使用了此类任务队列
  • 基于CAS+LockSupport实现同步
    • 只是大致扫了一下源码,发现虽然有用到ReentrantLock,但是并不是在队列中使用的,而是在序列化时使用的

DelayQueue

  • 一个使用优先级队列(PriorityBlockingQueue)实现的无界阻塞队列,在创建元素时,可以指定多久才能从队列中获取当前元素,只有延时期限满足之后才能从队列中获取元素,但是定时线程池并不是使用的此队列

DelayedWorkQueue

  • 相对于DelayQueue当前使用这个延迟队列最多,因为定时任务线程池使用的就是这个队列
  • 提交任务与获取任务用的也是同一把锁,但是考虑到其内部要对执行时间做排序,使用同一把锁似乎也是应该的
  • 此队列的使用在线程池文章中有介绍

LinkedTransferQueue

  • 一个由链表结构组成的无界阻塞队列,相对于其他队列,添加了transfertryTransfer的方法,貌似并不常用,与LinkedBlockingQueue相比多一个transfer方法,即如果当前有消费者正等待接收元素,可以把生产者传入的元素立刻传输给消费者,而不是排队

ConcurrentSkipListMap

  • 与其类似的是ConcurrentHashMap,二者最明显的不同就在于前者的底层是跳表,而后者是数组+链表(红黑树)或者说是哈希表

跳表的概念

  • 跳表是一种可以用来快速查找的数据结构,有点类似于多路平衡树。它们都可以对元素进行快速的查找;对于有序数组的快速查找可以使用二分法,对于有序链表的查找可以使用跳表

  • 但一个重要的区别是:对平衡树的插入和删除往往很可能导致平衡树进行一次全局的调整。而对跳表的插入和删除只需要对整个数据结构的局部进行操作即可

    • 这样带来的好处是:在高并发的情况下,会需要一个全局锁来保证整个平衡树的线程安全。而对于跳表,只需要部分锁即可。这样,在高并发环境下,就可以拥有更好的性能
  • 可以说跳表的优势除了体现在相对于链表的查询性能好之外,也体现在相对于平衡树的并发性能好

  • 跳表的本质是同时维护了多个链表(典型的空间换时间),并且链表是分层的(相当于为链表加索引)

    • 最低层的链表维护了跳表内所有的元素,每上面一层链表都是下面一层的子集
    • 跳表内的所有链表的元素都是排序的。查找时,可以从顶级链表开始找。一旦发现被查找的元素大于当前链表中的取值,就会转入下一层链表继续找。这也就是说在查找过程中,搜索是跳跃式的
    • 很显然,跳表是使用空间换时间来降低时间复杂度的
    image-20210406101633762 image-20210406101647064
  • 与ConcurrentHashMap比较的优劣

    • 跳表的查询性能要次于哈希算法
    • 跳表的并发性能高于ConcurrentHashMap因为前者维持平衡,只需要对局部区域做改动,但是后者需要对整个树做平衡,所以需要一个全局的锁,这会降低并发性能
    • 跳表维护了元素的顺序,而哈希算法不会维持顺序
  • 跳表的时间复杂度与红黑树一样都是O(log n) 时间复杂度是O(n)

  • Redis的有序集合使用到了跳表

  • 关于跳表结构的其他操作,参考漫画:什么是跳表

参考