并发_Atomic 原子类

Atomic 原子类

简介

当程序更新一个变量时,如果多线程同时更新这个变量,可能得到期望之外的值。

通常我们会使用synchronized来解决这个问题,synchronized会保证多线程不会同时更新变量。但是synchronized的开销比较大,执行效率较低。

JDK 1.5开始提供了java.util.concurrent.atomic包,这个包中的原子操作类提供了一种用法简单、性能高效、线程安全地更新一个变量的方式。

主要有四类原子更新方式:

  1. 原子更新基本类型
  2. 原子更新数组
  3. 原子更新引用
  4. 原子更新属性(字段)

Atomic包里的类基本都是使用Unsafe实现的包装类。

学习Unsafehttps://tech.meituan.com/2019/02/14/talk-about-java-magic-class-unsafe.html

原子更新基本类型

AtomicLong

AtomicBoolean

AtomicInteger

AtomicInteger 的方法

方法 描述
public final int get() 获取当前的值
public final void set(int newValue) 设置为给定值
public final void lazySet(int newValue) 最终设置为给定值,设置后可能读到旧值(unsafe.putOrderedInt)
public final int getAndSet(int newValue) 原子方式设置为给定值,并返回旧值(unsafe.getAndSetInt)
public final boolean compareAndSet(int expect, int update) 原子方式将该值设置为给定的更新值,如果当前值==预期值(unsafe.compareAndSwapInt)
public final int getAndIncrement() 获取当前的值,并自增1
public final int getAndDecrement() 获取当前的值,并自减1
public final int getAndAdd(int delta) 获取当前的值,并加上给定的值
public final int incrementAndGet() 自增1,并获取新值
public final int decrementAndGet() 自减1,并获取新值
public final int addAndGet(int delta) 加上给定的值,并获取新值
public final int getAndUpdate(IntUnaryOperator updateFunction) 获取当前的值,并设置根据函数得到的新值
public final int updateAndGet(IntUnaryOperator updateFunction) 设置根据函数得到的新值,并得到新值

使用示例

public class AtomicIntegerTest {

   private static final AtomicInteger atomicInteger = new AtomicInteger();
   private static final CountDownLatch countDownLatch = new CountDownLatch(10);

   public static void main(String[] args) throws InterruptedException {
      for (int i = 0; i < 10; i++) {
         new Thread(() -> {
            atomicInteger.getAndIncrement();
            countDownLatch.countDown();
         }).start();
      }
      countDownLatch.await();
      System.out.println(atomicInteger.get());
   }

}

/*
返回值:10
*/

主要源码

主要利用了CAS的操作、使用volitileUnsafenative方法来保证了变量的原子操作。

// setup to use Unsafe.compareAndSwapInt for updates
private static final Unsafe unsafe = Unsafe.getUnsafe();
private static final long valueOffset;

static {
    try {
        valueOffset = unsafe.objectFieldOffset
            (AtomicInteger.class.getDeclaredField("value"));
    } catch (Exception ex) { throw new Error(ex); }
}

private volatile int value;

原子更新数组

AtomicIntegerArray

AtomicLongArray

AtomicReferenceArray

AtomicIntegerArray 方法

基本类似于AtomicInteger,会多个参数int i 来确定位置。

方法 描述
public final void set(int i, int newValue) 设置在位置i的元素为给定值。
public final int getAndSet(int i, int newValue) 以原子方式设置在位置i的元素为给定值,并返回原来的值。
public final boolean compareAndSet(int i, int expect, int update) 原子方式将位置i的元素设置为给定的更新值,如果当前值==预期值

使用示例

public class AtomicIntegerArrayTest {

   private static final int[] value = {1, 2, 3, 4};
   private static final AtomicIntegerArray atomicIntegerArray = new AtomicIntegerArray(value);

   public static void main(String[] args) {
      for (int i = 0; i < 4; i++) {
         atomicIntegerArray.addAndGet(i, 5);
      }
      System.out.println(Arrays.toString(value));
      System.out.println(atomicIntegerArray.toString());
   }

}
/*
[1, 2, 3, 4]
[6, 7, 8, 9]
*/

源码

数组value通过构造方法传递进去,然后AtomicIntegerArray会复制一份,所以当AtomicIntegerArray对其数组内部进行修改时,不会影响传入的数组。

private static final Unsafe unsafe = Unsafe.getUnsafe();
private static final int base = unsafe.arrayBaseOffset(int[].class);
private static final int shift;
private final int[] array;

static {
    int scale = unsafe.arrayIndexScale(int[].class);
    if ((scale & (scale - 1)) != 0)
        throw new Error("data type scale not a power of two");
    shift = 31 - Integer.numberOfLeadingZeros(scale);
}

public AtomicIntegerArray(int length) {
    array = new int[length];
}

public AtomicIntegerArray(int[] array) {
    // Visibility guaranteed by final field guarantees
    this.array = array.clone();
}

原子更新引用

AtomicReference

AtomicMarkableReference

AtomicStampedReference

AtomicReference 方法

方法 描述
public final V getAndUpdate(UnaryOperator<V> updateFunction) 根据给定的函数原子更新值,并返回当前值。<br />如果多个线程竞争失败会重试。(compareAndSet)
public final V updateAndGet(UnaryOperator<V> updateFunction) 根据给定的函数原子更新值,并返回新值。

使用示例

public class AtomicReferenceTest {

   private static final AtomicReference<User> atomicReference = new AtomicReference<>();

   public static void main(String[] args) {
      User user = new User("张三",16);
      atomicReference.set(user);
      User newUser = new User("李四",18);
      atomicReference.compareAndSet(user, newUser);
      System.out.println(atomicReference.get());
   }

   @Data
   @AllArgsConstructor
   static class User {
      private String name;
      private int age;
   }

}

源码

private static final Unsafe unsafe = Unsafe.getUnsafe();
private static final long valueOffset;

static {
    try {
        valueOffset = unsafe.objectFieldOffset
            (AtomicReference.class.getDeclaredField("value"));
    } catch (Exception ex) { throw new Error(ex); }
}

private volatile V value;

原子更新属性(字段)

AtomicIntegerFieldUpdater

AtomicReferenceFieldUpdater

AtomicStampedReference

AtomicIntegerFieldUpdater 简单介绍

  1. 原子更新带有版本号的引用类都是抽象类,每次使用的时候用静态方法newUpdater来创建,设置tclassfieldName

    @CallerSensitive:通过使用一种机制来准确地识别此类方法,并允许可靠地发现其调用者,从而替换了现有的手动维护的调用者敏感方法列表,从而提高了JDK方法处理实现的安全性。

    jvm的开发者认为这些方法危险,不希望开发者调用,就把这种危险的方法用 @CallerSensitive修饰,并在jvm级别检查。

    http://openjdk.java.net/jeps/176

@CallerSensitive
public static <U> AtomicIntegerFieldUpdater<U> newUpdater(Class<U> tclass, String fieldName) {
    return new AtomicIntegerFieldUpdaterImpl<U>(tclass, fieldName, Reflection.getCallerClass());
}
  1. 更新的字段必须是public volitile修饰的,不然会抛出异常。
java.lang.RuntimeException: java.lang.IllegalAccessException: Class xxx.AtomicIntegerFieldUpdateTest can not access a member of class xxx.AtomicIntegerFieldUpdateTest$User with modifiers "private"
    
java.lang.ExceptionInInitializerError Caused by: java.lang.IllegalArgumentException: Must be volatile type
//源码: 
if (!Modifier.isVolatile(modifiers))  throw new IllegalArgumentException("Must be volatile type");

使用示例

public class AtomicIntegerFieldUpdateTest {

    private static final AtomicIntegerFieldUpdater<User> atomicIntegerFieldUpdater =
            AtomicIntegerFieldUpdater.newUpdater(User.class, "age");

    public static void main(String[] args) {
        User user = new User("张三",16);
        atomicIntegerFieldUpdater.set(user, 21);
        System.out.println(atomicIntegerFieldUpdater.get(user));
    }

    @Data
    @AllArgsConstructor
    static class User {
        private String name;
        public volatile int age;
    }

}

JDK 8 新增的原子操作类 LongAndder

为了解决高并发下多线程对一个变量CAS争夺失败后进行自旋而造成的降低开发性能问题,LongAdder在内部维护多个Cell元素(一个动态地Cell数组)来分担对单个变量进行争夺的开销。

LongAdder 的结构

LongAdder类继承自Striped,在Striped内部维护着三个变量。

  • base:基础值,默认为0。
  • cellsBusy:状态值,只有0和1,当创建Cell元素、扩容Cell数组或者初始化Cell数组时,使用CAS操作该变量来保证同时只有一个线程可以进行其中之一的操作。
  • cells: Table of cells. When non-null, size is a power of 2. 存储所有的Cell元素,初始值为null,大小是2的次幂。

当前线程访问 Cell 数组里面的那个元素

通过add方法中getProbe() & m来确定当前线程使用哪个元素。

getProbe获取当前线程中变量threadLocalRandomProbe的值,初始值为0,在longAccumulate中 初始化。

longAccumulatecells数组被初始化和扩容的地方。

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);
    }
}

final void longAccumulate(long x, LongBinaryOperator fn,
                          boolean wasUncontended) {
   int h;
   if ((h = getProbe()) == 0) {
      ThreadLocalRandom.current(); // 初始化 threadLocalRandomProbe
      h = getProbe();
      wasUncontended = true;
   }
   // ......
}

如何初始化 Cell 数组

cellsBusy == 0说明当前cells数组没有被初始化、扩容或者新建。

初始大小为2。

final void longAccumulate(long x, LongBinaryOperator fn,  boolean wasUncontended) {
   int h = getProbe();
   for (; ; ) {
      Cell[] as; Cell a; int n; long v;
      if ((as = cells) != null && (n = as.length) > 0) {
        // ......
      } 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
      }
   }
}

Cell 数组如何扩容

如果当前元素个数没有达到CPU个数并且有冲突则扩容。

将容量扩充为之前的两倍,并复制到新的数组中。

final void longAccumulate(long x, LongBinaryOperator fn,
                          boolean wasUncontended) {
   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) {
        // 如果当前元素个数没有达到CPU个数并且有冲突则扩容
        if (cellsBusy == 0 && casCellsBusy()) {
            try {
               if (cells == as) {      // Expand table unless stale
                  Cell[] rs = new Cell[n << 1];
                  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);
      }              
   }
}

处理线程访问分配的 Cell 元素的冲突问题

longAccumulate方法中使用advanceProbe(h),重新计算当前线程的随机数threadLocalRandomProbe,以减少下次访问cells元素的冲突。

static final int advanceProbe(int probe) {
    probe ^= probe << 13;   // xorshift
    probe ^= probe >>> 17;
    probe ^= probe << 5;
    UNSAFE.putInt(Thread.currentThread(), PROBE, probe);
    return probe;
}

如何保证线程操作被分配的 Cell 元素的原子性

为了保证变量的可见性将value声明为volatile

使用CAS保证了value的原子性。

@sun.misc.Contended为了避免伪共享。

@sun.misc.Contended static final class Cell {
    volatile long value;
    Cell(long x) { value = x; }
    final boolean cas(long cmp, long val) {
        return UNSAFE.compareAndSwapLong(this, valueOffset, cmp, val);
    }

    // Unsafe mechanics
    private static final sun.misc.Unsafe UNSAFE;
    private static final long valueOffset;
    static {
        try {
            UNSAFE = sun.misc.Unsafe.getUnsafe();
            Class<?> ak = Cell.class;
            valueOffset = UNSAFE.objectFieldOffset
                (ak.getDeclaredField("value"));
        } catch (Exception e) {
            throw new Error(e);
        }
    }
}
©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 203,324评论 5 476
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 85,303评论 2 381
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 150,192评论 0 337
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 54,555评论 1 273
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 63,569评论 5 365
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 48,566评论 1 281
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 37,927评论 3 395
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 36,583评论 0 257
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 40,827评论 1 297
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,590评论 2 320
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,669评论 1 329
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 33,365评论 4 318
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 38,941评论 3 307
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 29,928评论 0 19
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,159评论 1 259
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 42,880评论 2 349
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 42,399评论 2 342