VB.net 2010 视频教程 VB.net 2010 视频教程 python基础视频教程
SQL Server 2008 视频教程 c#入门经典教程 Visual Basic从门到精通视频教程
当前位置:
首页 > 编程开发 > Java教程 >
  • Java高并发16-LongAdder类源码解析(上)

一、复习

  • AtomicLong类

二、LongAdder源码分析

1.继承与实现关系

  • LongAdder类继承自Striped64这个类,同时实现了Serializable接口
abtract class Strped64{
    transient volatile Cell[] cells;
    transient volatile long base;
    transient volatile int cellsBusy;
    ........
}
  • 继承了三个变量,其中cells变量就是Cell实例数组,用于存放共享变量,base值是一个基础值,cellsBusy用来实现自旋锁,其值只有0和1
  • 「当创建Cell元素,扩容Cell数组或者初始化Cell数组的时候,使用CAS操作该变量来保证同时只有一个线程可以进行其中之一的操作。」

2.关键字transient

  • 讲这个关键字之前不得不提一下Seriable这个接口,我们以LongAdder这里类为例,由于它实现了序列化这个接口,也就意味着这个类可以序列化,并且保存到磁盘中,而不仅仅是只在内存中使用,然后对于有一些数据是不应该序列化的,比如密码,关键信息等等,这些数据保存到变量之后,然后被transient这个关键字修饰,那么将来序列化这个实例的时候,就不会保存这个变量,保证信息安全。

3.Cell源码

package com.ruigege.AtomicOperationClass4;
import sun.misc.Unsafe;

@sun.misc.Contended public class LongAdderTest {
 
 volatile long value;
 
 public LongAdderTest(long value) {
  this.value = value;
 }

 private static final Unsafe unsafe;
 private static final long valueOffset;
 static {
  try {
   unsafe = Unsafe.getUnsafe();
   Class<?> ak = Cell.class;
   valueOffset = unsafe.objectFieldOffset(ak.getDeclaredField("value"));
  }catch(Exception e) {
   throw new Error(e);
  }
 }
 
 final long cas(long cmp,long val) {
  return unsafe.compareAndSwapLong(this,valueOffset,cmp,val);
 }
}
  • cas函数使用了CAS操作,保证了当前线程更新时被分配的Cell元素中的value值的原子性。
 public long sum() {
  Cell[] as = cells;
  long sum = base;
  if(as != null) {
   for (int i=0;i<as.length;i++) {
    if(as[i] != null) {
     sum += as[i].value;
    }
   }
  }
 }
  • 上面这个函数就是求总值,但是由于没有加锁,所以在计算的过程中,有可能Cell实例的值变化了,所以得到的值可能不准。
 public void reset() {
  Cell[] as = cells;
  base = 0L;
  if(as != null) {
   for (int i=0;i<as.length;i++) {
    if(as[i] != null) {
     as[i].value = 0L;
    }
   }
  }
 }
  • 该函数用于重置该LongAdder实例
 public long sunThenReset() {
  Cell[] as = cells;
  int sum = base;
  if(as != null) {
   for (int i=0;i<as.length;i++) {
    if(as[i] != null) {
     sum += as[i];
     as[i] = 0L;
    }
   }
  }
 }
  • 先求和,然后Cell数组置空,但是这个方法没有加锁,容易造成数据不一致。
 public void add(long x) {
  Cell[] as;
  long b,v;
  int m;
  Cell a;
  if((as = cells) != null) || !caseBase(b=base,b+x)){
   //如果as是一个空数组,也就是第一个是false,那么会去判断第二个,第二个函数,其实就是一个CAS操作,
   //空数组就意味着没有初始化数组,这个LongAdder实例,里面就一个base变量,也就是说并发量太小了,不足以
   //初始化数组,这时这个实例就是一个AtomicLong实例,没有区别,第二个判断就是给base赋值为最新的增长量
   //如果cas失败了,那么就进入到下面的判断语句中
   boolean uncontended = true;
   if(as==null || (m = as.length-1)<0 || (a = as[getProbe() & m]) == null || !(uncontended=a.cas(v=a.value,v+x))){
    /**
     * 四个判断条件,逐步深入
     * (1)如果数组为空,继续执行判断体内;不为空,看第二个条件

     * (2)如果数组中元素为0,继续执行判断体内;不少于1个,看第三个条件
     * (3)getProbe()函数用于获取获取当前线程中变量threadLocalRandomProbe的值,这个值一开始为0,在接下
     * 来的判断体中会进行初始化,并且当前线程通过分配的Cell元素的cas函数来保证对Cell元素value值更新的原子性。

     * 这第三个就是为了找到一个数组中的Cell变量,我们暂且看到是0&m,那就是0了,也就是啊as[0]赋值给a,并且如果是空的,则直接
     * 进入到判断执行体;如果不为空,接着看第四个判断条件
     * (4)第四个判断条件就是刚才的那个a使用CAS操作来更新值,如果更新失败了就进入到执行体
     */
    longAccumulate(x,null,uncontended);
   }
  }
 }
 
 final boolean casBase(long cmp,long val) {
  return UNSAFE.compareAndSwapLong(this,BASE,cmp,val);
 }
  • 以上都是add函数,内部具体实现逻辑,
final void longAccumulate(long x,LongBinaryOperator fn,boolean wasUncontended) {
  //初始化当前线程的变量threadLocalRandomProbe的值
  int h;
  if(h = getProbe() == 0) {
   ThreadLocalRandom.current();
   h = getProbe();
   wasUncontended = true;
  }
  boolean collide = false;
  for(;;) {
   Cell[] as;Cell a;int n;long v;
   if((as=cells) != null && (n=as.length)>0) {
    if((a=as[(n-1) &h]) == null) {
     if(cellsBusy == 0) {
      Cell r = new Cell(x);
      if(cellsBusy == 0 && casCellsBusy()) {
       boolean created = false;
      }
      try {
       Cell[] rs;int m,j;
       if((rs=cells) != null && (m=rs.length)>0 && rs[j = (m-1) &h] == null) {
        rs[j] = r;
        created = true;
       }
      }finally {
       cellsBusy = 0;
      }
      if(created) {
       break;
      }
      continue;
     }
    }
    collide = false;
   }else if(!wasUntended) {
    wasUncontended = true;
   }else if(a.cas(v=a.value,((fn==null)?v+x : fn.applyAsLong(v,x)))) {
    break;
   }else if(n >= NCPU || cells != as) {
    collide = false;
   }else if(!collide) {
    collide = ture;
   }else if(cellsBusy == 0 && casCellsBusy()) {
    try {
     if(cells == as) {
      Cell[] rs = new Cell[n<<1];
      for(int i=0;i<n;++i) {
       rs[i] = as[i];
      }
      cells = rs;
     }
    }finally {
     cellsBusy = 0;
    }
   }
  }
 }

三、源码:

  • 所在包:com.ruigege.AtomicOperationClass4
  • https://github.com/ruigege66/ConcurrentJava
  • CSDN:https://blog.csdn.net/weixin_44630050
  • 博客园:https://www.cnblogs.com/ruigege0000/
出  处:https://www.cnblogs.com/ruigege0000/p/14191258.html
 

相关教程