您的位置 首页 java

Java Striped64解析

Java Striped64解析

Striped64是在java8中添加用来支持累加器的并发组件,它可以在并发环境下使用来做某种计数,Striped64的设计思路是在竞争激烈的是候尽量分散竞争,在实现上,Striped64维护了一个base Count和一个Cell数组,计数线程会首先试图更新base变量,如果成功则退出计数,否则会认为当前竞争是很激烈的,那么就会通过Cell数组来分散计数,Striped64根据线程来计算哈希,然后将不同的线程分散到不同的Cell数组的index上,然后这个线程的计数内容就会保存在该Cell的位置上面,基于这种设计,最后的总计数需要结合base以及散落在Cell数组中的计数内容。这种设计思路类似于java7的ConcurrentHashMap实现,也就是所谓的分段锁算法,ConcurrentHashMap会将记录根据key的hashCode来分散到不同的segment上,线程相要操作某个记录只需要锁住这个记录对应着的segment就可以了,而其他segment并不会被锁住,其他线程仍然可以去操作其他的segment,这样就显著提高了并发度,虽然如此,java8中的ConcurrentHashMap实现已经抛弃了java7中分段锁的设计,而采用更为轻量级的CAS来协调并发,效率更佳。

虽然Striped64的设计类似于分段锁算法,但是仍然有其独到之处,本文将分析Striped64的实现细节,并且会分析基于Striped64的计数类LongAdder。Striped64的实现还是较为复杂的,本文会尽量分析,对于没有充分了解的内容,或者分析有误的内容,会在未来不断修改补充。

下面首先展示了Striped64中的Cell类:

Cell类中仅有一个保存计数的变量value,并且为该变量提供了CAS操作方法,Cell类的实现虽然看起来很简单,但是它的作用是非常大的,它是Striped64实现分散计数的最为基础的数据结构,当然为了达到并发环境下的线程安全以及高效,Striped64做了很多努力。Striped64中有两个提供计数的api方法,分别为longAccumulate和doubleAccumulate,两者的实现思路是一致的,只是前者对long类型计数,而后者对double类型计数,本文只分析前者的实现,下面是longAccumulate方法的代码:

 
final void longAccumulate(long x, LongBinaryOperator fn,
                              boolean wasUncontended) {
        int h;
        if ((h = getProbe()) == 0) { //获取当前线程的probe值,如果为0,则需要初始化该线程的probe值
            ThreadLocalRandom.current(); // force initialization
            h = getProbe();
            wasUncontended = true;
        }
        boolean collide = false;                // True if last slot nonempty
        for (;;) {
            Cell[] as; Cell a; int n; long v;
            if ((as = cells) != null && (n = as.length) > 0) { //获取cell数组
                if ((a = as[(n - 1) & h]) == null) { // 通过(hashCode & (length - 1))这种算法来实现取模
                    if (cellsBusy == 0) {       // 如果当前位置为null说明需要初始化
                        Cell r = new Cell(x);   // Optimistically create
                        if (cellsBusy == 0 && casCellsBusy()) {
                            boolean created = false;
                            try {               // Recheck under lock
                                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;           // Slot is now non-empty
                        }
                    }
                    collide = false;
                } 
                //运行到此说明cell的对应位置上已经有想相应的Cell了,不需要初始化了
                else if (!wasUncontended)       // CAS already known to fail
                    wasUncontended = true;      // Continue after rehash
                    
                //尝试去修改a上的计数,a为Cell数组中index位置上的cell
                else if (a.cas(v = a.value, ((fn == null) ? v + x :
                                             fn.applyAsLong(v, x))))
                    break;
                    
                //cell数组最大为cpu的数量,cells != as表面cells数组已经被更新了    
                else if (n >= NCPU || cells != as)
                    collide = false;            // At max size or stale
                else if (!collide)
                    collide = true;
                else if (cellsBusy == 0 && casCellsBusy()) {
                    try {
                        if (cells == as) {      // Expand table unless stale
                            Cell[] rs = new Cell[n << 1]; //Cell数组扩容,每次扩容为原来的两倍
                            for (int i = 0; i < n; ++i)
                                rs[i] = as[i];
                            cells = rs;
                        }
                    } finally {
                        cellsBusy = 0;
                    }
                    collide = false;
                    continue;                   // Retry with expanded table
                }
                h = advanceProbe(h);
            }
            else if (cellsBusy == 0 && cells == as && casCellsBusy()) {
                boolean init = false;
                try {                           // Initialize table
                    if (cells == as) {
                        Cell[] rs = new Cell[2];
                        rs[h & 1] = new Cell(x);
                        cells = rs;
                        init = true;
                    }
                } finally {
                    cellsBusy = 0;
                }
                if (init)
                    break;
            }
            else if (casBase(v = base, ((fn == null) ? v + x :
                                        fn.applyAsLong(v, x))))
                break;                          // Fall back on using base
        }
    }

  

仅从代码量上就可以意识到longAccumulate的实现时异常复杂的,下面来梳理一下该方法的运行逻辑:

  • longAccumulate会根据当前线程来计算一个哈希值,然后根据算法(hashCode & (length – 1))来达到取模的效果以定位到该线程被分散到的Cell数组中的位置
  • 如果Cell数组还没有被创建,那么就去获取cellBusy这个共享变量(相当于锁,但是更为轻量级),如果获取成功,则初始化Cell数组,初始容量为2,初始化完成之后将x保证成一个Cell,哈希计算之后分散到相应的index上。如果获取cellBusy失败,那么会试图将x累计到base上,更新失败会重新尝试直到成功。
  • 如果Cell数组以经被初始化过了,那么就根据线程的哈希值分散到一个Cell数组元素上,获取这个位置上的Cell并且赋值给变量a,这个a很重要,如果a为null,说明该位置还没有被初始化,那么就初始化,当然在初始化之前需要竞争cellBusy变量。
  • 如果Cell数组的大小已经最大了(CPU的数量),那么就需要重新计算哈希,来重新分散当前线程到另外一个Cell位置上再走一遍该方法的逻辑,否则就需要对Cell数组进行扩容,然后将原来的计数内容迁移过去。这里面需要注意的是,因为Cell里面保存的是计数值,所以在扩容之后没有必要做其他的处理,直接根据index将旧的Cell数组内容直接复制到新的Cell数组中就可以了。

当然,上面的流程是高度概括的,longAccumulate的实际分支还要更多,并且为了保证线程安全做的判断更多。longAccumulate会根据不同的状态来执行不同的分支,比如在线程竞争非常激烈的时候,会通过对cells数组扩容或者重新计算哈希值来重新分散线程,这些做法的目的是将多个线程的计数请求分散到不同的cells的index上,其实这和java7中的ConcurrentHashMap的设计思路是完全一致的,但是java7中的ConcurrentHashMap实现在segment加锁使用了比较重的synchronized,而Striped64使用了java中较为底层的Unsafe类的CAS操作来进行并发操作,这种方式更为轻量级,因为它会不停的尝试,失败会返回,而加锁的方式会阻塞线程,线程需要被唤醒,这涉及到了线程的状态的改变,需要上下文切换,所以是比较重量级的。

Unsafe

在这里添加一点关于java中底层操作的类Unsafe类的使用方法,首先看下面的代码:

Unsafe需要关注的是Field的offset,然后在CAS的时候需要oldValue和expectValue以及newValue,它会在比较了oldValue == exceptValue的时候将oldValue设置为newValue,否则不会改变。这也是CAS的定义,(compare And set)下面的代码展示了CAS操作的示例:

 
UNSAFE.compareAndSwapLong(this, valueOffset, cmp, val)

this是需要改变的对象,valueOffset为需要修改的Field在该对象中的offset,这个值的获取可以参考上面展示的
图片,cmp为exceptValue,也就是我们希望他的旧值为cmp值,如果相等,则将该Field设置为val,否则别修改。

  

LongAdder实现细节

上文中分析了Striped64的实现细节,下面来分析一下LongAdder的实现细节,LongAdder的实现基于Striped64,理解了Striped64就很好理解LongAdder了。下面先来看一下LongAdder的add方法:

首先判断cells是否为null,如果为null,则会尝试将本次计数累计到base上,如果cells不为null,或者操作base失败,那么就会通过哈希值来获取当前线程对应的cells数组中的位置,获取该位置上的cell,如果该cell不为null,那么就试图将本次计数累计到该cell上,如果不成功,那么就需要借助Striped64类的longAccumulate方法来进行计数累计,关于longAccumulate的分析见上文。

当我们想要获得当前的总计数的时候,需要调用sum方法来获取,下面展示了该方法的细节:

它需要累计base和Cell数组中的Cell中的计数,base中的计数为线程竞争不是很激烈的时候累计的数,而在线程竞争比较激烈的时候就会将计数的任务分散到Cell数组中,所以在sum方法里,需要合并两处的计数值。

除了获取总计数,我们有时候想reset一下,下面的代码展示了这种操作:

 
    public void reset() {
        Cell[] as = cells; Cell a;
        base = 0L;
        if (as != null) {
            for (int i = 0; i < as.length; ++i) {
                if ((a = as[i]) != null)
                    a.value = 0L;
            }
        }
    }

  

同样注意点在于需要同时将base和Cell数组都reset。

Striped64在ConcurrentHashMap中的使用

Striped64的计数方法在java8的ConcurrentHashMap中也有使用,具体的实现细节可以参考addCount方法,下面来看一下ConcurrentHashMap的size方法的实现细节:

 
    public int size() {
        long n = sumCount();
        return ((n < 0L) ? 0 :
                (n > (long)Integer.MAX_VALUE) ? Integer.MAX_VALUE :
                (int)n);
    }

    final long sumCount() {
        CounterCell[] as = counterCells; CounterCell a;
        long sum = baseCount;
        if (as != null) {
            for (int i = 0; i < as.length; ++i) {
                if ((a = as[i]) != null)
                    sum += a.value;
            }
        }
        return sum;
    }
  

ConcurrentHashMap中的baseCount对应着Striped64中的base变量,而counterCells则对应着Striped64中的cells数组,他们的实现是一样的,更为详细的内容可以参考java8中的ConcurrentHashMap实现。

文章来源:智云一二三科技

文章标题:Java Striped64解析

文章地址:https://www.zhihuclub.com/171730.shtml

关于作者: 智云科技

热门文章

网站地图