Java多线程（四）之ConcurrentSkipListMap深入分析

一、前言

concurrentHashMap与ConcurrentSkipListMap性能测试

在4线程1.6万数据的条件下，ConcurrentHashMap 存取速度是ConcurrentSkipListMap 的4倍左右。

但ConcurrentSkipListMap有几个ConcurrentHashMap 不能比拟的优点：

1、ConcurrentSkipListMap 的key是有序的。

2、ConcurrentSkipListMap 支持更高的并发。ConcurrentSkipListMap 的存取时间是log（N），和线程数几乎无关。也就是说在数据量一定的情况下，并发的线程越多，ConcurrentSkipListMap越能体现出他的优势。

二、使用建议

在非多线程的情况下，应当尽量使用TreeMap。此外对于并发性相对较低的并行程序可以使用Collections.synchronizedSortedMap将TreeMap进行包装，也可以提供较好的效率。对于高并发程序，应当使用ConcurrentSkipListMap，能够提供更高的并发度。

所以在多线程程序中，如果需要对Map的键值进行排序时，请尽量使用ConcurrentSkipListMap，可能得到更好的并发度。
注意，调用ConcurrentSkipListMap的size时，由于多个线程可以同时对映射表进行操作，所以映射表需要遍历整个链表才能返回元素个数，这个操作是个O(log(n))的操作。

二、什么是SkipList

Skip list(跳表）是一种可以代替平衡树的数据结构，默认是按照Key值升序的。Skip list让已排序的数据分布在多层链表中，以0-1随机数决定一个数据的向上攀升与否，通过“空间来换取时间”的一个算法，在每个节点中增加了向前的指针，在插入、删除、查找时可以忽略一些不可能涉及到的结点，从而提高了效率。

从概率上保持数据结构的平衡比显示的保持数据结构平衡要简单的多。对于大多数应用，用Skip list要比用树算法相对简单。由于Skip list比较简单，实现起来会比较容易，虽然和平衡树有着相同的时间复杂度(O(logn)),但是skip list的常数项会相对小很多。Skip list在空间上也比较节省。一个节点平均只需要1.333个指针（甚至更少）。
                
图1-1 Skip list结构图（以7,14,21,32,37,71,85序列为例）

Skip list的性质

(1) 由很多层结构组成，level是通过一定的概率随机产生的。
(2) 每一层都是一个有序的链表，默认是升序，也可以根据创建映射时所提供的Comparator进行排序，具体取决于使用的构造方法。
(3) 最底层(Level 1)的链表包含所有元素。
(4) 如果一个元素出现在Level i 的链表中，则它在Level i 之下的链表也都会出现。
(5) 每个节点包含两个指针，一个指向同一链表中的下一个元素，一个指向下面一层的元素。

三、什么是ConcurrentSkipListMap

ConcurrentSkipListMap提供了一种线程安全的并发访问的排序映射表。内部是SkipList（跳表）结构实现，在理论上能够在O(log(n))时间内完成查找、插入、删除操作。
       注意，调用ConcurrentSkipListMap的size时，由于多个线程可以同时对映射表进行操作，所以映射表需要遍历整个链表才能返回元素个数，这个操作是个O(log(n))的操作。

ConcurrentSkipListMap存储结构

 


ConcurrentSkipListMap存储结构图

跳跃表（SkipList）：（如上图所示）
1.多条链构成，是关键字升序排列的数据结构；
2.包含多个级别，一个head引用指向最高的级别，最低（底部）的级别，包含所有的key；
3.每一个级别都是其更低级别的子集，并且是有序的；
4.如果关键字 key在 级别level=i中出现，则，level<=i的链表中都会包含该关键字key；

------------------------

ConcurrentSkipListMap主要用到了Node和Index两种节点的存储方式，通过volatile关键字实现了并发的操作

[java] view plaincopy

  1. static final class Node<K,V> {
  2. final K key;
  3. volatile Object value;//value值
  4. volatile Node<K,V> next;//next引用
  5. ……
  6. }
  7. static class Index<K,V> {
  8. final Node<K,V> node;
  9. final Index<K,V> down;//downy引用
  10. volatile Index<K,V> right;//右边引用
  11. ……
  12. }

------------------------

ConcurrentSkipListMap的查找

 

通过SkipList的方式进行查找操作：（下图以“查找91”进行说明：）

红色虚线，表示查找的路径，蓝色向右箭头表示right引用；黑色向下箭头表示down引用；

/get方法，通过doGet操作实现

[java] view plaincopy

  1. public V get(Object key) {
  2. return doGet(key);
  3. }
  4. //doGet的实现
  5. private V doGet(Object okey) {
  6. Comparable<? super K> key = comparable(okey);
  7. Node<K,V> bound = null;
  8. Index<K,V> q = head;//把头结点作为当前节点的前驱节点
  9. Index<K,V> r = q.right;//前驱节点的右节点作为当前节点
  10. Node<K,V> n;
  11. K k;
  12. int c;
  13. for (;;) {//遍历
  14. Index<K,V> d;
  15. // 依次遍历right节点
  16. if (r != null && (n = r.node) != bound && (k = n.key) != null) {
  17. if ((c = key.compareTo(k)) > 0) {//由于key都是升序排列的，所有当前关键字大于所要查找的key时继续向右遍历
  18. q = r;
  19. r = r.right;
  20. continue;
  21. } else if (c == 0) {
  22. //如果找到了相等的key节点，则返回该Node的value如果value为空可能是其他并发delete导致的，于是通过另一种
  23. //遍历findNode的方式再查找
  24. Object v = n.value;
  25. return (v != null)? (V)v : getUsingFindNode(key);
  26. } else
  27. bound = n;
  28. }
  29. //如果一个链表中right没能找到key对应的value，则调整到其down的引用处继续查找
  30. if ((d = q.down) != null) {
  31. q = d;
  32. r = d.right;
  33. } else
  34. break;
  35. }
  36. // 如果通过上面的遍历方式，还没能找到key对应的value，再通过Node.next的方式进行查找
  37. for (n = q.node.next;  n != null; n = n.next) {
  38. if ((k = n.key) != null) {
  39. if ((c = key.compareTo(k)) == 0) {
  40. Object v = n.value;
  41. return (v != null)? (V)v : getUsingFindNode(key);
  42. } else if (c < 0)
  43. break;
  44. }
  45. }
  46. return null;
  47. }

------------------------------------------------

ConcurrentSkipListMap的删除

 

通过SkipList的方式进行删除操作：（下图以“删除23”进行说明：）

红色虚线，表示查找的路径，蓝色向右箭头表示right引用；黑色向下箭头表示down引用；

[java] view plaincopy

  1. //remove操作，通过doRemove实现，把所有level中出现关键字key的地方都delete掉
  2. public V remove(Object key) {
  3. return doRemove(key, null);
  4. }
  5. final V doRemove(Object okey, Object value) {
  6. Comparable<? super K> key = comparable(okey);
  7. for (;;) {
  8. Node<K,V> b = findPredecessor(key);//得到key的前驱（就是比key小的最大节点）
  9. Node<K,V> n = b.next;//前驱节点的next引用
  10. for (;;) {//遍历
  11. if (n == null)//如果next引用为空，直接返回
  12. return null;
  13. Node<K,V> f = n.next;
  14. if (n != b.next)                    // 如果两次获得的b.next不是相同的Node，就跳转到第一层循环重新获得b和n
  15. break;
  16. Object v = n.value;
  17. if (v == null) {                    // 当n被其他线程delete的时候，其value==null，此时做辅助处理，并重新获取b和n
  18. n.helpDelete(b, f);
  19. break;
  20. }
  21. if (v == n || b.value == null)      // 当其前驱被delet的时候直接跳出，重新获取b和n
  22. break;
  23. int c = key.compareTo(n.key);
  24. if (c < 0)
  25. return null;
  26. if (c > 0) {//当key较大时就继续遍历
  27. b = n;
  28. n = f;
  29. continue;
  30. }
  31. if (value != null && !value.equals(v))
  32. return null;
  33. if (!n.casValue(v, null))
  34. break;
  35. if (!n.appendMarker(f) || !b.casNext(n, f))//casNext方法就是通过比较和设置b（前驱）的next节点的方式来实现删除操作
  36. findNode(key);                  // 通过尝试findNode的方式继续find
  37. else {
  38. findPredecessor(key);           // Clean index
  39. if (head.right == null)   //如果head的right引用为空，则表示不存在该level
  40. tryReduceLevel();
  41. }
  42. return (V)v;
  43. }
  44. }
  45. }

-------------------------------------

ConcurrentSkipListMap的插入

通过SkipList的方式进行插入操作：（下图以“添加55”的两种情况，进行说明：）

在level=2（该level存在）的情况下添加55的图示:只需在level<=2的合适位置插入55即可

--------

在level=4(该level不存在，图示level4是新建的)的情况下添加55的情况：首先新建level4,然后在level<=4的合适位置插入55

-----------

[java] view plaincopy

  1. //put操作，通过doPut实现
  2. public V put(K key, V value) {
  3. if (value == null)
  4. throw new NullPointerException();
  5. return doPut(key, value, false);
  6. }
  7. private V doPut(K kkey, V value, boolean onlyIfAbsent) {
  8. Comparable<? super K> key = comparable(kkey);
  9. for (;;) {
  10. Node<K,V> b = findPredecessor(key);//前驱
  11. Node<K,V> n = b.next;
  12. //定位的过程就是和get操作相似
  13. for (;;) {
  14. if (n != null) {
  15. Node<K,V> f = n.next;
  16. if (n != b.next)               // 前后值不一致的情况下，跳转到第一层循环重新获得b和n
  17. break;;
  18. Object v = n.value;
  19. if (v == null) {               // n被delete的情况下
  20. n.helpDelete(b, f);
  21. break;
  22. }
  23. if (v == n || b.value == null) // b 被delete的情况，重新获取b和n
  24. break;
  25. int c = key.compareTo(n.key);
  26. if (c > 0) {
  27. b = n;
  28. n = f;
  29. continue;
  30. }
  31. if (c == 0) {
  32. if (onlyIfAbsent || n.casValue(v, value))
  33. return (V)v;
  34. else
  35. break; // restart if lost race to replace value
  36. }
  37. // else c < 0; fall through
  38. }
  39. Node<K,V> z = new Node<K,V>(kkey, value, n);
  40. if (!b.casNext(n, z))
  41. break;         // restart if lost race to append to b
  42. int level = randomLevel();//得到一个随机的level作为该key-value插入的最高level
  43. if (level > 0)
  44. insertIndex(z, level);//进行插入操作
  45. return null;
  46. }
  47. }
  48. }
  49. /**
  50. * 获得一个随机的level值
  51. */
  52. private int randomLevel() {
  53. int x = randomSeed;
  54. x ^= x << 13;
  55. x ^= x >>> 17;
  56. randomSeed = x ^= x << 5;
  57. if ((x & 0x8001) != 0) // test highest and lowest bits
  58. return 0;
  59. int level = 1;
  60. while (((x >>>= 1) & 1) != 0) ++level;
  61. return level;
  62. }
  63. //执行插入操作：如上图所示，有两种可能的情况：
  64. //1.当level存在时，对level<=n都执行insert操作
  65. //2.当level不存在（大于目前的最大level）时，首先添加新的level，然后在执行操作1
  66. private void insertIndex(Node<K,V> z, int level) {
  67. HeadIndex<K,V> h = head;
  68. int max = h.level;
  69. if (level <= max) {//情况1
  70. Index<K,V> idx = null;
  71. for (int i = 1; i <= level; ++i)//首先得到一个包含1~level个级别的down关系的链表，最后的inx为最高level
  72. idx = new Index<K,V>(z, idx, null);
  73. addIndex(idx, h, level);//把最高level的idx传给addIndex方法
  74. } else { // 情况2 增加一个新的级别
  75. level = max + 1;
  76. Index<K,V>[] idxs = (Index<K,V>[])new Index[level+1];
  77. Index<K,V> idx = null;
  78. for (int i = 1; i <= level; ++i)//该步骤和情况1类似
  79. idxs[i] = idx = new Index<K,V>(z, idx, null);
  80. HeadIndex<K,V> oldh;
  81. int k;
  82. for (;;) {
  83. oldh = head;
  84. int oldLevel = oldh.level;
  85. if (level <= oldLevel) { // lost race to add level
  86. k = level;
  87. break;
  88. }
  89. HeadIndex<K,V> newh = oldh;
  90. Node<K,V> oldbase = oldh.node;
  91. for (int j = oldLevel+1; j <= level; ++j)
  92. newh = new HeadIndex<K,V>(oldbase, newh, idxs[j], j);//创建新的
  93. if (casHead(oldh, newh)) {
  94. k = oldLevel;
  95. break;
  96. }
  97. }
  98. addIndex(idxs[k], oldh, k);
  99. }
  100. }
  101. /**
  102. *在1~indexlevel层中插入数据
  103. */
  104. private void addIndex(Index<K,V> idx, HeadIndex<K,V> h, int indexLevel) {
  105. //  insertionLevel 代表要插入的level，该值会在indexLevel~1间遍历一遍
  106. int insertionLevel = indexLevel;
  107. Comparable<? super K> key = comparable(idx.node.key);
  108. if (key == null) throw new NullPointerException();
  109. // 和get操作类似，不同的就是查找的同时在各个level上加入了对应的key
  110. for (;;) {
  111. int j = h.level;
  112. Index<K,V> q = h;
  113. Index<K,V> r = q.right;
  114. Index<K,V> t = idx;
  115. for (;;) {
  116. if (r != null) {
  117. Node<K,V> n = r.node;
  118. // compare before deletion check avoids needing recheck
  119. int c = key.compareTo(n.key);
  120. if (n.value == null) {
  121. if (!q.unlink(r))
  122. break;
  123. r = q.right;
  124. continue;
  125. }
  126. if (c > 0) {
  127. q = r;
  128. r = r.right;
  129. continue;
  130. }
  131. }
  132. if (j == insertionLevel) {//在该层level中执行插入操作
  133. // Don‘t insert index if node already deleted
  134. if (t.indexesDeletedNode()) {
  135. findNode(key); // cleans up
  136. return;
  137. }
  138. if (!q.link(r, t))//执行link操作，其实就是inset的实现部分
  139. break; // restart
  140. if (--insertionLevel == 0) {
  141. // need final deletion check before return
  142. if (t.indexesDeletedNode())
  143. findNode(key);
  144. return;
  145. }
  146. }
  147. if (--j >= insertionLevel && j < indexLevel)//key移动到下一层level
  148. t = t.down;
  149. q = q.down;
  150. r = q.right;
  151. }
  152. }
  153. }

参考：

集合框架 Map篇（5）----ConcurrentSkipListMap http://hi.baidu.com/yao1111yao/item/0f3008163c4b82c938cb306d
Java里多个Map的性能比较（TreeMap、HashMap、ConcurrentSkipListMap） http://blog.hongtium.com/java-map-skiplist/
跳表SkipList的原理和实现 http://imtinx.iteye.com/blog/1291165

时间： 2024-10-10 19:51:58

Java多线程（四）之ConcurrentSkipListMap深入分析的相关文章

java多线程四种实现模板

假设一个项目拥有三块独立代码块,需要执行,什么时候用多线程? 这些代码块某些时候需要同时运行,彼此独立,那么需要用到多线程操作更快... 这里把模板放在这里,需要用的时候寻找合适的来选用. 总体分为两种: 一.使用匿名内部类的方式(推荐) 1)方式1开启多线程(Thread) //使用第一种方式,开启线程 new Thread() { public void run() { for(int i=0;i<100;i++) { System.out.println(Thread.currentThr

Java多线程(四) —— 线程并发库之Atomic

一.从原子操作开始 从相对简单的Atomic入手(java.util.concurrent是基于Queue的并发包,而Queue,很多情况下使用到了Atomic操作,因此首先从这里开始). 很多情况下我们只是需要一个简单的.高效的.线程安全的递增递减方案.注意,这里有三个条件: 简单,意味着程序员尽可能少的操作底层或者实现起来要比较容易: 高效意味着耗用资源要少,程序处理速度要快: 线程安全也非常重要,这个在多线程下能保证数据的正确性. 这三个条件看起来比较简单,但是实现起来却难以令人满意. 通

java多线程(四)－自定义线程池

当我们使用 线程池的时候,可以使用 newCachedThreadPool()或者 newFixedThreadPool(int)等方法,其实我们深入到这些方法里面,就可以看到它们的是实现方式是这样的. 1 public static ExecutorService newCachedThreadPool() { 2 return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 3 60L, TimeUnit.SECONDS, 4 new Synchro

Java多线程(四) 线程池

一个优秀的软件不会随意的创建很销毁线程,因为创建和销毁线程需要耗费大量的CPU时间以及需要和内存做出大量的交互.因此JDK5提出了使用线程池,让程序员把更多的精力放在业务逻辑上面,弱化对线程的开闭管理. JDK提供了四种不同的线程池给程序员使用 首先使用线程池,需要用到ExecutorService接口,该接口有个抽象类AbstractExecutorService对其进行了实现,ThreadPoolExecutor进一步对抽象类进行了实现.最后JDK封装了一个Executor类对ThreadP

java多线程四种实现方法

package com.zrun.TestThread; import java.text.SimpleDateFormat; import java.util.Date; import java.util.concurrent.Callable; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Exec

Java多线程——&lt;四&gt;让线程有返回值

一.概述 到目前为止,我们已经能够声明并使一个线程任务运行起来了.但是遇到一个问题:现在定义的任务都没有任何返回值,那么加入我们希望一个任务运行结束后告诉我一个结果,该结果表名任务执行成功或失败,此时该怎么办呢? 答案是使用Callable.之前定义的任务都直接实现了Runnable,该接口的run方法并无返回值.而Callable的call方法可以根据你传入的泛型参数返回对应类型的数据. 二.实现 1.实现Callable接口,定义可返回结果的线程任务 public class TaskCal

Java多线程——&lt;八&gt;多线程其他概念

一.概述 到第八节,就把多线程基本的概念都说完了.把前面的所有文章加连接在此: Java多线程——<一>概述.定义任务 Java多线程——<二>将任务交给线程,线程声明及启动 Java多线程——<三>简单的线程执行:Executor Java多线程——<四>让线程有返回值 Java多线程——<五>后台线程(daemon) Java多线程——<六>更方便的线程 Java多线程——<七>多线程的异常捕捉 均是个人理解和总结,代

Java多线程基础（四）Java传统线程同步通信技术

Java多线程基础(四)Java传统线程同步通信技术 编写代码实现以下功能 子线程循环10次,接着主线程循环100次,接着又回到子线程循环10次,接着再回到主线程又循环100次,如此循环50次. 分析 1)子线程循环10次与主线程循环100次必须是互斥的执行,不能出现交叉,下面代码中通过synchronized关键字实现此要求: 2)子线程与主线程必须交替出现,可以通过线程同步通信技术实现,下面代码中通过bShouldSub变量实现此要求: 其他需要注意的地方 1)其中business变量必须声

java多线程系列(四)

Lock的使用 前言:本系列将从零开始讲解java多线程相关的技术,内容参考于<java多线程核心技术>与<java并发编程实战>等相关资料,希望站在巨人的肩膀上,再通过我的理解能让知识更加简单易懂. 目录 认识cpu.核心与线程 java多线程系列(一)之java多线程技能 java多线程系列(二)之对象变量的并发访问 java多线程系列(三)之等待通知机制 java多线程系列(四)之ReentrantLock的使用 ReentrantLock(重入锁) public class

java基础知识回顾之java Thread类学习（四）--java多线程安全问题（锁）

上一节售票系统中我们发现,打印出了错票,0,-1,出现了多线程安全问题.我们分析为什么会发生多线程安全问题? 看下面线程的主要代码: @Override public void run() { // TODO Auto-generated method stub while(true){ if(ticket > 0){//当线程0被调起的时候,当执行到这条判断语句的时候,线程1被调起抢了CPU资源,线程0进入冻结状态. try { Thread.sleep(100);//中断当前活跃的线程,或者