并发容器梳理

记录一个问题先：

AndroidS Studio打包APK时出现问题：org.gradle.api.tasks.TaskExecutionException: Execution failed for task ':app:lintVitalRelease'.

什么意思？任务'：app：lintVitalRelease的执行失败。

关于lint是个什么东西看官网：https://developer.android.com/studio/write/lint?hl=zh-CN

或者：https://blog.csdn.net/u011240877/article/details/54141714#commentBox

总之，这是一个保证代码的可维护性与助力代码优化的一个工具，可以检查出代码当中不规范或者结构混乱等威胁代码质量的地方予以警告。

但是Lint的警告当中有的问题是不必要一定解决的，所以过度依赖这个工具，也可能造成代码编写时效率低下的问题。

所以在打包的过程当中，由于没有解决掉lint中警告的问题，没有办法通过编译进行打包。

所以可以在lint配置当中选择忽略它的警告，例如在app下的build.gradle下android闭包中添加：

lintOptions {
        checkReleaseBuilds false
        abortOnError false
    }

就能成功打包了。

并发容器

ConcurrentHashMap、CopyOnWriteArrayList、阻塞队列

---

 

并发容器概览

• ConcurrentHashMap：线程安全的HashMap

• CopyOnWriteArrayList：线程安全的List

• BlockingQueue：这是一个接口，表示阻塞队列，非常适合用于作为数据共享的通道。

• ConcurrentLinkedQueue：高效的非阻塞并发队列，使用链表实现。可以看做是一个线程安全的LinkedList。

• ConcurrentSkipListMap：是一个Map，使用跳表的数据结构进行快速查找。（不常用）

 

集合类的相关历史

• Vector和Hashtable

  这是早期JDK中线程安全的ArrayList和HashMap，并发性能差，所有方法都加上synchronized，那么意味着并发竞争大的时候性能不会太好。

• ArrayList和HashMap

  虽然这两个类不是线程安全的，但是可以用Collections.synchronizedList(new ArrayList<E>())和Collections.synchronizedMap(new HashMap<K，V>())使之变成安全的。这种方式也是根据传入的集合类的类型，比如是否RandomAccess类型，实现了RandomAccess接口，来返回一个对应的线程安全的SynchronizedRandomAccessList或者SynchronizedList，而在这些线程安全的list当中，使用synchronized方式并没有比Vector和Hashtable当中的方法高明到哪儿去，只是没有加在方法上而已。

• ConcurrentHashMap、CopyOnWriteArrayList

  这两种就到了比较不错的实现了，它们取代同步的HashMap和ArrayList。绝大多数并发情况下，ConcurrentHashMap和CopyOnWriteArrayList的性能都更好。除非是一个List经常被修改，那么用Collections.synchronizedList(new ArrayList<E>())会比使用CopyOnWriteArrayList性能更好，因为CopyOnWriteArrayList更适合读多写少的场景，每次写入都会完整复制整个链表，比较耗费资源。

 

 

---

ConcurrentHashMap

• Map

  Map是一个接口，有这些实现：

  • HashMap

  • Hashtable（性能太低，如果不需要并发就直接使用HashMap，而如果在并发场景下就使用ConcurrentHashMap）

  • LinkedHashMap：HashMap的一个子类，会保存键值对的插入顺序，在遍历的时候就有用了，顺序和插入顺序一致。

  • TreeMap：由于实现了SortedMap接口，所以也就具有排序的功能，也可以自定义排序的规则。所以遍历的时候也是排过序的。

  常用方法：

  int size();
  boolean isEmpty();
  boolean containsKey(Object key);
  boolean containsValue(Object value);
  V get(Object key);
  V put(K key, V value);
  V remove(Object key);
  Set<K> keySet();
  //等等

• 为什么需要ConcurrentHashMap

  为什么不用Collections.synchronizedMap(new HashMap<K,V>())?

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

  • 同时put碰撞导致数据丢失

  • 同时put扩容导致数据丢失（扩容之后的数组只有一个会被保存下来）

  • 死循环造成的CPU100%

    这个问题主要是出现JDK1.7及之前存在。

    核心原因就是在多线程同时扩容的时候会造成链表的死循环。但是这个问题吧，都说了HashMap不支持并发了，并发场景下使用自然会出现一些问题。

    CoolShell

• HashMap分析

  • JDK1.7拉链法

  • JDK1.8拉链法升级为红黑树

    关于红黑树：红黑树是对二叉查找树BST的一种平衡策略，O(logN)vsO(N)，会自动平衡，防止极端不平衡从而影响查找效率 的情况发生。红黑树的每个节点要么是红色，要么是黑色，但是根节点永远是黑色的；而且红色节点不能连续（也就是，红色节点的孩子和父亲都不能是红色）；从任一节点到其子树中每个叶子节点的路径都有相同数量的黑色结点；所有叶结点都是黑色的。

  • HashMap关于并发的特点：

    1. 非线程安全

    2. 迭代的时候不允许修改内容

    3. 只读的并发是安全的

    4. 如果一定要把HashMap用在并发环境下，

       用Collections.synchronizedMap(new HashMap<K,V>())。

• JDK1.7中的ConcurrentHashMap的实现分析

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

• JDK1.8中的ConcurrentHashMap实现和分析

  Java8中的ConcurrentHashMap是把代码完全的重写了，代码量也从一千多行涨到了六千多行。

  putVal流程：

  1. 判断key不为空

  2. 计算hash值

  3. 根据对应位置的节点类型，来赋值，或者helpTransfer，或者增长链表，或者给红黑树增加节点。

  4. 检查满足阈值就“红黑树化”

  5. 返回oldVal。

  get流程：

  1. 计算hash值

  2. 找到对应的位置，根据情况进行：

  3. 直接取值：

  4. 红黑树里取值

  5. 遍历链表取值

  6. 返回找到的结果

  图解ConcurrentHashMap

  源码分析：

  public class ConcurrentHashMap<K,V> extends AbstractMap<K,V>
      implements ConcurrentMap<K,V>, Serializable {
      private static final long serialVersionUID = 7249069246763182397L;
      
      //省略。。。
      
      //初始化，可见没有任何内容，真正的初始化要等到putval方法中调用initTable方法
      public ConcurrentHashMap() {
          
      }
      
      /**
       * Initializes table, using the size recorded in sizeCtl.
       * 使用sizeCtl中记录的大小初始化表。
       */
      //初始化 table，通过对 sizeCtl 的变量赋值来保证数组只能被初始化一次。
      private final Node<K, V>[] initTable() {
          Node<K, V>[] tab;
          int sc;
          //通过自旋保证初始化成功
          while ((tab = table) == null || tab.length == 0) {
              // 小于 0 代表有线程正在初始化，释放当前 CPU 的调度权，重新发起锁的竞争
              if ((sc = sizeCtl) < 0)
                  Thread.yield(); // lost initialization race; just spin
                  // CAS 赋值保证当前只有一个线程在初始化，-1 代表当前只有一个线程能初始化
                  // 保证了数组的初始化的安全性
              else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
                  try {
                      // 很有可能执行到这里的时候，table 已经不为空了，这里是双重 check
                      if ((tab = table) == null || tab.length == 0) {
                          // 进行初始化
                          int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
                          @SuppressWarnings("unchecked")
                          Node<K, V>[] nt = (Node<K, V>[]) new Node<?, ?>[n];
                          table = tab = nt;
                          sc = n - (n >>> 2);
                      }
                  } finally {
                      sizeCtl = sc;
                  }
                  break;
              }
          }
          return tab;
      }
      
      //新增
      public V put(K key, V value) {
          return putVal(key, value, false);
      }
      
      final V putVal(K key, V value, boolean onlyIfAbsent) {
          if (key == null || value == null) throw new NullPointerException();
          //计算hash
          int hash = spread(key.hashCode());
          int binCount = 0;
          for (Node<K, V>[] tab = table; ; ) {
              Node<K, V> f;
              int n, i, fh;
              //table是空的，进行初始化
              if (tab == null || (n = tab.length) == 0)
                  tab = initTable();//哈希表的初始化
                  //如果当前索引位置没有值，直接创建
              else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
                  //cas 在 i 位置创建新的元素，当 i 位置是空时，即能创建成功，结束for自循，
                  //否则继续自旋
                  if (casTabAt(tab, i, null,
                          new Node<K, V>(hash, key, value, null)))
                      break;                   // no lock when adding to empty bin
              }
              //如果当前槽点是转移节点，表示该槽点正在扩容，就会一直等待扩容完成
              //转移节点的 hash 值是固定的，都是 MOVED
              else if ((fh = f.hash) == MOVED)
                  tab = helpTransfer(tab, f);
                  //槽点上有值的
              else {
                  V oldVal = null;
                  //锁定当前槽点，其余线程不能操作，保证了安全
                  synchronized (f) {
                      //这里再次判断 i 索引位置的数据没有被修改
                      //binCount 被赋值的话，说明走到了修改表的过程里面
                      if (tabAt(tab, i) == f) {
                          //链表
                          if (fh >= 0) {
                              binCount = 1;
                              for (Node<K, V> e = f; ; ++binCount) {
                                  K ek;
                                  //值有的话，直接返回
                                  if (e.hash == hash &&
                                          ((ek = e.key) == key ||
                                                  (ek != null && key.equals(ek)))) {
                                      oldVal = e.val;
                                      if (!onlyIfAbsent)
                                          e.val = value;
                                      break;
                                  }
                                  Node<K, V> pred = e;
                                  //把新增的元素赋值到链表的最后，退出自旋
                                  if ((e = e.next) == null) {
                                      pred.next = new Node<K, V>(hash, key,
                                              value, null);
                                      break;
                                  }
                              }
                          }
                          //红黑树，这里没有使用 TreeNode,使用的是 TreeBin，
                          //TreeNode 只是红黑树的一个节点
                          //TreeBin 持有红黑树的引用，并且会对其加锁，保证其操作的线程安全
                          else if (f instanceof TreeBin) {
                              Node<K, V> p;
                              binCount = 2;
                              //满足if的话，把老的值给oldVal
                              //在putTreeVal方法里面，在给红黑树重新着色旋转的时候
                              //会锁住红黑树的根节点
                              if ((p = ((TreeBin<K, V>) f).putTreeVal(hash, key,
                                      value)) != null) {
                                  oldVal = p.val;
                                  if (!onlyIfAbsent)
                                      p.val = value;
                              }
                          }
                      }
                  }
                  //binCount不为空，并且 oldVal 有值的情况，说明已经新增成功了
                  if (binCount != 0) {
                      // 链表是否需要转化成红黑树
                      if (binCount >= TREEIFY_THRESHOLD)
                          treeifyBin(tab, i);
                      if (oldVal != null)
                          return oldVal;
                      //这一步几乎走不到。槽点已经上锁，只有在红黑树或者链表新增失败的时候
                      //才会走到这里，这两者新增都是自旋的，几乎不会失败
                      break;
                  }
              }
          }
          //check 容器是否需要扩容，如果需要去扩容，调用 transfer 方法去扩容
          //如果已经在扩容中了，check有无完成
          addCount(1L, binCount);
          return null;
      }
      /* put方法的大致思路：
      1.如果数组为空，初始化，初始化完成之后，走 2；
      2.计算当前槽点有没有值，没有值的话，cas 创建，失败继续自旋（for 死循环），直到成功，
      槽点有值的话，走 3；
      3.如果槽点是转移节点(正在扩容)，就会一直自旋等待扩容完成之后再新增，不是转移节点走 4；
      4.槽点有值的，先锁定当前槽点，保证其余线程不能操作，如果是链表，新增值到链表的尾部，
      如果是红黑树，使用红黑树新增的方法新增；
      5.新增完成之后 check 需不需要扩容，需要的话去扩容。*/
      
      
      private final void addCount(long x, int check) {
          CounterCell[] as; long b, s;
          if ((as = counterCells) != null ||
              !U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {
              CounterCell a; long v; int m;
              boolean uncontended = true;
              if (as == null || (m = as.length - 1) < 0 ||
                  (a = as[ThreadLocalRandom.getProbe() & m]) == null ||
                  !(uncontended =
                    U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) {
                  fullAddCount(x, uncontended);
                  return;
              }
              if (check <= 1)
                  return;
              s = sumCount();
          }
          if (check >= 0) {
              Node<K,V>[] tab, nt; int n, sc;
              while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&
                     (n = tab.length) < MAXIMUM_CAPACITY) {
                  int rs = resizeStamp(n);
                  if (sc < 0) {
                      if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
                          sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
                          transferIndex <= 0)
                          break;
                      if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
                          transfer(tab, nt);
                  }
                  else if (U.compareAndSwapInt(this, SIZECTL, sc,
                                               (rs << RESIZE_STAMP_SHIFT) + 2))
                      transfer(tab, null);
                  s = sumCount();
              }
          }
      }
      
      
      // 扩容主要分 2 步，第一新建新的空数组，第二移动拷贝每个元素到新数组中去
  // tab：原数组，nextTab：新数组
      private final void transfer(Node<K, V>[] tab, Node<K, V>[] nextTab) {
          // 老数组的长度
          int n = tab.length, stride;
          if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
              stride = MIN_TRANSFER_STRIDE; // subdivide range
          // 如果新数组为空，初始化，大小为原数组的两倍，n << 1
          if (nextTab == null) {            // initiating
              try {
                  @SuppressWarnings("unchecked")
                  Node<K, V>[] nt = (Node<K, V>[]) new Node<?, ?>[n << 1];
                  nextTab = nt;
              } catch (Throwable ex) {      // try to cope with OOME
                  sizeCtl = Integer.MAX_VALUE;
                  return;
              }
              nextTable = nextTab;
              transferIndex = n;
          }
          // 新数组的长度
          int nextn = nextTab.length;
          // 代表转移节点，如果原数组上是转移节点，说明该节点正在被扩容
          ForwardingNode<K, V> fwd = new ForwardingNode<K, V>(nextTab);
          boolean advance = true;
          boolean finishing = false; // to ensure sweep before committing nextTab
          // 无限自旋，i 的值会从原数组的最大值开始，慢慢递减到 0
          for (int i = 0, bound = 0; ; ) {
              Node<K, V> f;
              int fh;
              while (advance) {
                  int nextIndex, nextBound;
                  // 结束循环的标志
                  if (--i >= bound || finishing)
                      advance = false;
                      // 已经拷贝完成
                  else if ((nextIndex = transferIndex) <= 0) {
                      i = -1;
                      advance = false;
                  }
                  // 每次减少 i 的值
                  else if (U.compareAndSwapInt
                          (this, TRANSFERINDEX, nextIndex,
                                  nextBound = (nextIndex > stride ?
                                          nextIndex - stride : 0))) {
                      bound = nextBound;
                      i = nextIndex - 1;
                      advance = false;
                  }
              }
              // if 任意条件满足说明拷贝结束了
              if (i < 0 || i >= n || i + n >= nextn) {
                  int sc;
                  // 拷贝结束，直接赋值，因为每次拷贝完一个节点，都在原数组上放转移节点，所以拷贝完成的节点的数据一定不会再发生变化。
                  // 原数组发现是转移节点，是不会操作的，会一直等待转移节点消失之后在进行操作。
                  // 也就是说数组节点一旦被标记为转移节点，是不会再发生任何变动的，所以不会有任何线程安全的问题
                  // 所以此处直接赋值，没有任何问题。
                  if (finishing) {
                      nextTable = null;
                      table = nextTab;
                      sizeCtl = (n << 1) - (n >>> 1);
                      return;
                  }
                  if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
                      if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
                          return;
                      finishing = advance = true;
                      i = n; // recheck before commit
                  }
              } else if ((f = tabAt(tab, i)) == null)
                  advance = casTabAt(tab, i, null, fwd);
              else if ((fh = f.hash) == MOVED)
                  advance = true; // already processed
              else {
                  synchronized (f) {
                      // 进行节点的拷贝
                      if (tabAt(tab, i) == f) {
                          Node<K, V> ln, hn;
                          if (fh >= 0) {
                              int runBit = fh & n;
                              Node<K, V> lastRun = f;
                              for (Node<K, V> p = f.next; p != null; p = p.next) {
                                  int b = p.hash & n;
                                  if (b != runBit) {
                                      runBit = b;
                                      lastRun = p;
                                  }
                              }
                              if (runBit == 0) {
                                  ln = lastRun;
                                  hn = null;
                              } else {
                                  hn = lastRun;
                                  ln = null;
                              }
                              // 如果节点只有单个数据，直接拷贝，如果是链表，循环多次组成链表拷贝
                              for (Node<K, V> p = f; p != lastRun; p = p.next) {
                                  int ph = p.hash;
                                  K pk = p.key;
                                  V pv = p.val;
                                  if ((ph & n) == 0)
                                      ln = new Node<K, V>(ph, pk, pv, ln);
                                  else
                                      hn = new Node<K, V>(ph, pk, pv, hn);
                              }
                              // 在新数组位置上放置拷贝的值
                              setTabAt(nextTab, i, ln);
                              setTabAt(nextTab, i + n, hn);
                              // 在老数组位置上放上 ForwardingNode 节点
                              // put 时，发现是 ForwardingNode 节点，就不会再动这个节点的数据了
                              setTabAt(tab, i, fwd);
                              advance = true;
                          }
                          // 红黑树的拷贝
                          else if (f instanceof TreeBin) {
                              // 红黑树的拷贝工作，同 HashMap 的内容，代码忽略
                          …………
                              // 在老数组位置上放上 ForwardingNode 节点
                              setTabAt(tab, i, fwd);
                              advance = true;
                          }
                      }
                  }
              }
          }
      }
      /*ConcurrentHashMap 的扩容时机和 HashMap 相同，
      都是在 put 方法的最后一步检查是否需要扩容，
      如果需要则进行扩容，但两者扩容的过程完全不同，
      ConcurrentHashMap 扩容的方法叫做 transfer，
      从 put 方法的 addCount 方法进去，就能找到。
      
      1.首先需要把老数组的值全部拷贝到扩容之后的新数组上，先从数组的队尾开始拷贝；
      2.拷贝数组的槽点时，先把原数组槽点锁住，保证原数组槽点不能操作，
      成功拷贝到新数组时，把原数组槽点赋值为转移节点；
      3.这时如果有新数据正好需要 put 到此槽点时，发现槽点为转移节点，就会一直等待，
      所以在扩容完成之前，该槽点对应的数据是不会发生变化的；
      4.从数组的尾部拷贝到头部，每拷贝成功一次，就把原数组中的节点设置成转移节点；
      5.直到所有数组数据都拷贝到新数组时，直接把新数组整个赋值给数组容器，拷贝完成。*/
      
      
      //获取
      public V get(Object key) {
          Node<K, V>[] tab;
          Node<K, V> e, p;
          int n, eh;
          K ek;
          //计算hashcode
          int h = spread(key.hashCode());
          //不是空的数组 && 并且当前索引的槽点数据不是空的
          //否则该key对应的值不存在，返回null
          if ((tab = table) != null && (n = tab.length) > 0 &&
                  (e = tabAt(tab, (n - 1) & h)) != null) {
              //槽点第一个值和key相等，直接返回
              if ((eh = e.hash) == h) {
                  if ((ek = e.key) == key || (ek != null && key.equals(ek)))
                      return e.val;
              }
              //如果是红黑树或者转移节点，使用对应的find方法
              else if (eh < 0)
                  return (p = e.find(h, key)) != null ? p.val : null;
              //如果是链表，遍历查找
              while ((e = e.next) != null) {
                  if (e.hash == h &&
                          ((ek = e.key) == key || (ek != null && key.equals(ek))))
                      return e.val;
              }
          }
          return null;
      }
      /*ConcurrentHashMap 的获取，先获取数组的下标，
      然后通过判断数组下标的 key 是否和我们的 key 相等，相等的话直接返回，
      如果下标的槽点是链表或红黑树的话，分别调用相应的查找数据的方法，
      整体思路和 HashMap 相似。*/
  }

  • 数组初始化时的线程安全

    数组初始化时，首先通过自旋来保证一定可以初始化成功，然后通过 CAS 设置 SIZECTL 变量的值，来保证同一时刻只能有一个线程对数组进行初始化，CAS 成功之后，还会再次判断当前数组是否已经初始化完成，如果已经初始化完成，就不会再次初始化，通过自旋 + CAS + 双重 check 等手段保证了数组初始化时的线程安全。sizeCtl 默认值为0，当一个线程初始化数组时，会将 sizeCtl 改成 -1，由于被 volatile 修饰，对于其他线程来说这个变化是可见的，上面代码看到后续线程判断 sizeCtl 小于0 就会让出执行权。

  • 新增槽点值时的线程安全

    保证了各种情况下的新增（不考虑扩容的情况下）线程安全，通过自旋 + CAS + 锁。

    1. 通过自旋死循环保证一定可以新增成功。

       在新增之前，通过 for (Node[] tab = table;;) 这样的死循环来保证新增一定可以成功，一旦新增成功，就可以退出当前死循环，新增失败的话，会重复新增的步骤，直到新增成功为止。

    2. 当前槽点为空时，通过 CAS 新增。

       这里的写法非常严谨，没有在判断槽点为空的情况下直接赋值，因为在判断槽点为空和赋值的瞬间，很有可能槽点已经被其他线程赋值了，所以采用 CAS 算法，能够保证槽点为空的情况下赋值成功，如果恰好槽点已经被其他线程赋值，当前 CAS 操作失败，会再次执行 for 自旋，再走槽点有值的 put 流程，这里就是自旋 + CAS 的结合。

    3. 当前槽点有值，锁住当前槽点。

       put 时，如果当前槽点有值，就是 key 的 hash 冲突的情况，此时槽点上可能是链表或红黑树，我们通过synchronized (f)锁住槽点，来保证同一时刻只会有一个线程能对槽点进行修改。

    4. 红黑树旋转时，锁住红黑树的根节点，保证同一时刻，当前红黑树只能被一个线程旋转。

  • 扩容时的线程安全

    Java8 ConcunrrentHashMap 支持并发扩容，之前扩容总是由一个线程将旧数组中的键值对转移到新的数组中，支持并发的话，转移所需要的时间就可以缩短了，当然相应的并发处理控制逻辑也就更复杂了，扩容转移通过 transfer 方法完成。

    扩容方法通过在原数组上设置转移节点，put 时碰到转移节点时会等待扩容成功之后才能 put 的策略，来保证了整个扩容过程中肯定是线程安全的，因为数组的槽点一旦被设置成转移节点，在没有扩容完成之前，是无法进行操作的。

    1. 拷贝槽点时，会把原数组的槽点锁住；

    2. 拷贝成功之后，会把原数组的槽点设置成转移节点，这样如果有数据需要 put 到该节点时，发现该槽点是转移节点，会一直等待直到扩容成功，才能继续 put，可参考 put 方法中的 helpTransfer 方法；

    3. 从尾到头进行拷贝，拷贝成功就把原数组的槽点设置成转移节点。

    4. 等扩容拷贝都完成之后，直接把新数组的值赋值给数组容器，之前等待 put 的数据才能继续 put。

• 对于JDK1.7和1.8的缺点，为什么要把1.7的结构改成1.8的结构？

  • 数据结构

  • Hash碰撞

  • 保证并发安全

  • 查询复杂度

  • 为什么超过8要转为红黑树？

    红黑树的每个结点占用的空间是链表的两倍，空间损耗是要大些的比起链表，所以开始的时候是默认使用占用空间更少的链表，8的这个取值是通过泊松分布得出的，链表长度增加到8也就是冲突达到8次的时候概率已经相当小，所以一般情况下链表的长度不会达到8，如果遇到了这种达到8情况，只能说明hash算法出现了问题。所以为了保证这种极端情况下，ConcurrentHashMap仍然有较高的查询效率，所以就从这个情况开始转化为红黑树。

• 组合操作：ConcurrentHashMap也不是线程安全的？

  错误使用可能会造成线程不安全的情况，而不是说一个工具是线程安全的，使用起来就一定是线程安全的。

  所以ConcurrentHashMap提供了一些组合操作：

  replace方法把一个值根据当前的值决定是否进行替换。

  putIfAbsent如果当前key不存在值的时候才放入。

• 实际生产案例

 

---

CopyOnWriteArrayList

• 诞生历史和原因

  从JDK1.5就存在。是List当中的重要并发工具，代替Vector和SynchronizedList，就和ConcurrentHashMap代替SynchronizedMap的原因一样。Vector和SynchronizedList锁粒度太大，并发效率相对较低，并且迭代时无法编辑。除了CopyOnWriteArrayList以外，Copy-On-Write并发容器还包括了CopyOnWriteArraySet，用来替代同步Set。

• 适用场景

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

  读多写少：黑名单，每日更新；监听器，迭代操作远多于修改操作。

• 读写规则

  读写锁当中是读读共享，其他互斥（写写、读写、写读）。

  在CopyOnWriteArrayList中读写锁规则升级：读取完全不用加锁，并且写入也不会阻塞读取操作。只有写入和写入之间才需要进行同步等待。

  但是在迭代过程中进行修改以后，读过程对写过程是没有感知的，读出的数据是修改之前的数据。

• 实现原理

  CopyOnWrite创建新副本，读写分离。底层通过新建一个新的数组，修改的时候对原有数据进行一次复制，把修改的内容写入新的副本中，最后再替换回去，所以是一种读写分离的思想，读和写使用完全不同的容器。因为对数据的实时性要求不高的情况下使用。旧容器具有不可变的特性，完全可以并发读没有问题。迭代的时候读的是旧数据，迭代时修改不报错，只是说迭代的数据可能是过期的。

  public class ArrayList<E> extends AbstractList<E>
          implements List<E>, RandomAccess, Cloneable, java.io.Serializable
  {
      private class Itr implements Iterator<E> {
          //修改的次数在创建迭代器的时候就记录下来了。
          int expectedModCount = modCount;
          
          @SuppressWarnings("unchecked")
          public E next() {
              //这个方法里面会去检查修改的次数
              checkForComodification();
              int i = cursor;
              if (i >= size)
                  throw new NoSuchElementException();
              Object[] elementData = ArrayList.this.elementData;
              if (i >= elementData.length)
                  throw new ConcurrentModificationException();
              cursor = i + 1;
              return (E) elementData[lastRet = i];
          }
          
          final void checkForComodification() {
              //可以看到当修改的次数和记录的次数不一样的时候就会抛出一个异常
              if (modCount != expectedModCount)
                  throw new ConcurrentModificationException();
          }
      }
  }

• 缺点

  数据一致性问题：CopyOnWrite容器只能保证数据的最终一致性，不能保证数据的实时一致性。所以如果希望写入的数据马上能够读到，就不应该使用CopyOnWrite容器。

  内存占用问题：因为CopyOnWrite的写时复制机制，所以在进行写操作的时候，内存里会同时驻扎两个对象的内存。

• 源码分析

  数据结构：

  //JDK1.8源码
  public class CopyOnWriteArrayList<E>
      implements List<E>, RandomAccess, Cloneable, java.io.Serializable {
      /** The lock protecting all mutators */
      final transient ReentrantLock lock = new ReentrantLock();
      /** The array, accessed only via getArray/setArray. */
      private transient volatile Object[] array;
      
      /**
       * Creates an empty list.
       */
      public CopyOnWriteArrayList() {
          setArray(new Object[0]);
      }
      
      public boolean add(E e) {
          final ReentrantLock lock = this.lock;
          lock.lock();
          try {
              Object[] elements = getArray();
              int len = elements.length;
              //关键部分，copy数组
              Object[] newElements = Arrays.copyOf(elements, len + 1);
              newElements[len] = e;
              setArray(newElements);
              return true;
          } finally {
              lock.unlock();
          }
      }
      
      //没有看到get操作有进行过加锁
      public E get(int index) {
          return get(getArray(), index);
      }
      
      private E get(Object[] a, int index) {
          return (E) a[index];
      }
  }

   

并发队列Queue（阻塞队列与非阻塞队列）

• 为什么使用队列

  用队列可以在线程间传递数据：生产者消费者模式、银行转账。

  考虑锁等线程安全问题的重任从使用者转移到了队列上了。

• 并发队列

  Queue

  BlockingQueue

• 各个并发队列关系

• 阻塞队列BlockingQueue

  什么是阻塞队列：阻塞队列是具有阻塞功能的队列，所以首先它是一个队列，其次是具有阻塞功能。通常，阻塞队列的一端是给生产者放数据用，另一端给消费者拿数据用。阻塞队列是线程安全的，所以生产者和消费者都可以是多线程的。

  主要方法：最有特色的带有两个阻塞功能的方法是take()和put()。take()方法：获取并且移除队列的头结点，一旦如果执行take的时候，队列里无数据，就阻塞，直到队列里有数据。put()方法：插入元素。但是如果队列已满，那么就无法继续插入，则阻塞，直到队列有了空闲空间。

  • put、take（阻塞）

  • add、remove、element

  • offer、poll、peek

  是否有界（容量多大）：这是一个非常重要的属性，无界队列意味着里面可以容纳非常多（Integer.MAX_VALUE，约为2的31次，是一个非常大的数，可以近似认为是无限容量）

  阻塞队列和线程池的关系：阻塞队列是线程池的重要组成部分。

  • ArrayBlockingQueue

    有界，可以指定容量

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

    //Java10
    public class ArrayBlockingQueue<E> extends AbstractQueue<E>
            implements BlockingQueue<E>, java.io.Serializable {
        //省略
        public void put(E e) throws InterruptedException {
            Objects.requireNonNull(e);
            final ReentrantLock lock = this.lock;
            //意味着put方法在这期间是可以被中断的
            lock.lockInterruptibly();
            try {
                //如果当前队列满了，就会开始等待
                while (count == items.length)
                    notFull.await();
                //否则，入队
                enqueue(e);
            } finally {
                lock.unlock();
            }
        }
    }

  • LinkedBlockingQueue

    无界，容量Integer.MAX_VALUE

    内部结构：Node、两把锁。

    public class LinkedBlockingQueue<E> extends AbstractQueue<E>
            implements BlockingQueue<E>, java.io.Serializable {
        /**
         * Linked list node class.
         */
        static class Node<E> {
            E item;
            Node<E> next;
            Node(E x) { item = x; }
        }
        
        /** Lock held by take, poll, etc */
        private final ReentrantLock takeLock = new ReentrantLock();
        /** Lock held by put, offer, etc */
        private final ReentrantLock putLock = new ReentrantLock();
        //两把锁，都是ReentrantLock，就可以让take和put互不干扰
        
        public void put(E e) throws InterruptedException {
            //不允许放入空
            if (e == null) throw new NullPointerException();
            // Note: convention in all put/take/etc is to preset local var
            // holding count negative to indicate failure unless set.
            int c = -1;
            Node<E> node = new Node<E>(e);
            final ReentrantLock putLock = this.putLock;
            final AtomicInteger count = this.count;
            //上锁
            putLock.lockInterruptibly();
            try {
                /*
                 * Note that count is used in wait guard even though it is
                 * not protected by lock. This works because count can
                 * only decrease at this point (all other puts are shut
                 * out by lock), and we (or some other waiting put) are
                 * signalled if it ever changes from capacity. Similarly
                 * for all other uses of count in other wait guards.
                 */
                //满了，进入阻塞
                while (count.get() == capacity) {
                    notFull.await();
                }
                //没满，放入队列
                enqueue(node);
                c = count.getAndIncrement();
                //如果放入以后还没有满，唤醒一个线程
                if (c + 1 < capacity)
                    notFull.signal();
            } finally {
                //释放锁
                putLock.unlock();
            }
            if (c == 0)
                signalNotEmpty();
        }
    }

  • PriorityBlockingQueue

    支持优先级

    自然排序（而不是先进先出）（可以自己指定排序规则）

    无界队列（容量不够会进行扩容，所以put不会阻塞，take可能阻塞）

    PriorityQueue的线程安全版本（不能插入null值，并且插入的值一定是可以比较的）

  • SynchronousQueue

    容量为0，它内部是不存储的。

    需要注意的是，SynchronousQueue的容量不是1而是0，因为SynchronousQueue不需要去持有元素，它要做的就是直接传递，中间不做存储，所以效率很高。是一个极好的用来直接传递的并发数据结构。

    SynchronousQueue是线程池Executors.newCachedThreadPool()使用的阻塞队列。

    SynchronousQueue没有peek等函数，因为peek的含义是取出头结点，但是SynchronousQueue的容量是0，所以连头结点都没有，也就没有peek方法。同理也没有iterate方法。

  • DelayQueue

    无界延迟队列，根据延迟时间排序。元素必须实现Delayed接口，规定排序规则。

• 非阻塞队列

  并发包中的非阻塞队列只有ConcurrentLinkedQueue这一种，顾名思义就是使用链表作为数据结构，使用CAS非阻塞算法来实现线程安全（不具备阻塞功能），适合用在对性能要求较高的并发场景。用的相对较少些。

  //Java8
  class ConcurrentLinkedQueue{
      //省略。。。
      
      public boolean offer(E e) {
          checkNotNull(e);
          final Node<E> newNode = new Node<E>(e);
  ​
          for (Node<E> t = tail, p = t;;) {
              Node<E> q = p.next;
              if (q == null) {
                  // p is last node
                  //CAS操作
                  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".
                      if (p != t) // hop two nodes at a time
                          casTail(t, newNode);  // Failure is OK.
                      return true;
                  }
                  // Lost CAS race to another thread; re-read next
              }
              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;
          }
      }
      
      
      private static class Node<E> {
          //省略。。。
          boolean casNext(Node<E> cmp, Node<E> val) {
              return UNSAFE.compareAndSwapObject(this, nextOffset, cmp, val);
          }
      }
  }

• 如何选择适合的队列

  • 边界以及扩容

  • 空间

  • 吞吐量，LinkedBlockingQueue的锁粒度比ArrayBlockingQueue更小，两把锁。等等

     

各并发容器总结

juc包中提供的容器分为三类：Concurrent*、CopyOnWrite*、Blocking*。

Concurrent的特点是大部分通过CAS来实现并发，而CopyOnWrite则是通过复制一份原数据来实现的，Blocking通过AQS实现。