当前位置 : 主页 > 编程语言 > java >

Java并发JUC——并发容器

来源：互联网 收集：自由互联 发布时间：2023-02-04
引言 容器是Java基础类库中使用频率最高的一部分，Java集合包中提供了大量的容器类来帮组我们简化开发，我前面的文章中对Java集合包中的关键容器进行过一个系列的分析，但这些集合

引言

容器是Java基础类库中使用频率最高的一部分，Java集合包中提供了大量的容器类来帮组我们简化开发，我前面的文章中对Java集合包中的关键容器进行过一个系列的分析，但这些集合类都是非线程安全的，即在多线程的环境下，都需要其他额外的手段来保证数据的正确性，最简单的就是通过synchronized关键字将所有使用到非线程安全的容器代码全部同步执行。这种方式虽然可以达到线程安全的目的，但存在几个明显的问题：首先编码上存在一定的复杂性，相关的代码段都需要添加锁。其次这种一刀切的做法在高并发情况下性能并不理想，基本相当于串行执行。JDK1.5中为我们提供了一系列的并发容器，集中在java.util.concurrent包下，用来解决这两个问题.

 

并发容器是解决并发情况下的容器线程安全问题的。给多线程环境准备一个线程安全的容器对象。 线程安全的容器对象：Vector, Hashtable。线程安全容器对象，都是使用 synchronized 方法实现的。而concurrent 包中的同步容器，大多数是使用系统底层技术实现的线程安全。类似 native。 Java8 中使用 CAS。

容器概览

List简介

有序(存和取顺序一致)，有索引，可以存储重复，该有一个共性特点就是都可以操作角标。

Arraylist 内部是数组数据结构，是不同步的。替代了Vector。查询的速度快。

LinkedList 内部是链表数据结构，是不同步的。增删元素的速度很快，链表两端的元素也很快。

Vector 内部是数组数据结构，是同步的。增删，查询都很慢！

Set简介

无序(存和取顺序不一致)，无索引，不可以存储重复

HashSet 内部数据结构是哈希表 ，是不同步的。

LinkedHashSet Set集合里面比较独特的一个集合：具有HashSet的查询速度，且内部使用链表维护元素的存取顺序，所以存取有序

TreeSet 内部数据结构是二叉树，可以对Set集合中的元素进行排序。是不同步的。

Map简介

  • HashMap
  • Hashtable
  • LinkedHashMap
  • TreeMap
  • Properties Hashtable 内部结构是哈希表，是同步的。不允许null作为键，null作为值。

Properties 用来存储键值对型的配置文件的信息，可以和IO技术相结合

HashMap 内部结构是哈希表，不是同步的。允许null作为键，null作为值。

LinkedHashMap 用链表实现来扩展HashMap，可以实现存取有序

TreeMap 内部结构是二叉树，不是同步的。可以用继承Comparable类重写comparTo()或实现Comparator接口重写compare()来对Map集合中的键进行排序。要求Key是不可变对象。

并发容器概览

  • ConcurrentHashMap：线程安全的HashMap
  • CopyOnWriteArrayList：线程安全的List
  • BlockingQueue：这是一个接口，表示阻塞队列，非常适合用于作为数据共享的通道
  • ConcurrentLinkedQueue：高效的非阻塞并发队列，由链表实现，可以看成一个线程安全的LinkedList
  • ConcurrentSkipListMap：是一个Map，使用跳表的数据结构进行快速查找

古老和过时的同步容器

  • Vector和Hashtable：在方法上加上synchronized解决线程安全问题，并发性能低

HashMap和ArrayList

  • 虽然这两个类不是线程安全的，但可以用Collections.synchronizedList(new ArrayList<E>())和Collections.synchronizedMap(new HashMap<k,v>())使之变成线程安全的
  • 由源码可以得知，这种方式是在方法里面使用同步代码块的方式解决线程安全问题，本质上与Vector和Hashtable解决并发问题一样，并没有性能上的提升

ConcurrentHashMap和CopyOnWriteArrayList

  • 取代同步的HashMap和同步的ArrayList(时代的巨轮滚滚向前)
  • 绝大多数情况下，ConcurrentHashMap和CopyOnWriteArrayList的性能都更好

ConcurrentHashMap

为什么HashMap是线程不安全的

  • 同时put碰撞导致数据丢失
  • 同时put扩容导致数据丢失
  • 死循环造成的CPU100%
    • HashMap在高并发下的死循环(仅在JDK7以前存在)
    • 原因在于在多个线程同时扩容的时候，会造成链表的死循环，详解：https://coolshell.cn/articles/9606.html

HashMap特点

  • 非线程安全
  • 迭代时不允许修改内容
  • 只读的并发是安全的
  • 如果一定要把HashMap用在并发环境，那么使用Collections.synchronizedMap(new HashMap<k,v>())

JDK1.7ConcurrentHashMap

  • concurrenthashMap是采用一个叫做分段所的机制。
  • 它可以看作是一个二重hashMap，首先concurrentHashMap是一个segment数组，每个segment都是一个继承了ReentrantLock的类，这样就可以方便地在各个segment里面加锁所以每次需要加锁的操作锁住的是一个Segment，这样只要保证每个Segment是线程安全的，也就实现了全局的线程安全。
  • 这个最外面的Segment[]数组，是不可以扩容的，默认有16个Segments，所以最多支持16个线程并发写(操作分别分布在不同的Segment上)，这个默认值可以在初始化的时候设置为其他值，一旦初始化之后，是不可以更改的。
  • 然后进到Segment内部，会发现，每个Segment可以看作一个HashMap。也就是在一个Segment里面，有个HashEntry[]数组，然后这个数组是一个个桶，桶里面是单向链表。

JDK1.8ConcurrentHashMap

  • Java8 对 ConcurrentHashMap 进行了比较大的改动，可以参考 Java8 中 HashMap 相对于 Java7 HashMap 的改动，对于 ConcurrentHashMap，Java8 也引入了红黑树。
  • Java8 ConcurrentHashMap 源码真心不简单，最难的在于扩容，数据迁移操作不容易看懂。
  • 结构上和 Java8 的 HashMap 基本上一样，不过它要保证线程安全性，所以在源码上确实要复杂一些。

ConcurrentHashMap从JDK1.7该为JDK1.8

数据结构的改变带来并发处理能力的提升

  • 1.7中并发能力受限于Segment的个数，1.8中每个Node都是独立的

Hash碰撞

  • 1.8是直接采用数组+链表+红黑树来实现，时间复杂度在O(1)和O(n)之间，如果链表转化为红黑树了，那么就是O(1)到O(nlogn)。
  • 在这里值得一提的是，ConcurrentHashMap会判断tabAt(tab, i = (n - 1) & hash)是不是 null，是的话就直接采用CAS进行插入，而如果不为空的话，则是synchronized锁住当前Node的首节点，这是因为当该Node不为空的时候，证明了此时出现了Hash碰撞，就会涉及到链表的尾节点新增或者红黑树的节点新增以及红黑树的平衡，这些操作自然都是非原子性的。
  • 从而导致无法使用CAS，当Node的当前下标为null的时候，由于只是涉及数组的新增，所以用CAS即可。

保证并发安全

  • ConcurrentHashMap在JDK1.8的时候将put()方法中的分段锁Segment移除，转而采用一种CAS锁和synchronized来实现插入方法的线程安全。
  • 由于每一个Node的首节点都会被synchronized修饰，从而将一个元素的新增转化为一个原子操作，同时Node的value和next都是由volatile关键字进行修饰，从而可以保证可见性。

为什么ConcurrentHashMap的链表长度超过8会转为红黑树

  • 默认是链表结构而不是红黑树，因为链表所占用的空间更少
  • 链表长度想要达到8很困难，概率只有千万分之6，如果极端情况发生，那么可以保证在这种极端情况下的查询效率

ConcurrentHashMap不正确的使用方式会导致线程安全问题的发生

/** * @Description: ConcurrentHashMap组合操作，并不保证线程安全 */ public class ConcurrentHashMapDemo implements Runnable { public static ConcurrentHashMap<String,Integer> concurrentHashMap = new ConcurrentHashMap(); public static void main(String[] args) throws InterruptedException { concurrentHashMap.put("小明",0); Thread thread1 = new Thread(new ConcurrentHashMapDemo()); Thread thread2 = new Thread(new ConcurrentHashMapDemo()); thread1.start(); thread2.start(); thread1.join(); thread2.join(); System.out.println(concurrentHashMap); } @Override public void run() { for (int i = 0; i < 1000; i++) { Integer score = concurrentHashMap.get("小明"); int newScore = score + 1; concurrentHashMap.put("小明",newScore); } } }

针对上面的正确使用方法

public class ConcurrentHashMapDemo implements Runnable { public static ConcurrentHashMap<String,Integer> concurrentHashMap = new ConcurrentHashMap(); public static void main(String[] args) throws InterruptedException { concurrentHashMap.put("小明",0); Thread thread1 = new Thread(new ConcurrentHashMapDemo()); Thread thread2 = new Thread(new ConcurrentHashMapDemo()); thread1.start(); thread2.start(); thread1.join(); thread2.join(); System.out.println(concurrentHashMap); } @Override public void run() { for (int i = 0; i < 1000; i++) { while (true){ Integer score = concurrentHashMap.get("小明"); int newScore = score + 1; boolean replace = concurrentHashMap.replace("小明", score, newScore); if(replace){ break; } } } } }

public V putIfAbsent(K key, V value)

  • 传统的put方法，只要key存在，value值就会被覆盖,注意put方法返回的是put之前的值，如果无put之前的值返回null
  • putIfAbsent方法，只有在key不存在或者key为null的时候，value值才会被覆盖

CopyOnWriteArrayList

  • CopyOnWriteArrayList用于代替Vector和SynchronizedList，就和ConcurrentHashMap用于代替SynchronizedMap一样
  • Vector和SynchronizedList的锁的粒度太大，并发效率相对较低，并且迭代时无法编辑
  • Copy-On-Write并发容器还包括CopyOnWriteArraySet，用来替代同步Set

CopyOnWriteArrayList适用场景

  • 读操作可以尽可能的快，而写操作即使慢一些也没有太大关系
  • 读多写少：黑名单，每日更新；监听器：迭代操作远多于修改操作

CopyOnWriteArrayList读写规则

  • 回顾读写锁：读读共享，其他都互斥(写写互斥、读写互斥)
  • 读写锁规则的升级：读取是完全不用加锁的，并且更厉害的是，写入也不会阻塞读取操作，只有写入和写入之间需要进行同步等待

CopyOnWriteArrayList实现原理

写入时复制（CopyOnWrite）思想 写入时复制（CopyOnWrite，简称COW）思想是计算机程序设计领域中的一种优化策略。其核心思想是，如果有多个调用者（Callers）同时要求相同的资源（如内存或者是磁盘上的数据存储），他们会共同获取相同的指针指向相同的资源，直到某个调用者视图修改资源内容时，系统才会真正复制一份专用副本（private copy）给该调用者，而其他调用者所见到的最初的资源仍然保持不变。这过程对其他的调用者都是透明的（transparently）。此做法主要的优点是如果调用者没有修改资源，就不会有副本（private copy）被创建，因此多个调用者只是读取操作时可以共享同一份资源。

 

CopyOnWriteArrayList这是一个ArrayList的线程安全的变体，其原理大概可以通俗的理解为:初始化的时候只有一个容器，很常一段时间，这个容器数据、数量等没有发生变化的时候，大家(多个线程)，都是读取(假设这段时间里只发生读取的操作)同一个容器中的数据，所以这样大家读到的数据都是唯一、一致、安全的，但是后来有人往里面增加了一个数据，这个时候CopyOnWriteArrayList 底层实现添加的原理是先copy出一个容器(可以简称副本)，再往新的容器里添加这个新的数据，最后把新的容器的引用地址赋值给了之前那个旧的的容器地址，但是在添加这个数据的期间，其他线程如果要去读取数据，仍然是读取到旧的容器里的数据。

CopyOnWriteArrayList缺点

  • 数据一致性问题：CopyOnWrite容器只能保证数据的最终一致性，不能保证数据的实时一致性，所以如果你希望写入的数据能马上读取到，那么请不要使用CopyOnWrite容器
  • 内存占用问题：因为CopyOnWrite容器的写是复制机制，所以在进行写操作的时候，内存里会同时驻扎两个对象的内存

并发队列Queue

为什么要使用队列

  • 用队列可以在线程间传递数据：生产者消费者模式、银行转账
  • 考虑锁等线程安全问题的重任从“你”转移到了“队列上”

在并发队列上JDK提供了两套实现，

  • 一个是以ConcurrentLinkedQueue为代表的高性能队列非阻塞，
  • 一个是以BlockingQueue接口为代表的阻塞队列，无论哪种都继承自Queue。

阻塞队列与非阻塞队

阻塞队列与普通队列的区别在于： 阻塞队列： 当队列是空的时，从队列中获取元素的操作将会被阻塞，试图从空的阻塞队列中获取元素的线程将会被阻塞，直到其他的线程往空的队列插入新的元素； 当队列是满时，往队列里添加元素的操作会被阻塞。试图往已满的阻塞队列中添加新元素的线程同样也会被阻塞，直到其他的线程使队列重新变得空闲起来，如从队列中移除一个或者多个元素，或者完全清空队列.

BlockingQueue

阻塞队列（BlockingQueue）是一个支持两个附加操作的队列。这两个附加的操作是：

  • 在队列为空时，获取元素的线程会等待队列变为非空。
  • 当队列满时，存储元素的线程会等待队列可用。

在Java中，BlockingQueue的接口位于java.util.concurrent 包中(在Java5版本开始提供)，由上面介绍的阻塞队列的特性可知，阻塞队列是线程安全的。

  • 阻塞功能：最有特色的两个带有阻塞功能的方法是
  • take()：获取并移除队列的头结点，一旦如果执行take()的时候，队列里无数据则阻塞，直到队列里有数据
  • put()：插入元素，但是如果队列已满，那么就无法继续插入，则阻塞，直到队列里有了空闲空间
  • 是否有界(容量有多大)：这是一个非常重要的属性，无界队列意味着里面可以容纳非常多(Integer.MAX_VALUE)
  • 阻塞队列和线程池的关系：阻塞队列是线程池的重要组成部分

BlockingQueue主要方法

  • put()、take()：最重要的方法，并且会阻塞住
  • add()、remove()、element()：
    • add：在添加元素的时候，若超出了度列的长度会直接抛出异常
    • remove：若队列为空，抛出NoSuchElementException异常
  • offer()、poll()、peek()：
    • offer：在添加元素时，如果发现队列已满无法添加的话，会直接返回false。
    • poll：若队列为空，返回null。
ArrayBlockingQueue

ArrayBlockingQueue是一个有边界的阻塞队列，它的内部实现是一个数组。有边界的意思是它的容量是有限的，我们必须在其初始化的时候指定它的容量大小，容量大小一旦指定就不可改变。

  • 公平：还可以指定是否需要保证公平，如果想保证公平的话，那么等待了最长时间的线程会被优先处理，不过这会同时带来一定的性能损耗

ArrayBlockingQueue是以先进先出的方式存储数据，最新插入的对象是尾部，最新移出的对象是头部。

ArrayBlockingQueue<String> arrays = new ArrayBlockingQueue<String>(3); arrays.offer("张三"); arrays.offer("李四"); arrays.offer("王五"); arrays.offer("666", 3, TimeUnit.SECONDS); // 队列满了，阻塞3秒后向下执行 System.out.println(arrays.poll()); // 张三 System.out.println(arrays.poll()); // 李四 System.out.println(arrays.poll()); // 王五 System.out.println(arrays.poll(3, TimeUnit.SECONDS)); //队列为空，阻塞3秒后结束

LinkedBlockingQueue

LinkedBlockingQueue阻塞队列大小的配置是可选的，如果我们初始化时指定一个大小，它就是有边界的，如果不指定，它就是无边界的。说是无边界，其实是采用了默认大小为Integer.MAX_VALUE的容量 。它的内部实现是一个链表。

 

和ArrayBlockingQueue一样，LinkedBlockingQueue 也是以先进先出的方式存储数据，最新插入的对象是尾部，最新移出的对象是头部。下面是一个初始化和使LinkedBlockingQueue的例子：

PriorityBlockingQueue（有界，快满时自动扩容，看似无界）

  • PriorityBlockingQueue是一个没有边界的队列，它的排序规则和 java.util.PriorityQueue一样。需要注意，PriorityBlockingQueue中允许插入null对象。
  • 所有插入PriorityBlockingQueue的对象必须实现 java.lang.Comparable接口，队列优先级的排序规则就是按照我们对这个接口的实现来定义的。
  • 另外，我们可以从PriorityBlockingQueue获得一个迭代器Iterator，但这个迭代器并不保证按照优先级顺序进行迭代。

SynchronousQueue

  • SynchronousQueue没有容量，是无缓冲等待队列，是一个不存储元素的阻塞队列，会直接将任务交给消费者，必须等队列中的添加元素被消费后才能继续添加新的元素。
  • 拥有公平（FIFO）和非公平(LIFO)策略，非公平侧罗会导致一些数据永远无法被消费的情况？
  • 线程池使用SynchronousQueue阻塞队列一般要求maximumPoolSizes为无界，避免线程拒绝执行操作。
SynchronousQueue注意点
  • SynchronousQueue没有peek等函数，因为peek的含义是取出头结点，但是SynchronousQueue的容量为0，所以连头结点都没有，那么也就没有peek方法，同理也没有iterator相关方法
  • 是一个极好的用来直接传递的并发数据结构
  • SynchronousQueue是线程池Executors.newCacheThreadPool()使用的阻塞队列

DelayQueue

  • 延迟队列，根据延迟时间排序
  • 元素需要实现Delayed接口，规定排序规则

ConcurrentLinkedQueue

ConcurrentLinkedQueue : 是一个适用于高并发场景下的队列，通过无锁的方式，实现 了高并发状态下的高性能，通常ConcurrentLinkedQueue性能好于BlockingQueue.它 是一个基于链接节点的无界线程安全队列，使用CAS非阻塞算法来实现线程安全(不具备阻塞功能)。该队列的元素遵循先进先出的原则。头是最先 加入的，尾是最近加入的，该队列不允许null元素。

// 非阻塞式队列，无界队列 ConcurrentLinkedDeque q = new ConcurrentLinkedDeque(); q.offer("张三"); q.offer("李四"); q.offer("王五"); //从头获取元素,删除该元素 System.out.println(q.poll()); //从头获取元素,不刪除该元素 System.out.println(q.peek()); //获取总长度 System.out.println(q.size());

如何选择合适的队列

  • 边界
  • 空间
  • 吞吐量

队列汇总

  • ArrayDeque, （数组双端队列）
  • PriorityQueue, （优先级队列）
  • ConcurrentLinkedQueue, （基于链表的并发队列）
  • DelayQueue, （延期阻塞队列）（阻塞队列实现了BlockingQueue接口）
  • ArrayBlockingQueue, 常用（基于数组的并发阻塞队列）
  • LinkedBlockingQueue, 常用（基于链表的FIFO阻塞队列）
  • LinkedBlockingDeque, （基于链表的FIFO双端阻塞队列）
  • PriorityBlockingQueue,常用 （带优先级的无界阻塞队列，）
  • SynchronousQueue常用 （并发同步阻塞队列）

参考： https://www.cnblogs.com/wangshen31/p/10464488.html

http://itxm.cn/post/13192.html

https://blog.csdn.net/yb223731/article/details/89928207

https://www.cnblogs.com/myseries/p/10877420.html

https://www.cnblogs.com/codeobj/p/11667737.html

https://www.cnblogs.com/keiyoumi520/p/10800290.html

上一篇：Java并发JUC——并发流程控制
下一篇：没有了
网友评论
相关栏目
最近更新
热门文章