文章详情

短信预约-IT技能 免费直播动态提醒

请输入下面的图形验证码

提交验证

短信预约提醒成功

详解LongAdder实现原理

2024-12-03 11:55

关注

AtomicInteger、AtomicLong使用非阻塞的CAS算法原子性地更新某一个变量,比synchronized这些阻塞算法拥有更好的性能,但是在高并发情况下,大量线程同时去更新一个变量,由于同一时间只有一个线程能够成功,绝大部分的线程在尝试更新失败后,会通过自旋的方式再次进行尝试,严重占用了CPU的时间片。

AtomicLong的实现原理图:

 

 

LongAdder是JDK8新增的原子操作类,它提供了一种新的思路,既然AtomicLong的性能瓶颈是由于大量线程同时更新一个变量造成的,那么能不能把这个变量拆分出来,变成多个变量,然后让线程去竞争这些变量,最后合并即可?LongAdder的设计精髓就在这里,通过将变量拆分成多个元素,降低该变量的并发度,最后进行合并元素,变相的减少了CAS的失败次数。

LongAdder的实现原理图:


常用方法

 

  1. public class LongAdder extends Striped64 implements Serializable { 
  2.     //构造方法 
  3.     public LongAdder() { 
  4.     } 
  5.     //加1操作 
  6.     public void increment(); 
  7.     //减1操作 
  8.     public void decrement(); 
  9.     //获取原子变量的值 
  10.     public long longValue(); 
  11.  

 

下面给出一个简单的例子,模拟50线程同时进行更新

 

  1. package com.xue.testLongAdder; 
  2.  
  3. import java.util.concurrent.ExecutorService; 
  4. import java.util.concurrent.Executors; 
  5. import java.util.concurrent.atomic.LongAdder; 
  6.  
  7. public class Main { 
  8.     public static void main(String[] args) { 
  9.         LongAdder adder = new LongAdder(); 
  10.         ExecutorService threadPool = Executors.newFixedThreadPool(20); 
  11.         for (int i = 0; i < 50; i++) { 
  12.             Runnable r = () -> { 
  13.                 adder.add(1); 
  14.             }; 
  15.             threadPool.execute(r); 
  16.         } 
  17.         threadPool.shutdown(); 
  18.         //若关闭线程池后,所有任务执行完毕,则isTerminated()返回true 
  19.         while (!threadPool.isTerminated()) { 
  20.             System.out.println(adder.longValue()); 
  21.             break; 
  22.         } 
  23.     } 

 

输出结果是50

其中,如果对线程池不熟悉的同学,可以先参考我的另外一篇文章说说线程池

原理解析

类图


LongAdder内部维护了一个Cell类型的数组,其中Cell是Striped64中的一个静态内部类。

Cell类

 

  1. abstract class Striped64 extends Number { 
  2.    
  3.     @sun.misc.Contended static final class Cell { 
  4.         volatile long value; 
  5.         Cell(long x) { value = x; } 
  6.         final boolean cas(long cmp, long val) { 
  7.             return UNSAFE.compareAndSwapLong(this, valueOffset, cmp, val); 
  8.         } 
  9.     } 
  10.  

 

Cell用来封装被拆分出来的元素,内部用一个value字段保存当前元素的值,等到需要合并时,则累加所有Cell数组中的value。Cell内部使用CAS操作来更新value值,对CAS操作不熟悉的同学,可以参考我的另外一篇文章浅探CAS实现原理

可以注意到,Cell类被 @sun.misc.Contended注解修饰了,这个注解是为了解决伪共享问题的,什么是伪共享?

对伪共享还不理解的同学,可以参考这位大佬的文章伪共享(False Sharing)底层原理及其解决方式

而LongAdder采用的是Cell数组,而数组元素是连续的,因此多个Cell对象共享一个缓存行的情况非常普遍,因此这里@sun.misc.Contended注解对单个Cell元素进行字节填充,确保一个Cell对象占据一个缓存行,即填充至64字节。

关于如何确定一个对象的大小,可以参考我的另外一篇文章对象的内存布局,怎样确定对象的大小,这样可以算出来,还需要填充多少字节。

longValue()

longValue()返回累加后的值

 

  1. public long longValue() { 
  2.      return sum(); 
  3.  } 
  4.  
  5.  public long sum() { 
  6.      Cell[] as = cells; Cell a; 
  7.      long sum = base; 
  8.      //当Cell数组不为null时,进行累加后返回,否则直接返回基准数base 
  9.      if (as != null) { 
  10.          for (int i = 0; i < as.length; ++i) { 
  11.              if ((a = as[i]) != null
  12.                  sum += a.value; 
  13.          } 
  14.      } 
  15.      return sum
  16.  } 

 

这可能是LongAdder中最简单的方法了,就不进行赘述了。什么,你要看复杂的?好的,这就来了。

increment()

 

  1. public void increment() { 
  2.        add(1L); 
  3.     } 
  4.  
  5.  public void add(long x) { 
  6.      Cell[] as; long b, v; int m; Cell a; 
  7.       
  8.      if ((as = cells) != null || !casBase(b = base, b + x)) { 
  9.          //uncontended判断cells数组中,当前线程要做cas累加操作的某个元素是否#不#存在争用, 
  10.          //如果cas失败则存在争用;uncontended=false代表存在争用,uncontended=true代表不存在争用。 
  11.          boolean uncontended = true
  12.           
  13.          if (as == null || (m = as.length - 1) < 0 || 
  14.                  (a = as[getProbe() & m]) == null || 
  15.                  !(uncontended = a.cas(v = a.value, v + x))) 
  16.              longAccumulate(x, null, uncontended); 
  17.      } 
  18.  } 

 

其中longAccumulate()的解析如下:

 

  1. final void longAccumulate(long x, LongBinaryOperator fn, boolean wasUncontended) { 
  2.     //获取当前线程的threadLocalRandomProbe值作为hash值,如果当前线程的threadLocalRandomProbe为0, 
  3.     // 说明当前线程是第一次进入该方法,则强制设置线程的threadLocalRandomProbe为ThreadLocalRandom类的成员 
  4.     // 静态私有变量probeGenerator的值,后面会详细将hash值的生成; 
  5.     //另外需要注意,如果threadLocalRandomProbe=0,代表新的线程开始参与cell争用的情况 
  6.     //1.当前线程之前还没有参与过cells争用(也许cells数组还没初始化,进到当前方法来就是为了初始化cells数组 
  7.     //后争用的), 
  8.     // 是第一次执行base的cas累加操作失败; 
  9.     //2.或者是在执行add方法时,对cells某个位置的Cell的cas操作第一次失败,则将wasUncontended设置为false, 
  10.     // 那么这里会将其重新置为true;第一次执行操作失败; 
  11.     //凡是参与了cell争用操作的线程threadLocalRandomProbe都不为0; 
  12.     int h; 
  13.     if ((h = getProbe()) == 0) { 
  14.         //初始化ThreadLocalRandom; 
  15.         ThreadLocalRandom.current(); // force initialization 
  16.         //将h设置为0x9e3779b9 
  17.         h = getProbe(); 
  18.         //设置未竞争标记为true 
  19.         wasUncontended = true
  20.     } 
  21.     //cas冲突标志,表示当前线程hash到的Cells数组的位置,做cas累加操作时与其它线程发生了冲突,cas失败; 
  22.     // collide=true代表有冲突,collide=false代表无冲突 
  23.     boolean collide = false
  24.     for (;;) { 
  25.         Cell[] as; Cell a; int n; long v; 
  26.         //这个主干if有三个分支 
  27.         //1.主分支一:处理cells数组已经正常初始化了的情况(这个if分支处理add方法的四个条件中的3和4) 
  28.         //2.主分支二:处理cells数组没有初始化或者长度为0的情况;(这个分支处理add方法的四个条件中的1和2) 
  29.         //3.主分支三:处理如果cell数组没有初始化,并且其它线程正在执行对cells数组初始化的操作, 
  30.         // 及cellbusy=1; 
  31.         // 则尝试将累加值通过cas累加到base上 
  32.         //先看主分支一 
  33.         if ((as = cells) != null && (n = as.length) > 0) { 
  34.              
  35.             if ((a = as[(n - 1) & h]) == null) { 
  36.                 //cellsBusy == 0 代表当前没有线程cells数组做修改 
  37.                 if (cellsBusy == 0) { 
  38.                     //将要累加的x值作为初始值创建一个新的Cell对象, 
  39.                     Cell r = new Cell(x); 
  40.                     //如果cellsBusy=0无锁,则通过cas将cellsBusy设置为1加锁 
  41.                     if (cellsBusy == 0 && casCellsBusy()) { 
  42.                         //标记Cell是否创建成功并放入到cells数组被hash的位置上 
  43.                         boolean created = false
  44.                         try { 
  45.                             Cell[] rs; int m, j; 
  46.                             //再次检查cells数组不为null,且长度不为空,且hash到的位置的Cell为null 
  47.                             if ((rs = cells) != null && 
  48.                                     (m = rs.length) > 0 && 
  49.                                     rs[j = (m - 1) & h] == null) { 
  50.                                 //将新的cell设置到该位置 
  51.                                 rs[j] = r; 
  52.                                 created = true
  53.                             } 
  54.                         } finally { 
  55.                             //去掉锁 
  56.                             cellsBusy = 0; 
  57.                         } 
  58.                         //生成成功,跳出循环 
  59.                         if (created) 
  60.                             break; 
  61.                         //如果created为false,说明上面指定的cells数组的位置cells[m%cells.length] 
  62.                         // 已经有其它线程设置了cell了, 
  63.                         // 继续执行循环。 
  64.                         continue
  65.                     } 
  66.                 } 
  67.                 //如果执行的当前行,代表cellsBusy=1,有线程正在更改cells数组,代表产生了冲突,将collide设置为false 
  68.                 collide = false
  69.  
  70.                  
  71.             } else if (!wasUncontended) 
  72.                 //设置未竞争标志位true,继续执行,后面会算一个新的probe值,然后重新执行循环。 
  73.                 wasUncontended = true
  74.              
  75.             else if (a.cas(v = a.value, ((fn == null) ? v + x : 
  76.                     fn.applyAsLong(v, x)))) 
  77.                 break; 
  78.              
  79.             else if (n >= NCPU || cells != as
  80.                 collide = false
  81.              
  82.             else if (!collide) 
  83.                 //设置冲突标志,表示发生了冲突,需要再次生成hash,重试。 
  84.                 // 如果下次重试任然走到了改分支此时collide=true,!collide条件不成立,则走后一个分支 
  85.                 collide = true
  86.              
  87.             else if (cellsBusy == 0 && casCellsBusy()) { 
  88.                 try { 
  89.                     //检查cells是否已经被扩容 
  90.                     if (cells == as) {      // Expand table unless stale 
  91.                         Cell[] rs = new Cell[n << 1]; 
  92.                         for (int i = 0; i < n; ++i) 
  93.                             rs[i] = as[i]; 
  94.                         cells = rs; 
  95.                     } 
  96.                 } finally { 
  97.                     cellsBusy = 0; 
  98.                 } 
  99.                 collide = false
  100.                 continue;                   // Retry with expanded table 
  101.             } 
  102.             //为当前线程重新计算hash值 
  103.             h = advanceProbe(h); 
  104.  
  105.             //这个大的分支处理add方法中的条件1与条件2成立的情况,如果cell表还未初始化或者长度为0, 
  106.             // 先尝试获取cellsBusy锁。 
  107.         }else if (cellsBusy == 0 && cells == as && casCellsBusy()) { 
  108.             boolean init = false
  109.             try {  // Initialize table 
  110.                 //初始化cells数组,初始容量为2,并将x值通过hash&1,放到0个或第1个位置上 
  111.                 if (cells == as) { 
  112.                     Cell[] rs = new Cell[2]; 
  113.                     rs[h & 1] = new Cell(x); 
  114.                     cells = rs; 
  115.                     init = true
  116.                 } 
  117.             } finally { 
  118.                 //解锁 
  119.                 cellsBusy = 0; 
  120.             } 
  121.             //如果init为true说明初始化成功,跳出循环 
  122.             if (init) 
  123.                 break; 
  124.         } 
  125.          
  126.         else if (casBase(v = base, ((fn == null) ? v + x : fn.applyAsLong(v, x))))  
  127.             // Fall back on using base 
  128.             break; 
  129.     } 

 

以上2个方法的解析搬自于源码阅读:全方位讲解LongAdder,此处对代码做了微调,方便阅读。

总结

LongAdder在没有线程竞争的时候,只使用base值,此时的情况就类似与AtomicLong。但LongAdder的高明之处在于,发生线程竞争时,便会使用到Cell数组,所以该数组是惰性加载的。

Cell数组初始值为2,每次扩容(当线程竞争异常激烈时,发生扩容)为上次长度的2倍,因此数组长度一直是2的次幂,但是当数组长度≥CPU的核心数时,就不再进行扩容。为什么?我的理解是在一台电脑中,最多能有CPU核心数个线程能够并行,因此同时也就这么多个线程操作Cell数组,每个线程更新一个位置上的元素,且又因为数组中每个元素由于字节填充机制,十分的占据内存。考虑到这两个因素,Cell数组在长度≥CPU核心数时,停止扩容。

确实,LongAdder花了很多心思提高了高并发下程序运行的效率,每一步都是在没有更好的办法下才会去选择开销更大的操作。在低并发下,LongAdder和AtomicLong效率上差不多,但LongAdder更加耗费内存。不过在高并发下,LongAdder将更加高效。

 

来源:今日头条内容投诉

免责声明:

① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的,并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据,供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。

② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

软考中级精品资料免费领

  • 历年真题答案解析
  • 备考技巧名师总结
  • 高频考点精准押题
  • 2024年上半年信息系统项目管理师第二批次真题及答案解析(完整版)

    难度     813人已做
    查看
  • 【考后总结】2024年5月26日信息系统项目管理师第2批次考情分析

    难度     354人已做
    查看
  • 【考后总结】2024年5月25日信息系统项目管理师第1批次考情分析

    难度     318人已做
    查看
  • 2024年上半年软考高项第一、二批次真题考点汇总(完整版)

    难度     435人已做
    查看
  • 2024年上半年系统架构设计师考试综合知识真题

    难度     224人已做
    查看

相关文章

发现更多好内容

猜你喜欢

AI推送时光机
位置:首页-资讯-后端开发
咦!没有更多了?去看看其它编程学习网 内容吧
首页课程
资料下载
问答资讯