并发容器

• 为什么需要并发容器

java提供了很多丰富的容器，主要可以分为四大类别List、Map、Set、Queue，但是这些容器的实现类有各种各样的缺点，如我们常见的ArrayList、HashMap是线程不安全的，如Hashtable、Vector是线程安全，但是效率很差，这些线程安全的容器是通过synchronized关键字来实现的，这样做就削弱了程序的并发性，所以并发容器就诞生了。

• 并发容器的定义

并发容器是专门针对多线程并发设计的，使用了锁分段技术，只对操作位置进行同步操作，但是其它其它线程可以对没有操作的位置继续进行访问。

1、ConcurrentHashMap

1.1数据结构

ConcurrentHashMap是java.util.concurrent(JUC)包下的成员，它支持线程安全、支持多线程操作时并发读写操作，在jdk1.7中，它采用的是Segment分段锁，但是在1.8中，使用的是CAS+synchronized，底层使用的是数组+链表+红黑树

• jdk1.7

jdk1.7中的ConcurrentHashMap是由Segment数组结构和HashEntry数组结构组成，Segment是一种可重入锁（ReentrantLock），每个Segment都包含一个HashEntry[]数组，每个HashEntry是一个链表结构，当对HashEntry数组的数据进行修改时，必须首先获得与它对应的Segment锁。ConcurrentHashMap使用锁分段技术，首先将数据分成一段一段的存储，然后给每一段数据分配一把锁，当一个线程占用锁访问其中一个段数据的时候，其它段的数据也能被其它线程访问。

• jdk1.8

jdk1.8抛弃了Segment分段锁，采用的是CAS+synchronized，它的改进是为了解决查询链表效率太低的问题，将1.7中的HashEntry改为node，取消的ReentrantLock改为了synchronized，采用的数据结构为数组+链表+红黑树，保证了查询效率。

1.2代码测试

为了验证ConcurrentHashMap效率如何，我们编写一个Demo来测试下，将它与Hashtable比较，Hashtable容器是使用synchronized来保证线程安全的。

对比代码测试

/*** @Author: Simon Lang* @Date: 2020/5/12 10:40*/
public class test_concurrentHashMap {public static void main(String[] args){//使用HashTable测试final Map<String,String> map=new Hashtable<String, String>();//使用concurrentHashMap测试
//        final Map<String,String> map=new ConcurrentHashMap<String, String>();Thread[] array=new Thread[1000];final Random r=new Random();//定义一个门栓final CountDownLatch latch=new CountDownLatch(array.length);long begin=System.currentTimeMillis();for(int i=0;i<array.length;i++){array[i]=new Thread(new Runnable() {@Overridepublic void run() {for(int j=0;j<10000;j++){map.put("key"+r.nextInt(1000000),"value"+r.nextInt(1000000));}latch.countDown();}});}for (Thread t:array) {t.start();}try {latch.await();} catch (InterruptedException e) {e.printStackTrace();}long end=System.currentTimeMillis();System.out.println("执行时间为："+(end-begin));}}

测试结果

类型消耗时间Hashtable8947msConcurrentHashMap7265ms

在多线程并发情况下，ConcurrentHashMap比HashTable效率高。

2、CopyOnWriteArrayList

CopyOnWriteArrayList是juc包下提供的一个并发安全的ArrayList，它的主要特点是写操作时需要复制集合，会创建一个新的底层数组，读操作时只需要获得数组下标即可，它的工作方式采用的是用空间换时间，适用于读多写少的场景中。

2.1源码分析

它的底层是数据结构是数组

private transient volatile Object[] array;

它仅在写/修改/删除操作时存在线程安全，所以在这些操作中加入了ReentrantLock来保证线程安全，以写操作为例

  public boolean add(E e) {//获取锁final ReentrantLock lock = this.lock;//加锁lock.lock();try {//获得当前集合中保存数据的数组Object[] elements = getArray();//获取该数组的长度int len = elements.length;//拷贝当前的数组，并让其长度+1Object[] newElements = Arrays.copyOf(elements, len + 1);//将加入的元素放在新数组的最后一位newElements[len] = e;//将数组的引用指向新数组setArray(newElements);return true;} finally {//释放锁lock.unlock();}}

图解

添加元素时先将原数组复制，然后再将添加的元素放置到新数组的末尾，然后将数组的引用指向新数组。

2.2、对比代码测试

Vector实现线程安全的方法采用的是synchronized关键字，即每个方法执行的时候都要去获取锁，性能会下降，而CopyOnWriteArrayList仅在写/修改/删除操作上加锁，读操作不加锁，我们编写一个Demo来测试下

测试代码

/*** @Author: Simon Lang* @Date: 2020/5/12 11:38*/
public class test_copyonWriterlist {public static void main(String[] args){//测试Vectot
//       final List list=new Vector();//测试CopyOnWriterArrayListfinal List<String> list=new CopyOnWriteArrayList<String>();Thread[] array=new Thread[100];final Random r=new Random();//定义一个门栓final CountDownLatch latch=new CountDownLatch(array.length);long begin=System.currentTimeMillis();for (int i=0;i<array.length;i++){array[i]=new Thread(new Runnable() {@Overridepublic void run() {for (int j=0;j<1000;j++){list.add("value"+r.nextInt(100000));}latch.countDown();}});}for (Thread t:array){t.start();}try {latch.await();} catch (InterruptedException e) {e.printStackTrace();}long end=System.currentTimeMillis();System.out.println("执行时间："+(end-begin));}
}

测试结果

类型消耗时间Vector122CopyOnWriteArrayList5771

测试结果表明，在写操作时，CopyOnWriterArrayList消耗的时间明显高于Vector，所以CopyOnWriterArrayList适用于读操作的场景中。

3、Queue

3.1、ConcurrentLinkedQueue

在并发编程中，有两种方法可以实现线程安全的队列，一种使用的是阻塞算法，该算法使用同一个锁（入队出队使用同一把锁）或中两把锁（入队和出队使用不同的锁）实现，另一种是使用非阻塞算法，该算法是使用循环CAS的方法来实现，我们这里主要使用非阻塞算法的方法来实现ConcurrentLinkedQueue。

3.1.1数据结构

ConcurrentLinkedQueue有head结点和tail结点组成，每个结点有由节点元素(item)和下一个结点的引用(next)组成，结点之间就是通过这个next关联起来，组成一个链表结构的队列

volatile E item;
volatile Node<E> next;

结构图

3.1.2入队分析

入队就是将入队结点添加的队列的尾部，我们以分别添加4个元素为例来讲解

入队主要分两个步骤：①将入队结点设置为当前队尾结点的下一个结点②更新tail结点，如果tail结点的next结点不为空，则将入队结点设置为tail结点，如果tail结点的next结点为空，则将入队结点设置为tail的next结点，所以tail结点并不总是尾结点。

入队源码分析

public boolean add(E e) {return offer(e);
}public boolean offer(E e) {// 如果e为null，则直接抛出NullPointerException异常checkNotNull(e);// 创建入队节点final Node<E> newNode = new Node<E>(e);// 循环CAS直到入队成功// 1、根据tail节点定位出尾节点（last node）；2、将新节点置为尾节点的下一个节点；3、casTail更新尾节点for (Node<E> t = tail, p = t;;) {// p用来表示队列的尾节点，初始情况下等于tail节点// q是p的next节点Node<E> q = p.next;// 判断p是不是尾节点，tail节点不一定是尾节点，判断是不是尾节点的依据是该节点的next是不是null// 如果p是尾节点if (q == null) {// 设置p节点的下一个节点为新节点，设置成功则casNext返回true；否则返回false，说明有其他线程更新过尾节点if (p.casNext(null, newNode)) {// Successful CAS is the linearization point// for e to become an element of this queue,// and for newNode to become "live".// 如果p != t，则将入队节点设置成tail节点，更新失败了也没关系，因为失败了表示有其他线程成功更新了tail节点if (p != t) // hop two nodes at a timecasTail(t, newNode);  // Failure is OK.return true;}// Lost CAS race to another thread; re-read next}// 多线程操作时候，由于poll时候会把旧的head变为自引用，然后将head的next设置为新的head// 所以这里需要重新找新的head，因为新的head后面的节点才是激活的节点else if (p == q)// We have fallen off list.  If tail is unchanged, it// will also be off-list, in which case we need to// jump to head, from which all live nodes are always// reachable.  Else the new tail is a better bet.p = (t != (t = tail)) ? t : head;// 寻找尾节点else// Check for tail updates after two hops.p = (p != t && t != (t = tail)) ? t : q;}
}

整个入队就做两件事情：①定位尾结点②使用CAS算法将入队结点设置为尾结点的next结点，不成功则重试

3.1.3出队分析

出队列就是从队列里返回一个节点元素，并清空该节点对元素的引用

并不是每次出队时都要更新head结点，当head结点有元素时，直接弹出head结点里的元素，而不会更新head结点。只有当head结点里没有元素时，出队操作才会更新head结点。

3.2、LinkedBlockingQueue

LinkedBlockingQueue是一种用链表实现的有界阻塞队列，队列的默认长度为Integer.MAX_VALUE，此队列是按照先进先出的原则对元素进行排序。

我们编写一个Demo，一个线程不断的向队列中添加元素，另外的几个线程不断的从队列中取出元素，是一种生产者--消费者模式。

测试代码

/*** @Author: Simon Lang* @Date: 2020/5/12 14:50*//*** put:队列满时阻塞* take：队列为0时阻塞*/
public class test_linkedBlockingqueue {final BlockingQueue<String> queue=new LinkedBlockingQueue<String>();final Random r=new Random();public static void main(String[] args){final test_linkedBlockingqueue t=new test_linkedBlockingqueue();new Thread(new Runnable() {@Overridepublic void run() {while (true){try {t.queue.put("value"+t.r.nextInt(1000));TimeUnit.SECONDS.sleep(1);}catch (InterruptedException e){e.printStackTrace();}}}},"producer").start();for (int i=0;i<3;i++){new Thread(new Runnable() {@Overridepublic void run() {while (true){try {System.out.println(Thread.currentThread().getName()+" - "+t.queue.take());} catch (InterruptedException e) {e.printStackTrace();}}}},"consumer"+i).start();}}
}

部分测试结果

3.3、ArrayBlockingQueue

ArrayBlockingQueue是一个用数组实现的有界阻塞队列，此队列按照FIFO的原则对元素进行排序。根据调用API（add/put/offer）不同，有不同的特性

我们编写一个Demo分别测试这几种方法

测试代码

/*** @Author: Simon Lang* @Date: 2020/5/12 15:08*/
public class test_arrayBlockingqueue {final BlockingQueue<String> queue=new ArrayBlockingQueue<String>(3);public static void main(String[] args){final test_arrayBlockingqueue t=new test_arrayBlockingqueue();for (int i=0;i<5;i++){//addMethod
//            System.out.println("add method: "+ t.queue.add("value"+i));
//            try {
//                t.queue.put("put"+i);
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//            }
//            System.out.println("put method : "+i);
//            System.out.println("offer method : " +t.queue.offer("value"+i));try {System.out.println("offer method :"+t.queue.offer("value"+i,1, TimeUnit.SECONDS));} catch (InterruptedException e) {e.printStackTrace();}}System.out.println(t.queue);}
}

• add方法在容量不足的时候，抛出异常

• put方法在容量不足的时候，阻塞等待

• offer方法

单参数offer方法，不阻塞，当容量不足的时候，返回false，放弃超出容量的数据

三参数offer方法offer(value,times,timeout)，容量不足的时候，阻塞times时长，如果在阻塞时间内，有容量空闲，新增数据返回true；如果阻塞时长范围内无容量空闲，放弃超出容量的数据，返回false。