atomic 包、synchronized | Java 中线程安全 - Java

提示：扫一扫查出行
【扫一扫了解最新限行尾号】
复制提示

相关阅读

彻底搞懂 CPU 中的内存结构

Java 内存模型，一篇就够了！

多线程实现原理

之前已经说过了，我们在保证线程安全的时候主要就是保证线程安全的 3 大特性，原子性、可见性、有序性、而在 JMM 的层面也已经做了相关的操作，比方说 JMM 定义的 8 种原子操作以及相关的规则，happens-before 原则。

今天主要就来看看 Java 中实现线程安全的方法之二，使用 atomic 包，synchronized 关键字。

首先说说 AtomicInteger 这个类，我们来看一个例子，计数器。实现很简单，就是每个线程都过来加 1，我们期待的结果是 999，但是若不保证线程安全，结果往往不对。

import java.util.concurrent.atomic.AtomicInteger;

public class AtomicInt {

     public static void main(String[] args) {

          for(int i=0;i < 1000;i ++){

              Thread thread = new Thread(new Counter());

              thread.start();

          }

          System.out.println(Counter.getCount());

     }

}

class Counter implements Runnable{

//   private static int count = 0;

     private static AtomicInteger count = new  AtomicInteger(0);   

     public static int getCount(){

//        return count;

          return count.get();

     }

     @Override

     public void run() {

//        count++;

          count.incrementAndGet();

     }

}

下面就来分析一下 incrementAndGet 方法的具体实现

public final int incrementAndGet() {

    return unsafe.getAndAddInt(this, valueOffset, 1)  + 1;

}

// var1 ：count 对象；

// var2 ：加数，count 中封装的整数；

// var4 ：被加数 1；

public final int getAndAddInt(Object var1, long var2, int var4) {

    int var5;

    do {

        // 该方法的主要目的取出主内存中的加数，【即为当前 count.get() 值】

        var5 = this.getIntVolatile(var1, var2);

    } while(!this.compareAndSwapInt(var1, var2, var5, var5 + var4));

    return var5;

}

public native int getIntVolatile(Object var1, long var2);

public final native boolean compareAndSwapInt(Object var1, long var2, int var4, int var5);

compareAndSwapInt 方法既是我们经常看到的 CAS 的简写，但更多的是代表了一种思想。具体在这里该方法主要目的是比较我们从主存取出的整数值和我们从 count 中传过来的是否一致，count 中的整数也就是从线程的工作内存传过来的。若一致，则计算结果并返回。若不一致，为 var5 赋值【即修改主存中的值】，并继续从主存中取出 var5 ，继续比较，直到返回。

这中间的 CAS 的思想在其它的类中也很常用，理解其核心思想即可。比较工作内存和主存中的数据，使其一致再进行计算。

与上面类似的还有 AtomicBoolean 和 AtomicLong 对象，底层的实现还是 CAS。但是在 JDK1.8 中，出现了与 AtomicLong 类似的 LongAdder 对象。

我们知道，在 CAS 的实现中，主体部分在一个 while 循环中，会一直找到工作内存和主存一致的情况，若是竞争不激烈，这是没问题的，但是当竞争非常激烈的时候，一直返回不了结果，性能就会很差。CAS 也是一种乐观锁的表现，以为可以很快的找到并返回结果。

对于普通的 long 和 double 变量，JVM 必须将 64 位的读写操作拆成 2 个 32 位读写操作。而 LongAdder 这个类的实现基于的思想就是将【热点数据分离】，比方说可以将 AtomicLong 内部的 value 分离成一个数组，不同的线程根据 hash 可以操作不同的 cell ，最后再将整个数组中所有 cell 中的数累加。

这样做的结果就相当于在 AtomicLong 的基础上，将单点的压力，分散到不同的 cell 中。在低并发的时候可以不分离热点数据，使用 base 数据, 在高并发的分离数据，这样就保证了性能。缺点是 LongAdder 在统计的时候如果有并发更新，可能导致统计的数据有误差。

我们知道在 CAS 中，我们会持续的判断内存中的数和工作内存中是否一致，以此来判断有没有其它的线程修改了工作内存中的数据，但是存在一种情况，共享变量是 1 被修改为 2 ，而后又被修改为 1 ，此时已经线程不安全了。这个问题就是常说的 ABA 问题。

Java 中提供了 AtomicStampedReference<T>，这个类主要解决的就是 CAS 中 ABA 问题，为了解决 ABA 问题，引入了一个‘变量版本号’的概念，即每次修改版本号都会加 1。使用一个变量 stamp 来记录变量版本号。

AtomicBoolean 这个类中的 compareAndSet 方法还是比较常用的。用在标识变量 flag。若是某段代码只需要执行一次可以使用这个方法来做。

public class AtomicBooleanTest {

     private static AtomicBoolean flag = new  AtomicBoolean(false);

     public static void main(String[] args) {

          // 不管有多少个线程在执行，都能保证只有会一个线程执行下面这段代码

          // 如果 flag 是 false,修改为 true

          if(flag.compareAndSet(false, true)){

              System.out.println("the flag has been  changed ~");

              // Work();

          }

     }

}

Java 中的锁有两种，一种是 synchronized 关键字，依赖于 JVM ，还有一种是代码层面的 Lock，依赖于特殊的 CPU 指令，常用的实现类有 ReentrantLock。

synchronized 可以用来修饰代码块和方法，而锁住的对象又可分为当前对象和类对象。当修饰静态方法和代码块（类.class）时为类锁。修饰一般方法和代码块（this）时为对象锁。

举例看看对象锁

public class SynTest {

     public static void main(String args[]) {

          SyncThread s = new SyncThread();

          Thread t1 = new Thread(s);

          Thread t2 = new Thread(s);

          t1.start();

          t2.start();

     }

}

class SyncThread implements Runnable {

     private static int count = 0;

     public void run() {

          // 修饰代码块，锁住当前对象:一个线程访问一个对象中的  synchronized(this) 同步代码块时，其他试图访问该对象的线程将被阻塞

          synchronized (this) {

              for (int i = 0; i < 5; i++) {

                   try {

                        System.out.println(Thread.currentThread().getName()  + ":" + (count++));

                        Thread.sleep(100);

                   } catch (InterruptedException e) {}

              }

          }

     }

}