一、无锁解决线程安全问题

1.引例

原子整数使用：

  1. AtomicInteger balance = new AtomicInteger();
  1. package panw.unlocked;
  3. import lombok.extern.slf4j.Slf4j;
  5. import java.util.ArrayList;
  6. import java.util.concurrent.atomic.AtomicInteger;
  8. public class AcountTest {
  9.     public static void main(String[] args) {
  10.         UnsafeAccount unsafeAccount = new UnsafeAccount(10000);
  11.         Account.demo(unsafeAccount);
  12. //        SafeAccount safeAccount = new SafeAccount(10000);
  13. //        Account.demo(safeAccount);
  14.     }
  15. }
  16. interface Account{
  17.     Integer getBalance();
  18.     void withDraw(Integer amount);
  20.     static void demo(Account account){
  22.         ArrayList<Thread> list = new ArrayList<>();
  23.         long start = System.nanoTime();
  24.         for (int i = 0; i < 1000; i++) {
  25.             list.add(new Thread(()->{
  26.                 account.withDraw(10);
  27.             }));
  28.         }
  29.         list.forEach(Thread::start);
  30.         list.forEach((t)->{
  31.             try {
  32.                 t.join();
  33.             } catch (InterruptedException e) {
  34.                 throw new RuntimeException(e);
  35.             }
  36.         });
  37.         long end = System.nanoTime();
  38.         System.out.println(account.getBalance() + " cost: " + (end - start) / 1000_000 + " ms");
  39.     }
  40. }
  41. class UnsafeAccount implements Account{
  42.     private Integer balance;
  44.     public UnsafeAccount(Integer balance) {
  45.         this.balance = balance;
  46.     }
  48.     @Override
  49.     public Integer getBalance() {
  50.         return this.balance;
  51.     }
  53.     @Override
  54.     public void withDraw(Integer amount) {
  55.         synchronized (this){
  56.             this.balance -= amount;
  57.         }
  59.     }
  60. }
  61. class SafeAccount implements Account{
  63.     private AtomicInteger balance;
  65.     public SafeAccount(int balance) {
  66.         this.balance = new AtomicInteger(balance);
  67.     }
  69.     @Override
  70.     public Integer getBalance() {
  71.         return balance.get();
  72.     }
  74.     @Override
  75.     public void withDraw(Integer amount) {
  77. //        balance.getAndAdd(-amount);
  78.         while (true){
  79.             int prev  = balance.get();
  80.             int next = prev - amount;
  81.             if (balance.compareAndSet(prev,next)){
  82.                 break;
  83.             }
  84.         }
  87.     }
  88. }

以上的代码UnsafeAccount只有加上了synchronized代码块才能保证操作的原子性，但是这样相比于CAS（compare and swap）效率查了很多，那么CAS到底是什么呢？下面我们来看看

二、CAS与volatile

CAS是原子操作的一种，可用于在多线程编程中实现不被打断的数据交换操作，从而避免多线程同时改写某一数据时由于执行顺序不确定性以及中断的不可预知性产生的数据不一致问题。

1.三个操作参数

  1. 内存位置V（它的值是我们想要去更新的）
  2. 预期原值A（上一次从内存中读取的值）
  3. 新值B（应该写入的新值）

    2.CAS的操作过程

    将内存位置V的值与A比较（compare）
  • 如果相等，则说明没有其它线程来修改过这个值，所以把内存V的的值更新成B（swap）

  • 如果不相等，说明V上的值被修改过了，不更新，而是返回当前V的值，再重新执行一次任务再继续这个过程。

    3.volatile

    volatile 仅仅保证了共享变量的可见性，让其它线程能够看到新值，但不能解决指令交错问题（不能保证原子性）
    CAS 必须借助 volatile 才能读取到共享变量的新值来实现【比较并交换】的效果

    4.CAS特点

    结合 CAS 和 volatile 可以实现无锁并发，适用于线程数少、多核 CPU 的场景下。

  • CAS 是基于乐观锁的思想：乐观的估计，不怕别的线程来修改共享变量，就算改了也没关系，我再重试。

  • synchronized 是基于悲观锁的思想：悲观的估计，得防着其它线程来修改共享变量，我上了锁你们都别想改，我改完了解开锁，你们才有机会。

  • CAS 体现的是无锁并发、无阻塞并发

    • 因为没有使用 synchronized，所以线程不会陷入阻塞，这是效率提升的因素之一
    • 但如果竞争激烈，可以想到重试必然频繁发生，反而效率会受影响

      三、原子整数

      JUC并发包里面提供了

  • AtomicBoolean

  • AtomicInteger
  • AtomicLong

具体使用差不多，我们以AtomicInteger举例

AtomicInteger

  1. package panw.JUC;
  3. import java.util.concurrent.atomic.AtomicInteger;
  5. public class AtomicIntegerTest {
  6.     public static void main(String[] args) {
  7.         AtomicInteger ai = new AtomicInteger(1);
  8.         ai.getAndIncrement();
  9.         ai.incrementAndGet();
  10.         ai.getAndDecrement();
  11.         ai.decrementAndGet();
  12.         ai.getAndAdd(5);
  13.         ai.getAndAdd(-5);
  14.         ai.getAndUpdate((a)->a*10);
  15.         for (int i = 0; i < 3; i++) {
  17.             int andAccumulate = ai.getAndAccumulate(5, (a, b) -> a * b);
  18.             System.out.println(andAccumulate);
  19.         }
  21.     }
  22. }

四、原子引用

AtomicReference和AtomicInteger非常类似，不同之处就在于AtomicInteger是对整数的封装，而AtomicReference则对应普通的对象引用。也就是它可以保证你在修改对象引用时的线程安全性。

  1. package panw.JUC;
  3. import java.math.BigDecimal;
  4. import java.util.ArrayList;
  5. import java.util.concurrent.atomic.AtomicReference;
  7. public class AtomicReferenceTest {
  8.     public static void main(String[] args) {
  9.         SafeAccount safeAccount = new SafeAccount(new BigDecimal(10000));
  10.         Account.demo(safeAccount);
  11.     }
  12. }
  14. interface Account {
  15.     BigDecimal getBalance();
  17.     void withDraw(BigDecimal amount);
  19.     static void demo(Account account) {
  21.         ArrayList<Thread> list = new ArrayList<>();
  22.         long start = System.nanoTime();
  23.         for (int i = 0; i < 1000; i++) {
  24.             list.add(new Thread(() -> {
  25.                 account.withDraw(BigDecimal.TEN);
  26.             }));
  27.         }
  28.         list.forEach(Thread::start);
  29.         list.forEach((t) -> {
  30.             try {
  31.                 t.join();
  32.             } catch (InterruptedException e) {
  33.                 throw new RuntimeException(e);
  34.             }
  35.         });
  36.         long end = System.nanoTime();
  37.         System.out.println(account.getBalance() + " cost: " + (end - start) / 1000_000 + " ms");
  38.     }
  39. }
  40. class SafeAccount implements Account {
  42.     private AtomicReference<BigDecimal> balance;
  44.     public SafeAccount(BigDecimal balance) {
  45.         this.balance = new AtomicReference<BigDecimal>(balance);
  46.     }
  48.     @Override
  49.     public BigDecimal getBalance() {
  50.         return balance.get();
  51.     }
  53.     @Override
  54.     public void withDraw(BigDecimal amount) {
  56. //        balance.getAndAdd(-amount);
  57.         while (true){
  58.             BigDecimal prev  = balance.get();
  59.             BigDecimal next = prev.subtract(amount);
  60.             if (balance.compareAndSet(prev,next)){
  61.                 break;
  62.             }
  63.         }
  66.     }
  67. }

五、ABA问题

1.示例

  1. package panw.JUC;
  3. import lombok.extern.slf4j.Slf4j;
  5. import java.util.concurrent.atomic.AtomicReference;
  6. @Slf4j
  7. public class ABATest {
  8.     static AtomicReference<String> atomicReference = new AtomicReference<>("A");
  9.     public static void main(String[] args) throws InterruptedException {
  11.         Thread t1 = new Thread(() -> {
  12.             String s = atomicReference.get();
  14.             other();
  15.             try {
  16.                 Thread.sleep(1000);
  17.             } catch (InterruptedException e) {
  18.                 throw new RuntimeException(e);
  19.             }
  20.             log.debug("A->C  "+ atomicReference.compareAndSet("A","C"));
  21.         }, "t1");
  22.         t1.start();
  23.         t1.join();
  24.         log.debug("最终："+atomicReference.get());
  26.     }
  27.     static void other()  {
  28.         new Thread(()->{
  29.             log.debug("A->B  "+ atomicReference.compareAndSet("A","B"));
  30.         }).start();
  31.         try {
  32.             Thread.sleep(500);
  33.         } catch (InterruptedException e) {
  34.             throw new RuntimeException(e);
  35.         }
  36.         new Thread(()->{
  37.             log.debug("B->A  "+ atomicReference.compareAndSet("B","A"));
  38.         }).start();
  39.     }
  40. }

主线程仅能判断出共享变量的值与初值 A 是否相同，不能感知到这种从 A 改为 B 又 改回 A 的情况，如果主线程希望：
只要有其它线程【动过了】共享变量，那么自己的 cas 就算失败，这时，仅比较值是不够的，需要再加一个版本号，JUC提供了以下的实现

2.AtomicStampedReference

  1. package panw.JUC;
  3. import lombok.extern.slf4j.Slf4j;
  5. import java.util.concurrent.TimeUnit;
  6. import java.util.concurrent.atomic.AtomicStampedReference;
  7. @Slf4j
  8. public class AtomicStampedReferenceTest {
  9.     private static AtomicStampedReference<String> str = new AtomicStampedReference<>("A",0);
  10.     public static void main(String[] args) throws InterruptedException {
  11.         Thread t1 = new Thread(() -> {
  12.             String pre = str.getReference();
  13.             int stamp = str.getStamp();
  15.             other();
  16.             try {
  17.                 TimeUnit.SECONDS.sleep(2);
  18.             } catch (InterruptedException e) {
  19.                 throw new RuntimeException(e);
  20.             }
  22.             log.debug("change A->C stamp " + stamp + str.compareAndSet(pre, "C", stamp, stamp+1));
  23.         });
  24.         t1.start();
  25.         t1.join();
  26.     }
  27.     static void other()  {
  28.         new Thread(()->{
  29.             int stamp = str.getStamp();
  30.             log.debug("A->B stamp " + stamp + str.compareAndSet("A","B",stamp,stamp+1));
  31.         }).start();
  32.         try {
  33.             Thread.sleep(500);
  34.         } catch (InterruptedException e) {
  35.             throw new RuntimeException(e);
  36.         }
  37.         new Thread(()->{
  38.             int stamp = str.getStamp();
  39.             log.debug("B->A  stamp " + stamp + str.compareAndSet("B","A",stamp,stamp+1));
  40.         }).start();
  41.     }
  42. }

3.AtomicMarkableReference

  1. package panw.JUC;
  3. import lombok.extern.slf4j.Slf4j;
  5. import java.util.concurrent.atomic.AtomicMarkableReference;
  6. @Slf4j
  7. public class AtomicMarkableReferenceTest {
  8.     static AtomicMarkableReference<String> str = new AtomicMarkableReference<>("A",false);
  9.     public static void main(String[] args) {
  10.         new Thread(()->{
  11.             String pre = str.getReference();
  12.             other();
  13.             try {
  14.                 Thread.sleep(1000);
  15.             } catch (InterruptedException e) {
  16.                 throw new RuntimeException(e);
  17.             }
  18.             log.debug("A->C  " + str.compareAndSet("A","C",false,true));
  19.         }).start();
  21.     }
  22.     static void other()  {
  23.         new Thread(()->{
  25.             log.debug("A->A  " + str.compareAndSet("A","A",false,true));
  26.         }).start();
  28.     }
  29. }

两种区别

  • AtomicStampedReference 需要我们传入整型变量作为版本号，来判定是否被更改过

  • AtomicMarkableReference需要我们传入布尔变量作为标记，来判断是否被更改过

    六、原子数组

  • AtomicIntegerArray

  • AtomicLongArray
  • AtomicReferenceArray

具体的使用其实和上面相同，只是操作时候带上数据下标即可。

七、原子更新器

  • AtomicReferenceFieldUpdater // 域 字段
  • AtomicIntegerFieldUpdater
  • AtomicLongFieldUpdate

原子更新器用于帮助我们改变某个对象中的某个属性

  1. package panw.JUC;
  3. import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
  5. public class AtomicReferenceFieldUpdaterTest {
  6.     public static void main(String[] args) {
  7.         Person person = new Person();
  8.         AtomicReferenceFieldUpdater<Person, String> updater =AtomicReferenceFieldUpdater.newUpdater(Person.class,String.class,"name");
  9.         System.out.println("更新前"+person);
  10.         updater.compareAndSet(person,null,"panw");
  11.         System.out.println("更新后"+person);
  12.     }
  13. }
  14. class Person{
  15.     volatile String name;
  17.     @Override
  18.     public String toString() {
  19.         return "Person{" +
  20.                 "name='" + name + '\'' +
  21.                 '}';
  22.     }
  23. }

从上面的例子可以看出，原子更新器是通过newUpdater来获取实例的。其中传入了三个参数

  • 拥有属性的类的Class
  • 属性的Class
  • 属性的名称

大概可以猜出来，初始化过程用到了反射，下面来看看源码：

内部实现类

image.png
AtomicReferenceFieldUpdater为抽象类，该类内部有一个自己的实现类**AtomicReferenceFieldUpdaterImpl**

  1. private static final class AtomicReferenceFieldUpdaterImpl<T,V>
  2.         extends AtomicReferenceFieldUpdater<T,V>

newUpdater方法：

  1. @CallerSensitive
  2. public static <U,W> AtomicReferenceFieldUpdater<U,W> newUpdater(Class<U> tclass,
  3.                                                                 Class<W> vclass,
  4.                                                                 String fieldName) {
  5.     return new AtomicReferenceFieldUpdaterImpl<U,W>
  6.         (tclass, vclass, fieldName, Reflection.getCallerClass());
  7. }

**AtomicReferenceFieldUpdaterImpl**构造方法：

  1. AtomicReferenceFieldUpdaterImpl(final Class<T> tclass,
  2.                                         final Class<V> vclass,
  3.                                         final String fieldName,
  4.                                         final Class<?> caller) {
  5.             final Field field;
  6.             final Class<?> fieldClass;
  7.             final int modifiers;
  8.             try {
  9.                 field = AccessController.doPrivileged(
  10.                     new PrivilegedExceptionAction<Field>() {
  11.                         public Field run() throws NoSuchFieldException {
  12.                             return tclass.getDeclaredField(fieldName);
  13.                         }
  14.                     });
  15.                 modifiers = field.getModifiers();
  16.                 sun.reflect.misc.ReflectUtil.ensureMemberAccess(
  17.                     caller, tclass, null, modifiers);
  18.                 ClassLoader cl = tclass.getClassLoader();
  19.                 ClassLoader ccl = caller.getClassLoader();
  20.                 if ((ccl != null) && (ccl != cl) &&
  21.                     ((cl == null) || !isAncestor(cl, ccl))) {
  22.                     sun.reflect.misc.ReflectUtil.checkPackageAccess(tclass);
  23.                 }
  24.                 fieldClass = field.getType();
  25.             } catch (PrivilegedActionException pae) {
  26.                 throw new RuntimeException(pae.getException());
  27.             } catch (Exception ex) {
  28.                 throw new RuntimeException(ex);
  29.             }
  31.             if (vclass != fieldClass)
  32.                 throw new ClassCastException();
  33.             if (vclass.isPrimitive())
  34.                 throw new IllegalArgumentException("Must be reference type");
  36.             if (!Modifier.isVolatile(modifiers))
  37.                 throw new IllegalArgumentException("Must be volatile type");
  39.             this.cclass = (Modifier.isProtected(modifiers) &&
  40.                            tclass.isAssignableFrom(caller) &&
  41.                            !isSamePackage(tclass, caller))
  42.                           ? caller : tclass;
  43.             this.tclass = tclass;
  44.             this.vclass = vclass;
  45.             this.offset = U.objectFieldOffset(field);
  46.         }

从上面代码可以看到确实是通过反射来实现更新

八、LongAdder

LongAdder是JDK8新增的原子操作类，它提供了一种新的思路，AtomicLong的性能瓶颈是由于大量线程同时更新一个变量造成的，那么能不能把这个变量拆分出来，变成多个变量，然后让线程去竞争这些变量，最后合并即可？LongAdder的设计精髓就在这里，通过将变量拆分成多个元素，降低该变量的并发度，最后进行合并元素，变相的减少了CAS的失败次数。
共享模型之无锁 - 图2

九、Unsafe

Unsafe 对象提供了非常底层的，操作内存、线程的方法，Unsafe 对象不能直接调用，只能通过反射获得

  1. package panw.JUC;
  3. import sun.misc.Unsafe;
  5. import java.lang.reflect.Constructor;
  6. import java.lang.reflect.Field;
  7. import java.lang.reflect.InvocationTargetException;
  9. public class UnsafeGetTest {
  10.     public static void main(String[] args) throws NoSuchMethodException, InvocationTargetException, InstantiationException, IllegalAccessException, NoSuchFieldException {
  11.         Class<Unsafe> unsafeClass = Unsafe.class;
  12.         Constructor<Unsafe> constructor = unsafeClass.getDeclaredConstructor();
  13.         constructor.setAccessible(true);
  14.         Unsafe unsafe = constructor.newInstance();
  16.         Field field = Teacher.class.getDeclaredField("name");
  17.         long offset = unsafe.objectFieldOffset(field);
  18.         System.out.println(offset);
  19.         Person person = new Person();
  20.         unsafe.compareAndSwapObject(person,offset,null,"张三");
  21.         System.out.println(person);
  23.     }
  24. }
  25. class Teacher{
  26.     volatile String name;
  28.     @Override
  29.     public String toString() {
  30.         return "Teacher{" +
  31.                 "name='" + name + '\'' +
  32.                 '}';
  33.     }
  34. }