1、并发容器概览

  • ConcurrentHashMap 线程安全的HashMap
  • CopyOnWriteArrayList 线程安全的List
  • BlockingQueue 阻塞队列接口，非常适合数据共享的通道
  • ConcurrentLinkedQueue 高效的非阻塞并发队列，可以看作线程安全的LinkedList
  • ConcurrentSkipListMap 使用跳表的数据结构进行快速查找的map

2、集合类的历史

2.1、Vector和HashTable

方法syncronied修饰，并发性能差

  1. Vector:
  3.     public synchronized boolean add(E e) {
  4.         modCount++;
  5.         ensureCapacityHelper(elementCount + 1);
  6.         elementData[elementCount++] = e;
  7.         return true;
  8.     }
  10. HashTable:
  11.      public synchronized V put(K key, V value) {
  12.         // Make sure the value is not null
  13.         if (value == null) {
  14.             throw new NullPointerException();
  15.         }
  17.         // Makes sure the key is not already in the hashtable.
  18.         Entry<?,?> tab[] = table;
  19.         int hash = key.hashCode();
  20.         int index = (hash & 0x7FFFFFFF) % tab.length;
  21.         @SuppressWarnings("unchecked")
  22.         Entry<K,V> entry = (Entry<K,V>)tab[index];
  23.         for(; entry != null ; entry = entry.next) {
  24.             if ((entry.hash == hash) && entry.key.equals(key)) {
  25.                 V old = entry.value;
  26.                 entry.value = value;
  27.                 return old;
  28.             }
  29.         }
  31.         addEntry(hash, key, value, index);
  32.         return null;
  33.     }

2.2、 Collections工具类同步HashMap和ArrayList

方法类代码块同步锁，并发性能差

  1.   List<String> syncList = Collections.synchronizedList(new ArrayList<String>());
  2.   Map<String,String> hashMap = Collections.synchronizedMap(new HashMap<String,String>());
  4. List:
  6.    public E get(int index) {
  7.             synchronized (mutex) {return list.get(index);}
  8.         }
  9.         public E set(int index, E element) {
  10.             synchronized (mutex) {return list.set(index, element);}
  11.         }
  12.         public void add(int index, E element) {
  13.             synchronized (mutex) {list.add(index, element);}
  14.         }
  15.         public E remove(int index) {
  16.             synchronized (mutex) {return list.remove(index);}
  17.         }
  19. Map:
  20.         public boolean containsKey(Object key) {
  21.             synchronized (mutex) {return m.containsKey(key);}
  22.         }
  23.         public boolean containsValue(Object value) {
  24.             synchronized (mutex) {return m.containsValue(value);}
  25.         }
  26.         public V get(Object key) {
  27.             synchronized (mutex) {return m.get(key);}
  28.         }
  30.         public V put(K key, V value) {
  31.             synchronized (mutex) {return m.put(key, value);}
  32.         }
  33.         public V remove(Object key) {
  34.             synchronized (mutex) {return m.remove(key);}
  35.         }
  36.         public void putAll(Map<? extends K, ? extends V> map) {
  37.             synchronized (mutex) {m.putAll(map);}
  38.         }
  39.         public void clear() {
  40.             synchronized (mutex) {m.clear();}
  41.         }

2.3、ConcurrentHashMap 和 CopyOnWriteArrayList

取代同步的ArrayList和HashMap

绝大多数情况下，ConcurrentHashMap 和 CopyOnWriteArrayList 性能优于同步的ArrayList和HashMap（如果一个List经常被修改，那么同步的ArrayList性能优于CopyOnWriteArrayList ，CopyOnWriteArrayList 适合读多写少的场景，每次写入都会复制完整的列表，比较消耗资源）

3、ConcurrentHashMap

3.1、Map简介

image.png

3.2、 为什么HashMap是线程不安全的？

  • 同时put碰撞导致数据丢失
  • 同时put扩容导致数据丢失
  • 死循环造成cpu100%(jdk7之前) 扩容的时候链表死循环

3.3、ConcurrentHashMap 结构

3.3.1、Jdk1.7中的ConcurrentHashMap 结构

image.png

jdk1.7中的ConcurrentHashMap 最外层是多个segment，每个segment的底层数据结构于HashMap类似，仍然是数组和链表组成的拉链法。
每个segment设置了独立的ReentrantLock锁，每个segment之间互不影响，提高了并发效率
ConcurrentHashMap 默认有16个segment,所以最多可以支持16个线程并发写（操作分布在不同的segment上。默认值可以在初始化的时候设置，但是一旦初始化后，无法扩容

3.3.2、Jdk1.8中的ConcurrentHashMap 结构

image.png

  • 第一种是最简单的，空着的位置代表当前还没有元素来填充。
  • 第二种就是和 HashMap 非常类似的拉链法结构，在每一个槽中会首先填入第一个节点，但是后续如果计算出相同的 Hash 值，就用链表的形式往后进行延伸。链表长度大于某一个阈值（默认为 8），且同时满足一定的容量要求的时候，ConcurrentHashMap 便会把这个链表从链表的形式转化为红黑树的形式，目的是进一步提高它的查找性能。
  • 第三种结构就是红黑树结构
  1. final V putVal(K key, V value, boolean onlyIfAbsent){
  2.         if (key == null || value == null) throw new NullPointerException();
  3.         int hash = spread(key.hashCode());
  4.         int binCount = 0;
  5.         for (Node<K,V>[] tab = table;;) {
  6.             Node<K,V> f; int n, i, fh;
  7.             if (tab == null || (n = tab.length) == 0)
  8.                 tab = initTable();
  9.             else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
  10.                 if (casTabAt(tab, i, null,
  11.                              new Node<K,V>(hash, key, value, null)))
  12.                     break;                   // no lock when adding to empty bin
  13.             }
  14.             else if ((fh = f.hash) == MOVED)
  15.                 tab = helpTransfer(tab, f);
  16.             else {
  17.                 V oldVal = null;
  18.                 synchronized (f) {
  19.                     if (tabAt(tab, i) == f) {
  20.                         if (fh >= 0) {
  21.                             binCount = 1;
  22.                             for (Node<K,V> e = f;; ++binCount) {
  23.                                 K ek;
  24.                                 if (e.hash == hash &&
  25.                                     ((ek = e.key) == key ||
  26.                                      (ek != null && key.equals(ek)))) {
  27.                                     oldVal = e.val;
  28.                                     if (!onlyIfAbsent)
  29.                                         e.val = value;
  30.                                     break;
  31.                                 }
  32.                                 Node<K,V> pred = e;
  33.                                 if ((e = e.next) == null) {
  34.                                     pred.next = new Node<K,V>(hash, key,
  35.                                                               value, null);
  36.                                     break;
  37.                                 }
  38.                             }
  39.                         }
  40.                         else if (f instanceof TreeBin) {
  41.                             Node<K,V> p;
  42.                             binCount = 2;
  43.                             if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
  44.                                                            value)) != null) {
  45.                                 oldVal = p.val;
  46.                                 if (!onlyIfAbsent)
  47.                                     p.val = value;
  48.                             }
  49.                         }
  50.                     }
  51.                 }
  52.                 if (binCount != 0) {
  53.                     if (binCount >= TREEIFY_THRESHOLD)
  54.                         treeifyBin(tab, i);
  55.                     if (oldVal != null)
  56.                         return oldVal;
  57.                     break;
  58.                 }
  59.             }
  60.         }
  61.         addCount(1L, binCount);
  62.         return null;
  63.     }
  1.   public V get(Object key) {
  2.         Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
  3.         int h = spread(key.hashCode());
  4.         if ((tab = table) != null && (n = tab.length) > 0 &&
  5.             (e = tabAt(tab, (n - 1) & h)) != null) {
  6.             if ((eh = e.hash) == h) {
  7.                 if ((ek = e.key) == key || (ek != null && key.equals(ek)))
  8.                     return e.val;
  9.             }
  10.             else if (eh < 0)
  11.                 return (p = e.find(h, key)) != null ? p.val : null;
  12.             while ((e = e.next) != null) {
  13.                 if (e.hash == h &&
  14.                     ((ek = e.key) == key || (ek != null && key.equals(ek))))
  15.                     return e.val;
  16.             }
  17.         }
  18.         return null;
  19.     }

3.3.3、为什么要把1.7的结构变为1.8的结构？

数据结构 ： Java 7 采用 Segment 分段锁来实现，而 Java 8 中的 ConcurrentHashMap 使用数组 + 链表 + 红黑树
hash碰撞： 先使用拉链法，再转红黑树
保证并发安全：ava 7 采用 Segment 分段锁，而 Java 8采用CAS
查询复杂度：链表查询复杂度为O(n),但是java8中一旦转为红黑树，就变为O(logn)

为什么超过8转红黑树？
默认链表，占用内存更少，红黑树占用
达到冲突8次概率只有千万分之一，转红黑树可以确保极端情况下的查询效率

3.3.4、错误操作ConcurrentHashMap 不保证线程安全

读和写两个独立操作组合到一起就不是线程安全的，可以采用cas实现

  1. public class ConcurrenthashMapDemo {
  3.     private static  ConcurrentHashMap<String,Integer> map = new ConcurrentHashMap();
  5.     public static void main(String[] args) throws InterruptedException {
  6.         int threadCount = 50;
  7.         long start = System.currentTimeMillis();
  8.         map.put("count", 0);
  9.         Runnable r = () -> {
  10.             IntStream.range(0, 100000).forEach(e -> {
  11.                 //可以实现但是并发效率低
  12.                 synchronized (ConcurrenthashMapDemo.class){
  13.                     Integer count = map.get("count");
  14.                     count ++;
  15.                     map.put("count",count);
  16.                 }
  17.             });
  18.         };
  21.         Thread[] threads = new Thread[threadCount];
  22.         IntStream.range(0, threadCount).forEach(e -> {
  23.             threads[e] = new Thread(r);
  24.         });
  25.         IntStream.range(0, threadCount).forEach(e -> {
  26.             threads[e].start();
  27.             try {
  28.                 threads[e].join();
  29.             } catch (InterruptedException interruptedException) {
  30.                 interruptedException.printStackTrace();
  31.             }
  32.         });
  34.         long end = System.currentTimeMillis();
  35.         System.out.println("syncnized方式 count结果：" + map.get("count") + " 耗时：" + (end - start));
  38.         long start2 = System.currentTimeMillis();
  39.         map.put("count", 0);
  40.         Runnable r2 = () -> {
  41.             IntStream.range(0, 100000).forEach(e -> {
  42.                 //并发效率高
  43.                 while(true){
  44.                     Integer count = map.get("count");
  45.                     Integer newCount = count + 1;
  46.                     boolean result = map.replace("count", count, newCount);
  47.                     if(result){
  48.                         break;
  49.                     }
  50.                 }
  52.             });
  53.         };
  54.         Thread[] otherThreads = new Thread[threadCount];
  55.         IntStream.range(0, threadCount).forEach(e -> {
  56.             otherThreads[e] = new Thread(r2);
  57.         });
  58.         IntStream.range(0, threadCount).forEach(e -> {
  59.             otherThreads[e].start();
  60.             try {
  61.                 otherThreads[e].join();
  62.             } catch (InterruptedException interruptedException) {
  63.                 interruptedException.printStackTrace();
  64.             }
  65.         });
  66.         long end2 = System.currentTimeMillis();
  67.         System.out.println("CAS方式 count结果：" + map.get("count") + " 耗时：" + (end2 - start2));
  68.     }
  69. }

4、CopyOnWriteArrayList

4.1、作用与适合场景

CopyOnWrite 的意思是说，当容器需要被修改的时候，不直接修改当前容器，而是先将当前容器进行 Copy，复制出一个新的容器，然后修改新的容器，完成修改之后，再将原容器的引用指向新的容器。
CopyOnWriteArrayList 的所有修改操作（add，set等）都是通过创建底层数组的新副本来实现的，所以 CopyOnWrite 容器也是一种读写分离的思想体现，读和写使用不同的容器。
作用

代替Vector和SynchronizedList，Vector和SynchronizedList锁粒度太大，并发效率比较低，并且迭代时无法编辑。

适合场景

读操作尽可能的块，而写操作慢一些也没有太大关系。

例如：黑名单

4.2、读写规则

读取完全不用加锁，写入不会阻塞读取操作，只有写入和写入之间需要同步等待。

4.3、实例

ArrayList迭代时无法编辑，CopyOnWriteArrayList迭代时可以修改，但是修改对迭代没有影响，还是会迭代修改之前的数据

  1. public class CopyOnWriteArrayListDemo {
  3.     public static void main(String[] args) {
  5.         //异常ConcurrentModificationException 不能再迭代时修改
  6. //        List<String> list = new ArrayList<>();
  8.         List<String> list = new CopyOnWriteArrayList<>();
  9.         IntStream.range(1,6).forEach(e -> list.add(e + ""));
  11.         Iterator<String> iterator = list.iterator();
  12.         while (iterator.hasNext()) {
  13.             String next = iterator.next();
  14.             if("2".equals(next)){
  15.                 list.remove("5");
  16.             }
  17.             if("3".equals(next)){
  18.                 list.add("6");
  19.             }
  20.             System.out.println("list: " + list);
  21.             System.out.println(next);
  22.         }
  23.     }
  24. }

ArrayList输出

  1. list: [1, 2, 3, 4, 5]
  2. 1
  3. list: [1, 2, 3, 4]
  4. 2
  5. Exception in thread "main" java.util.ConcurrentModificationException
  6.     at java.util.ArrayList$Itr.checkForComodification(ArrayList.java:909)
  7.     at java.util.ArrayList$Itr.next(ArrayList.java:859)
  8.     at com.imooc.thread_demo.collection.CopyOnWriteArrayListDemo.main(CopyOnWriteArrayListDemo.java:26)

CopyOnWriteArrayList输出

  1. list: [1, 2, 3, 4, 5]
  2. 1
  3. list: [1, 2, 3, 4]
  4. 2
  5. list: [1, 2, 3, 4, 6]
  6. 3
  7. list: [1, 2, 3, 4, 6]
  8. 4
  9. list: [1, 2, 3, 4, 6]
  10. 5

4.4、缺点

数据一致性问题：只能保证数据最终一致，不能保证数据实时一致。
内存占用问题：CopyOnWriteArrayList时创建副本实现写，内存占用大

4.5、源码分析

数据结构

  1.     /** The lock protecting all mutators */
  2.     final transient ReentrantLock lock = new ReentrantLock();
  4.     /** The array, accessed only via getArray/setArray. */
  5.     private transient volatile Object[] array;
  1.   public boolean add(E e) {
  2.         final ReentrantLock lock = this.lock;
  3.         lock.lock();
  4.         try {
  5.             Object[] elements = getArray();
  6.             int len = elements.length;
  7.             Object[] newElements = Arrays.copyOf(elements, len + 1);
  8.             newElements[len] = e;
  9.             setArray(newElements);
  10.             return true;
  11.         } finally {
  12.             lock.unlock();
  13.         }
  14.     }
  17.      public E get(int index) {
  18.         return get(getArray(), index);
  19.     }

5、并发队列

image.png

5.1、阻塞队列

从队列取数据，队列里无数据，则阻塞，往队列放数据，如果队列已满，那么就无法继续插入，则阻塞

5.1.1、阻塞队列的常用方法


  1. 抛出异常：add、remove、element
    • add 方法是往队列里添加一个元素，如果队列满了，就会抛出异常来提示队列已满
    • remove 方法的作用是删除元素，如果我们删除的队列是空的，那么 remove 方法就会抛出异常。
    • element 方法是返回队列的头部节点，但是并不删除，如果队列是空的，抛出异常。
  2. 返回结果但不抛出异常：offer、poll、peek
    • offer 方法用来插入一个元素，并用返回值来提示插入是否成功，如果添加成功会返回 true，如果队列满了，无法插入，返回false
    • 移除并返回队列的头节点,如果队列是空的，返回 null 作为提示
    • 返回队列的头元素但并不删除，如果队列里面是空的，它便会返回 null 作为提示

3.阻塞：put、take

  • put 队列已满，那么就无法继续插入，则阻塞
  • take 从队列取数据，队列里无数据，则阻塞


5.1.2、ArrayBlockingQueue

有界
指定容量
公平：可以指定是否需要保证公平，如果想保证公平，等待最长的线程会被优先处理，不过同时会带来一定的性能损耗

5.1.3、LinkedBlockingQueue

无界
容量 Integer.MAX_VALUE
内部结构：node，两把锁

5.1.4、PriorityBlockingQueue

支持优先级的无界阻塞队列

  1.   public static void main(String[] args) {
  2.         PriorityBlockingQueue priorityBlockingQueue = new PriorityBlockingQueue(10,
  3.                  Comparator.comparingInt(o -> ((Student)o).getAge()));
  4.         priorityBlockingQueue.add(new Student(13,"张三"));
  5.         priorityBlockingQueue.add(new Student(17,"李四"));
  6.         priorityBlockingQueue.add(new Student(12,"王五"));
  7.             Student take = null;
  8.             do{
  9.                 take = (Student)priorityBlockingQueue.poll();
  10.                 if(take != null){
  11.                     System.out.println(take.getName());
  12.                 }
  13.             }while(take != null);
  14.  }

5.1.5、SynchronousQueue

SynchronousQueue 最大的不同之处在于，它的容量为 0，所以没有一个地方来暂存元素，导致每次取数据都要先阻塞，直到有数据被放入；同理，每次放数据的时候也会阻塞，直到有消费者来取

5.1.6、DelayQueue

延迟队列，根据延迟时间排序

5.2、非阻塞并发队列


ConcurrentLinkedQueue

非阻塞队列 ConcurrentLinkedQueue 使用 CAS 非阻塞算法 + 不停重试，来实现线程安全，适合用在不需要阻塞功能，且并发不是特别剧烈的场景