当前位置:   article > 正文

Java并发成神系列(8)—线程安全的集合类基本使用和源码分析_java线程安全的集合类

java线程安全的集合类

目录

1.ConcurrentHashMap

1.1 ConcurrentHashMap使用

1.2 HashMap存在问题

1)HashMap原理

1.3 ConcurrentHashMap(JDK8)

1)重要属性和内部类

2)构造器

3)get()方法

4)put()方法

5)size()方法

4.ConcurrenHashMap (JDK7)

 2.LinkedBlockingQueue(阻塞队列)

1.基本的入队出队

1)入队

 2)出队

2.加锁分析

3.put()和take()方法

1)put操作

2)take操作

与ArrayBlockingQueue的性能比较

3.ConcurrentLinkedQueue

4.CopyOnWriteArrayList


 

线程安全的集合类(三大类):

1)遗留的线程安全集合HashtableVector

2)使用Collections装饰的线程安全集合:

  • Collections.synchronizedCollection

  • Collections.synchronizedList

  • Collections.synchronizedMap

  • Collections.synchronizedSet

  • Collections.synchronizedNavigableMap

  • Collections.synchronizedNavigableSet

  • Collections.synchronizedSortedMap

  • Collections.synchronizedSortedSet

上面的这些集合都是讲非线程安全的集合包装后,在调用方法时包裹了一层synchronized代码块;

3)java.util.concurrent包下的线程安全集合类:(重点)

主要包含三个关键词:Blocking、CopyOnWrite、Concurrent

Blocking:这类集合大部分是基于锁实现的

CopyOnWrite:这类集合容器修改开销相对严重

Concurrent

  • 优点: 提高性能;因为其内部很多操作使用 cas 优化,一般可以提供较高吞吐量

  • 缺点

    具有一定的弱一致性:遍历时弱一致性;求大小弱一致性;读取弱一致性;(也就是遍历、求大小、读数据读到的数据可能是旧的!)

1.ConcurrentHashMap

1.1 ConcurrentHashMap使用

它的API和我们普通的HashMap都是一样的。对–>HashMap;

但是,我们要注意的是线程安全的集合类只是代表它的每个方法是线程安全的,方法之间是无法保证的;

比如:错误思路!

  1. demo(
  2. // 创建 map 集合
  3. // 虽然ConcurrentHashMap是线程安全的,但是当多个线程来临时,方法之间无法保证
  4. () -> new ConcurrentHashMap<String, Integer>(),
  5. // 进行计数
  6. (map, words) -> { //这里的get()方法和put()方法之间不是线程安全的!!
  7. for (String word : words) {
  8. Integer counter = map.get(word);
  9. int newValue = counter == null ? 1 : counter + 1;
  10. map.put(word, newValue);
  11. }
  12. }
  13. );

解决办法:

computeIfAbsent,这个方法包含了上述的三个操作,而更重要的是它的加锁只是加载了集合的一个链表上,并发性还是很高的;

  1. demo(
  2. () -> new ConcurrentHashMap<String, LongAdder>(),
  3. (map, words) -> {
  4. for (String word : words) {
  5. // 注意不能使用 putIfAbsent,此方法返回的是上一次的 value,首次调用返回 null
  6. // a 0,同一个key就是同一个累加器
  7. map.computeIfAbsent(word, (key) -> new LongAdder()).increment();
  8. }
  9. }
  10. );

1.2 HashMap存在问题

1)HashMap原理

HashMap数据结构:数组+链表;

首先,维护一个数组;先计算Key对应的哈希码,然后取模(数组个数),然后就把key放到对应的数组元素那;当有哈希冲突时,就需要构建一个链表(拉链法);

注意:JDK8中,后加入的元素总会加入链表的头部;JDK7中后加入的元素总会放入链表的头部;

扩容机制:

当数组元素超过阈值(3/4)时,就会扩容;(长度翻倍)

HashMap在JDK7环境下存在并发死链的问题:

在JDK7下,当多线程并发扩容时就容易产生并发死链问题;

比如:

一个链表1->35->16->null;扩容后会变为35->1->null和16->null;

当T0拿到1->35和35->16信息后准备迁移元素时,它时间片到了;T1来了,T1完整的完成了迁移操作,这时链表已经变为了35->1->null和16->null;当T0再去迁移元素时就会发生1->35->1的死链;

  1. 原始链表,格式:[下标] (key,next)
  2. [1] (1,35)->(35,16)->(16,null)
  3. 线程 a 执行到 1 处 ,此时局部变量 e 为 (1,35),而局部变量 next 为 (35,16) 线程 a 挂起
  4. 线程 b 开始执行
  5. 第一次循环
  6. [1] (1,null)
  7. 第二次循环
  8. [1] (35,1)->(1,null)
  9. 第三次循环
  10. [1] (35,1)->(1,null)
  11. [17] (16,null)
  12. 切换回线程 a,此时局部变量 e 和 next 被恢复,引用没变但内容变了:e 的内容被改为 (1,null),而 next 的内
  13. 容被改为 (35,1) 并链向 (1,null)
  14. 第一次循环
  15. [1] (1,null)
  16. 第二次循环,注意这时 e 是 (35,1) 并链向 (1,null) 所以 next 又是 (1,null)
  17. [1] (35,1)->(1,null)
  18. 第三次循环,e 是 (1,null),而 next 是 null,但 e 被放入链表头,这样 e.next 变成了 352 处)
  19. [1] (1,35)->(35,1)->(1,35)
  20. 已经是死链了

HashMap产生问题的源码:

  1. // 将 table 迁移至 newTable
  2. void transfer(Entry[] newTable, boolean rehash) {
  3. int newCapacity = newTable.length;
  4. for (Entry<K,V> e : table) {
  5. while(null != e) {
  6. Entry<K,V> next = e.next;
  7. // 1 处
  8. if (rehash) {
  9. e.hash = null == e.key ? 0 : hash(e.key);
  10. }
  11. int i = indexFor(e.hash, newCapacity);
  12. // 2 处
  13. // 将新元素加入 newTable[i], 原 newTable[i] 作为新元素的 next
  14. e.next = newTable[i];
  15. newTable[i] = e;
  16. e = next;
  17. }
  18. }
  19. }

总结:

1)问题的原因就是在多线程环境下使用了非线程安全的map集合;(JDK7)

2)JDK 8 虽然将扩容算法做了调整,不再将元素加入链表头(而是保持与扩容前一样的顺序),但是它会出现其它比如扩容丢数据的问题;

1.3 ConcurrentHashMap(JDK8)

1)重要属性和内部类

属性:

  1. // 默认为 0
  2. // 当初始化时, 为 -1
  3. // 当扩容时, 为 -(1 + 扩容线程数)
  4. // 当初始化或扩容完成后,为 下一次的扩容的阈值大小
  5. private transient volatile int sizeCtl;
  6. // 整个 ConcurrentHashMap 就是一个 Node[]
  7. static class Node<K,V> implements Map.Entry<K,V> {}
  8. // hash 表
  9. transient volatile Node<K,V>[] table;
  10. // 扩容时的 新 hash 表
  11. private transient volatile Node<K,V>[] nextTable;
  12. // 扩容时如果某个 bin 迁移完毕, 用 ForwardingNode 作为旧 table bin 的头结点
  13. static final class ForwardingNode<K,V> extends Node<K,V> {}
  14. // 用在 compute 以及 computeIfAbsent 时, 用来占位, 计算完成后替换为普通 Node
  15. static final class ReservationNode<K,V> extends Node<K,V> {}
  16. // 作为 treebin 的头节点, 存储 root 和 first
  17. static final class TreeBin<K,V> extends Node<K,V> {}
  18. // 作为 treebin 的节点, 存储 parent, left, right
  19. static final class TreeNode<K,V> extends Node<K,V> {}

其中,要注意的是:

ForwardingNode:扩容时,处理过的节点加上fnode(代表被处理过,其它线程来读需要读新的table)

TreeBin:当数组元素扩容到64之后,并且单链表个数超过8个之后,会将链表转换为红黑树(来提升查询效率),这时候当红黑树的节点又小于6之后,又会退化为链表;

重要方法

  1. // 获取 Node[] 中第 i 个 Node,链表头中的node
  2. static final <K,V> Node<K,V> tabAt(Node<K,V>[] tab, int i)
  3. // cas 修改 Node[] 中第 i 个 Node 的值, c 为旧值, v 为新值
  4. static final <K,V> boolean casTabAt(Node<K,V>[] tab, int i, Node<K,V> c, Node<K,V> v)
  5. // 直接修改 Node[] 中第 i 个 Node 的值, v 为新值
  6. static final <K,V> void setTabAt(Node<K,V>[] tab, int i, Node<K,V> v)

2)构造器

  1. // 初始容量 负载因子 并发度
  2. public ConcurrentHashMap(int initialCapacity, float loadFactor, int concurrencyLevel) {
  3. if (!(loadFactor > 0.0f) || initialCapacity < 0 || concurrencyLevel <= 0)
  4. throw new IllegalArgumentException();
  5. if (initialCapacity < concurrencyLevel) // Use at least as many bins
  6. initialCapacity = concurrencyLevel; // as estimated threads,当初始容量小于并发度时,将其改为并发度
  7. long size = (long)(1.0 + (long)initialCapacity / loadFactor);
  8. // tableSizeFor 仍然是保证计算的大小是 2^n, 即 16,32,64 ...
  9. int cap = (size >= (long)MAXIMUM_CAPACITY) ?
  10. MAXIMUM_CAPACITY : tableSizeFor((int)size);
  11. this.sizeCtl = cap; //真正大小
  12. }

它实现了懒惰初始化;在构造方法中仅仅计算了table的大小;以后在第一次使用时,才会真正创建;

3)get()方法

  1. public V get(Object key) {
  2. Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
  3. // spread 方法能确保返回结果是正数
  4. int h = spread(key.hashCode()); //h真正的哈希码
  5. if ((tab = table) != null && (n = tab.length) > 0 &&
  6. (e = tabAt(tab, (n - 1) & h)) != null) {
  7. // 如果头结点已经是要查找的 key
  8. if ((eh = e.hash) == h) {
  9. if ((ek = e.key) == key || (ek != null && key.equals(ek)))
  10. return e.val;
  11. }
  12. // hash 为负数表示该 bin 在扩容中或是 treebin, 这时调用 find 方法来查找
  13. else if (eh < 0)
  14. return (p = e.find(h, key)) != null ? p.val : null;
  15. // 正常遍历链表, 用 equals 比较
  16. while ((e = e.next) != null) {
  17. if (e.hash == h &&
  18. ((ek = e.key) == key || (ek != null && key.equals(ek))))
  19. return e.val;
  20. }
  21. }
  22. return null;
  23. }

get()流程:

  • 如果table不为空且长度大于0且索引位置有元素

    • if 链表头节点key的hash值和我们要找的key的相等

      • if头节点的key指向同一个地址或者equals我们要找的key

        • 返回value

    • else if 头节点的hash为负数(bin在扩容或者是treebin)

      • 调用find方法查找

    • 然后再遍历链表(且e的下一个节点不为空):

      • 节点key的hash值相等,且key指向同一个地址或equals

        • 返回value

  • 返回null

4)put()方法

table:数组;bin:链表;

  1. public V put(K key, V value) {
  2. return putVal(key, value, false);
  3. }
  4. final V putVal(K key, V value, boolean onlyIfAbsent) {
  5. //不允许有空的键值;普通的HashMap是允许的;
  6. if (key == null || value == null) throw new NullPointerException();
  7. // 其中 spread 方法会综合高位低位, 具有更好的 hash 性
  8. int hash = spread(key.hashCode()); //保证hash码为正整数(负的有其他用途)
  9. int binCount = 0;
  10. for (Node<K,V>[] tab = table;;) {
  11. // f 是链表头节点
  12. // fh 是链表头结点的 hash
  13. // i 是链表在 table 中的下标
  14. Node<K,V> f; int n, i, fh;
  15. // 要创建 table
  16. if (tab == null || (n = tab.length) == 0)
  17. // 初始化 table 使用了 cas, 无需 synchronized 创建成功, 进入下一轮循环
  18. tab = initTable();
  19. // 要创建链表头节点
  20. else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
  21. // 添加链表头使用了 cas, 无需 synchronized;不成功的话再进行循环
  22. if (casTabAt(tab, i, null,
  23. new Node<K,V>(hash, key, value, null)))
  24. break;
  25. }
  26. // 帮忙扩容;发现其它线程在扩容,就会帮忙(锁住某一个链表即可,以链表为单位扩容)
  27. else if ((fh = f.hash) == MOVED)
  28. // 帮忙之后, 进入下一轮循环
  29. tab = helpTransfer(tab, f);
  30. else {
  31. V oldVal = null;
  32. // 锁住链表头节点(巧妙,只锁住一个链表就OK了)
  33. synchronized (f) {
  34. // 再次确认链表头节点没有被移动
  35. if (tabAt(tab, i) == f) {
  36. // 链表
  37. if (fh >= 0) { //>0代表是普通节点
  38. binCount = 1;
  39. // 遍历链表
  40. for (Node<K,V> e = f;; ++binCount) {
  41. K ek;
  42. // 找到相同的 key
  43. if (e.hash == hash &&
  44. ((ek = e.key) == key ||
  45. (ek != null && key.equals(ek)))) {
  46. oldVal = e.val;
  47. // 更新
  48. if (!onlyIfAbsent)
  49. e.val = value;
  50. break;
  51. }
  52. Node<K,V> pred = e;
  53. // 已经是最后的节点了, 新增 Node, 追加至链表尾
  54. if ((e = e.next) == null) {
  55. pred.next = new Node<K,V>(hash, key,
  56. value, null);
  57. break;
  58. }
  59. }
  60. }
  61. // 红黑树
  62. else if (f instanceof TreeBin) { //特殊节点
  63. Node<K,V> p;
  64. binCount = 2;
  65. // putTreeVal 会看 key 是否已经在树中, 是, 则返回对应的 TreeNode
  66. if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
  67. value)) != null) {
  68. oldVal = p.val;
  69. if (!onlyIfAbsent)
  70. p.val = value;
  71. }
  72. }
  73. }
  74. // 释放链表头节点的锁
  75. }
  76. if (binCount != 0) {
  77. if (binCount >= TREEIFY_THRESHOLD)
  78. // 如果链表长度 >= 树化阈值(8), 进行链表转为红黑树
  79. treeifyBin(tab, i);
  80. if (oldVal != null)
  81. return oldVal;
  82. break;
  83. }
  84. }
  85. }
  86. // 增加 size 计数
  87. addCount(1L, binCount);
  88. return null;
  89. }
  90. private final Node<K,V>[] initTable() {
  91. Node<K,V>[] tab; int sc;
  92. while ((tab = table) == null || tab.length == 0) {
  93. if ((sc = sizeCtl) < 0)
  94. Thread.yield();
  95. // 尝试将 sizeCtl 设置为 -1(表示初始化 table)
  96. else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
  97. // 获得锁, 创建 table, 这时其它线程会在 while() 循环中 yield 直至 table 创建
  98. try {
  99. if ((tab = table) == null || tab.length == 0) {
  100. int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
  101. Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
  102. table = tab = nt;
  103. sc = n - (n >>> 2);
  104. }
  105. } finally {
  106. sizeCtl = sc;
  107. }
  108. break;
  109. }
  110. }
  111. return tab;
  112. }
  113. // check 是之前 binCount 的个数
  114. private final void addCount(long x, int check) {
  115. CounterCell[] as; long b, s;
  116. if (
  117. // 已经有了 counterCells, 向 cell 累加
  118. (as = counterCells) != null ||
  119. // 还没有, 向 baseCount 累加
  120. !U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)
  121. ) {
  122. CounterCell a; long v; int m;
  123. boolean uncontended = true;
  124. if (
  125. // 还没有 counterCells
  126. as == null || (m = as.length - 1) < 0 ||
  127. // 还没有 cell
  128. (a = as[ThreadLocalRandom.getProbe() & m]) == null ||
  129. // cell cas 增加计数失败
  130. !(uncontended = U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))
  131. ) {
  132. // 创建累加单元数组和cell, 累加重试
  133. fullAddCount(x, uncontended);
  134. return;
  135. }
  136. if (check <= 1)
  137. return;
  138. // 获取元素个数
  139. s = sumCount();
  140. }
  141. if (check >= 0) {
  142. Node<K,V>[] tab, nt; int n, sc;
  143. while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&
  144. (n = tab.length) < MAXIMUM_CAPACITY) {
  145. int rs = resizeStamp(n);
  146. if (sc < 0) {
  147. if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
  148. sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
  149. transferIndex <= 0)
  150. break;
  151. // newtable 已经创建了,帮忙扩容
  152. if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
  153. transfer(tab, nt);
  154. }
  155. // 需要扩容,这时 newtable 未创建
  156. else if (U.compareAndSwapInt(this, SIZECTL, sc,
  157. (rs << RESIZE_STAMP_SHIFT) + 2))
  158. transfer(tab, null);
  159. s = sumCount();
  160. }
  161. }
  162. }

 

put流程:

  • 判断键值是否为空;为空则抛出异常

  • spread方法保证Hash码的正数

  • 进入for循环:

    • if table为null或者长度 为0 (table需要初始化)

      • 初始化hash表;initTable()

    • else if 索引处还没有节点 (添加头节点:索引处还没有节点的情况下,添加到头节点即可)

      • 使用CAS的方式创建头节点,填入key和value,放入table,退出循环;若添加不成功,则继续for循环;

    • else if 索引处节点的hash值为MOVE(ForwardingNode),表示正在扩容和迁移(帮忙扩容)

      • 帮忙扩容;(只要锁住某一个链表扩容即可)

    • else (存在哈希冲突的情况)

      • 锁住头节点

        • if 再次确认头节点没有被移动

          • if 头节点hash值大于0(表示这是一个链表)

            • 遍历链表找到对应key,做更新;如果没有,在链表尾创建新的node。

          • else if 节点为红黑树节点(Treebin)

            • 调用putTreeVal查看是否有对应key的数节点

              • 如果有且为覆盖模式,将值覆盖,返回旧值

              • 如果没有,创建并插入,返回null

        • 解锁

      • if binCount不为0 (bigCount为链表的节点数)

        • 如果binCount大于树化阈值8

          • 转换为红黑树

        • 如果旧值不为null

          • 返回旧值

        • break

  • 增加size计数(longAdder的思想)

  • return null

initTable()table初始化:

采用CAS的方式保证只有一个线程可以创建table,别的线程只是盲等待,并没有阻塞住;

流程:

  • 判断table是否为空(while)

    • if sizeCtl小于0

      • yieid(),代表别的线程在进行扩容

    • if CAS方法尝试将 sizeCtl 设置为 -1(表示初始化 table) 成功

      • 创建table

  • 返回table

5)size()方法

size 计算实际发生在 put,remove 改变集合元素的操作之中 (结果不是特别确,因为会有其它线程来添加或者删除)

  • 没有竞争发生,向 baseCount 累加计数

  • 有竞争发生,新建 counterCells,向其中的一个 cell 累加计

    • counterCells 初始有两个 cell

    • 如果计数竞争比较激烈,会创建新的 cell 来累加计数

  1. public int size() {
  2. long n = sumCount();
  3. return ((n < 0L) ? 0 :
  4. (n > (long)Integer.MAX_VALUE) ? Integer.MAX_VALUE :
  5. (int)n);
  6. }
  7. final long sumCount() {
  8. CounterCell[] as = counterCells; CounterCell a;
  9. // 将 baseCount 计数与所有 cell 计数累加
  10. long sum = baseCount;
  11. if (as != null) {
  12. for (int i = 0; i < as.length; ++i) {
  13. if ((a = as[i]) != null)
  14. sum += a.value;
  15. }
  16. }
  17. return sum;
  18. }

总结:

1)table初始化: 使用了CAS来保证并发安全的,懒惰初始化table。(一个线程在初始化时,别的线程在盲等待,没有阻塞)

2)转换红黑树: 当 table.length < 64 时,先尝试扩容,超过 64 时,并且 bin.length > 8 时,会将链表树化,树化过程 会用 synchronized 锁住链表头

3)put()方法: 如果头节点还没有,使用CAS方式创建;如果有了,则锁住链表,在链表尾添加节点;

4)get()方法: 是无锁操作,可以保证可见性;扩容过程中 get 操作拿到的是 ForwardingNode 它会让 get 操作在新 table 进行搜索;

5)size()方法: 不是一个准确值;

6)扩容时平均只有 1/6 的节点会把复制到新 table 中

4.ConcurrenHashMap (JDK7)

在JDK7中,它维护了一个segment数组,每个segment对应一把锁(每个segement对应一个HashEntry数组);

优点: 多个线程访问不同的 segment,也是没有冲突的,这与 jdk8 中是类似的,提高了并发度;

缺点: Segments 数组默认大小为16,这个容量初始化指定后就不能改变了,并且不是懒惰初始化

 2.LinkedBlockingQueue(阻塞队列)

1.基本的入队出队

1)入队

  1. public class LinkedBlockingQueue<E> extends AbstractQueue<E>
  2. implements BlockingQueue<E>, java.io.Serializable {
  3. static class Node<E> {
  4. E item;
  5. /**
  6. * next的三种情况:
  7. * - 真正的后继节点
  8. * - 指向自己, 发生在出队时
  9. * - null, 表示是没有后继节点, 是最后了
  10. */
  11. Node<E> next;
  12. Node(E x) { item = x; }
  13. }
  14. }

初始化链表last = head = new Node(null);,创建一个用来占的Dummy节点,item为null;

 ②当一个节点入队时,last = last.next = node

 ③再来一个节点

 2)出队

  1. //临时变量h用来指向哨兵
  2. Node<E> h = head;
  3. //first用来指向第一个元素
  4. Node<E> first = h.next;
  5. h.next = h; // help GC
  6. //head赋值为first,表示first节点就是下一个哨兵。
  7. head = first;
  8. E x = first.item;
  9. //删除first节点中的数据,表示真正成为了哨兵,第一个元素出队。
  10. first.item = null;
  11. return x;

h = head,将临时变量h指向哨兵节点

 ②first = h.next,将first指向下一个节点,也就是第一个元素

 ③h.next = h,将Dummy的next指向自己

 ④head = first,将第一个元素的item变为null,让其变为Dummy节点

2.加锁分析

LinkedBlocking的高明之处在于它用了两把锁以及Dummy节点:

  • 用一把锁的话,同一时刻,最多只允许有一个线程(生产者或者消费者)执行。

  • 用两把锁的话,同一时刻,可以有两个线程同时(一个生产者一个消费者)执行。

    • 消费者与消费者仍然串行

    • 生产者与生产者仍然串行

线程安全分析:

  • 当正常节点数大于1时,putLock保证的是last节点的线程安全,takeLock保证的是head节点的线程安全。两把锁保证了入队和出队没有竞争。

  • 当正常节点数等于1时(也就是一个Dummy节点,一个正常节点),这时候,仍然两把锁锁两个对象,不会竞争

  • 当只有一个Dummy节点时,这时take线程会被notEmpty条件阻塞

 // 用于 put锁
private final ReentrantLock putLock = new ReentrantLock();
// 用户 take锁
private final ReentrantLock takeLock = new ReentrantLock();

3.put()和take()方法

1)put操作

  1. public void put(E e) throws InterruptedException {
  2. //LinkedBlockingQueue不支持空元素
  3. if (e == null) throw new NullPointerException();
  4. int c = -1;
  5. Node<E> node = new Node<E>(e);
  6. final ReentrantLock putLock = this.putLock;
  7. // count 用来维护元素计数
  8. final AtomicInteger count = this.count;
  9. putLock.lockInterruptibly();
  10. try {
  11. // 满了等待
  12. while (count.get() == capacity) {
  13. // 倒过来读就好: 等待 notFull
  14. notFull.await();
  15. }
  16. // 有空位, 入队且计数加一
  17. enqueue(node);
  18. c = count.getAndIncrement();
  19. // 除了自己 put 以外, 队列还有空位, 由自己叫醒其他 put 线程
  20. if (c + 1 < capacity)
  21. notFull.signal();
  22. } finally {
  23. putLock.unlock();
  24. }
  25. // 如果队列中有一个元素, 叫醒 take 线程
  26. if (c == 0)
  27. // 这里调用的是 notEmpty.signal() 而不是 notEmpty.signalAll() 是为了减少竞争
  28. signalNotEmpty();
  29. }

put流程:

  • 判断要加入的元素是否为空,为空则抛出异常;(因为它不允许有空元素)

  • 创建原子计数的count

  • putLock

    • while 判断队列是否已满

      • 满的话,就await()等待;

    • 有空位的话,就入队

    • count 计数加1

    • if 自己put之后,队列还有空位

      • 唤醒其它的等待一个put线程 (只唤醒一个的目的是减少竞争)

  • 解锁

  • if 队列只有一个元素 (put完只有一个元素,代表之前可能有take阻塞住了)

    • signalNotEmpty()唤醒一个等待的take线程

2)take操作

  1. public E take() throws InterruptedException {
  2. E x;
  3. int c = -1;
  4. final AtomicInteger count = this.count;
  5. final ReentrantLock takeLock = this.takeLock;
  6. takeLock.lockInterruptibly();
  7. try {
  8. while (count.get() == 0) {
  9. notEmpty.await();
  10. }
  11. x = dequeue();
  12. c = count.getAndDecrement();
  13. if (c > 1)
  14. notEmpty.signal();
  15. } finally {
  16. takeLock.unlock();
  17. }
  18. // 如果队列中只有一个空位时, 叫醒 put 线程
  19. // 如果有多个线程进行出队, 第一个线程满足 c == capacity, 但后续线程 c < capacity
  20. if (c == capacity)
  21. // 这里调用的是 notFull.signal() 而不是 notFull.signalAll() 是为了减少竞争
  22. signalNotFull()
  23. return x;
  24. }
与ArrayBlockingQueue的性能比较
  • Linked 支持有界,Array 强制有界

  • Linked 实现是链表,Array 实现是数组

  • Linked 是懒惰的,而 Array 需要提前初始化 Node 数组

  • Linked 每次入队会生成新 Node,而 Array 的 Node 是提前创建好的

  • Linked 两把锁,Array 一把锁(所以Array在性能上不如Linked,所以推荐使用Linked

3.ConcurrentLinkedQueue

特点:

ConcurrentLinkedQueue 的设计与 LinkedBlockingQueue 非常像:

  • 都是用了两把“锁”, 同一时刻,可以允许两个线程同时(一个生产者一个消费者)执行

  • Dummy节点的引入让两把“锁”将来锁住的是不同的对象,避免竞争

  • 注意:ConcurrentLinkedQueue这里的“锁”是采用了CAS的方式,并不是真正的锁;

应用: 并发度比较高的时候适合使用,因为采用了CAS的方式!

比如,Tomcat 的 Connector 结构,Acceptor 作为生产者向 Poller 消费者传递事件信息时,正是采用了 ConcurrentLinkedQueue 将 SocketChannel 给 Poller 使用

4.CopyOnWriteArrayList

它的底层实现采用了写入时拷贝的思想,增删改操作都会将底层数据拷贝一份,更改操作在新的数组上执行,这时不影响其它线程的并发读,实现了读-读分离,读-写分离(但是读的是旧数组上的数据),只有写-写互斥;

新增操作:

  1. public boolean add(E e) {
  2. synchronized (lock) {
  3. // 获取旧的数组
  4. Object[] es = getArray();
  5. int len = es.length;
  6. // 拷贝新的数组(这里是比较耗时的操作,但不影响其它读线程)
  7. es = Arrays.copyOf(es, len + 1);
  8. // 添加新元素
  9. es[len] = e;
  10. // 替换旧的数组
  11. setArray(es);
  12. return true;
  13. }
  14. }

读操作并未加锁;

应用: 适合读多写少的场景!

缺点:

①get弱一致性;

 ②迭代器弱一致性

  1. CopyOnWriteArrayList<Integer> list = new CopyOnWriteArrayList<>();
  2. list.add(1);
  3. list.add(2);
  4. list.add(3);
  5. Iterator<Integer> iter = list.iterator();
  6. new Thread(() -> {
  7. list.remove(0);
  8. System.out.println(list);
  9. }).start();
  10. sleep1s();
  11. //此时主线程的iterator依旧指向旧的数组。
  12. while (iter.hasNext()) {
  13. System.out.println(iter.next());
  14. }

声明:本文内容由网友自发贡献,不代表【wpsshop博客】立场,版权归原作者所有,本站不承担相应法律责任。如您发现有侵权的内容,请联系我们。转载请注明出处:https://www.wpsshop.cn/w/AllinToyou/article/detail/670938
推荐阅读
相关标签
  

闽ICP备14008679号