526互联

atomic原子类

发布时间 2023-04-08 14:29:21作者: 李勇888

原子类介绍

  • java.util.concurrent.atomic
    • java并发包下的类,用于多线程情况下保证线程安全的

API

  • 基本类型原子类
    • AtomicInteger
  • 数组类型原子类
    • AtomicIntegerArray
  • 引用类型原子类
    • AtomicReference
  • 对象的属性修改原子类
    • AtomicIntegerFieldUpdate
  • 原子操作增强类
    • LongAdder
    • LongAccumulator

原子类高并发场景下效率对比

  • LongAccumulator > LongAdder > AtomicInteger > synchronized
    • 代码
      class Number {
    int number = 0;

    // synchronized
    public synchronized void clickBySynchronized(){
        number ++;
    }

    // AtomicInteger
    AtomicInteger atomicInteger = new AtomicInteger(0);
    public void clickAtomic(){
        atomicInteger.incrementAndGet();
    }

    // LongAdder
    LongAdder longAdder = new LongAdder();
    public void clickLongAdder(){
        longAdder.increment();
    }

    // LongAccumulator
    LongAccumulator longAccumulator = new LongAccumulator((x, y) -> x + y, 0);
    public void clickLongAccumulator(){
        longAccumulator.accumulate(1);
    }
}

/**
 * 50个线程，每个线程点赞100万次 性能对比
 *
 * result
 *
 * synchronized      6361毫秒
 * atomic            898毫秒
 * longAdder         187毫秒
 * longAccumulator   172毫秒
 *
 */
public class 点赞量统计性能对比 {

    private static final int _w = 10000;
    private static final int size = 100;

    public static void main(String[] args) {

        Number number = new Number();
        CountDownLatch countDownLatch = new CountDownLatch(size);

        long startTime = System.currentTimeMillis();
        for (int i = 1;i <= size;i++){
            new Thread(() -> {
                for (int k = 0;k < _w * 100;k++){
                    number.clickLongAdder();
                }

                countDownLatch.countDown();
            }).start();
        }
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }

        System.out.println("result: " + number.number);
        long endTime = System.currentTimeMillis();
        System.out.println("时间：" + (endTime - startTime) + "毫秒");
    }
}

LongAdder底层原理

  • LongAdder在无竞争的情况下,跟AtomicLong一样,对同一个base进行操作,当出现竞争关系是采用 化整为零分散热点的做法,用空间换时间,用一个数组cells,将value拆分成进入cells数组. 多个线程需要同时对value进行操作时,对线程id进行hash得到hash值,再根据hash值映射到这个数组cells的某个下标,再对下标所对应的值进行自增操作. 当所有线程操作完毕,将数组cells 和base 都加起来作为 最终结果

  • 源码分析 add方法

    public void add(long x) {
    Cell[] as; long b, v; int m; Cell a;
    if ((as = cells) != null || !casBase(b = base, b + x)) {
        boolean uncontended = true;
        if (as == null || (m = as.length - 1) < 0 ||
            (a = as[getProbe() & m]) == null ||
            !(uncontended = a.cas(v = a.value, v + x)))
            longAccumulate(x, null, uncontended);
    }
}

    • 如果cells 数组为空,尝试用CAS更新base字段,成功则退出,反之更新base失败了,说明出现了线程竞争, uncontended为true,调用longAccumulate

    • 如果cells数组 非空,且当前线程映射的槽位为空,uncontended为true,调用longAccumulate

    • 如果cells数组 非空,且当前线程映射的槽位非空,CAS更新cell的值,成功则返回,否则uncontended设为false,调用longAccumulate

  • sum方法

    • sum()会将所有cell数组中的value和base累加作为返回值

  • 总结

    • 核心思想就是将之前AtomicLong 中的value 的更新压力分散到多个value中去,从而降级更新热点
    • LongAdder不是强一致性,是最终一致性