在使用多线程并发编程的时,经常会遇到对共享变量修改操作。此时我们可以选择ConcurrentHashMap,ConcurrentLinkedQueue来进行安全地存储数据。但如果单单是涉及状态的修改,线程执行顺序问题,使用Atomic开头的原子组件或者ReentrantLock、CyclicBarrier之类的同步组件,会是更好的选择,下面将一一介绍它们的原理和用法

  • 原子组件的实现原理CAS
  • AtomicBoolean、AtomicIntegerArray等原子组件的用法、
  • 同步组件的实现原理
  • ReentrantLock、CyclicBarrier等同步组件的用法
  • 可用来实现变量、状态在多线程下的原子性操作
  • 可用于实现同步锁(ReentrantLock)
  • 原子组件的原子性操作是靠使用cas来自旋操作volatile变量实现的
  • volatile的类型变量保证变量被修改时,其他线程都能看到最新的值
  • cas则保证value的修改操作是原子性的,不会被中断
  1. AtomicBoolean //布尔类型
  2. AtomicInteger //正整型数类型
  3. AtomicLong //长整型类型
  • 使用示例
  1. public static void main(String[] args) throws Exception {
  2. AtomicBoolean atomicBoolean = new AtomicBoolean(false);
  3. //异步线程修改atomicBoolean
  4. CompletableFuture<Void> future = CompletableFuture.runAsync(() ->{
  5. try {
  6. Thread.sleep(1000); //保证异步线程是在主线程之后修改atomicBoolean为false
  7. atomicBoolean.set(false);
  8. }catch (Exception e){
  9. throw new RuntimeException(e);
  10. }
  11. });
  12. atomicBoolean.set(true);
  13. future.join();
  14. System.out.println("boolean value is:"+atomicBoolean.get());
  15. }
  16. ---------------输出结果------------------
  17. boolean value is:false
  1. AtomicReference
  2. //加时间戳版本的引用类原子类
  3. AtomicStampedReference
  4. //相当于AtomicStampedReference,AtomicMarkableReference关心的是
  5. //变量是否还是原来变量,中间被修改过也无所谓
  6. AtomicMarkableReference
  • AtomicReference的源码如下,它内部定义了一个volatile V value,并借助VarHandle(具体子类是FieldInstanceReadWrite)实现原子操作,MethodHandles会帮忙计算value在类的偏移位置,最后在VarHandle调用Unsafe.public final native boolean compareAndSetReference(Object o, long offset, Object expected, Object x)方法原子修改对象的属性
  1. public class AtomicReference<V> implements java.io.Serializable {
  2. private static final long serialVersionUID = -1848883965231344442L;
  3. private static final VarHandle VALUE;
  4. static {
  5. try {
  6. MethodHandles.Lookup l = MethodHandles.lookup();
  7. VALUE = l.findVarHandle(AtomicReference.class, "value", Object.class);
  8. } catch (ReflectiveOperationException e) {
  9. throw new ExceptionInInitializerError(e);
  10. }
  11. }
  12. private volatile V value;
  13. ....
  • 线程X准备将变量的值从A改为B,然而这期间线程Y将变量的值从A改为C,然后再改为A;最后线程X检测变量值是A,并置换为B。但实际上,A已经不再是原来的A了
  • 解决方法,是把变量定为唯一类型。值可以加上版本号,或者时间戳。如加上版本号,线程Y的修改变为A1->B2->A3,此时线程X再更新则可以判断出A1不等于A3
  • AtomicStampedReference的实现和AtomicReference差不多,不过它原子修改的变量是volatile Pair<V> pair;,Pair是其内部类。AtomicStampedReference可以用来解决ABA问题
  1. public class AtomicStampedReference<V> {
  2. private static class Pair<T> {
  3. final T reference;
  4. final int stamp;
  5. private Pair(T reference, int stamp) {
  6. this.reference = reference;
  7. this.stamp = stamp;
  8. }
  9. static <T> Pair<T> of(T reference, int stamp) {
  10. return new Pair<T>(reference, stamp);
  11. }
  12. }
  13. private volatile Pair<V> pair;
  • 如果我们不关心变量在中间过程是否被修改过,而只是关心当前变量是否还是原先的变量,则可以使用AtomicMarkableReference
  • AtomicStampedReference的使用示例
  1. public class Main {
  2. public static void main(String[] args) throws Exception {
  3. Test old = new Test("hello"), newTest = new Test("world");
  4. AtomicStampedReference<Test> reference = new AtomicStampedReference<>(old, 1);
  5. reference.compareAndSet(old, newTest,1,2);
  6. System.out.println("对象:"+reference.getReference().name+";版本号:"+reference.getStamp());
  7. }
  8. }
  9. class Test{
  10. Test(String name){ this.name = name; }
  11. public String name;
  12. }
  13. ---------------输出结果------------------
  14. 对象:world;版本号:2
  1. AtomicIntegerArray  //整型数组
  2. AtomicLongArray  //长整型数组
  3. AtomicReferenceArray //引用类型数组
  • 数组原子类内部会初始一个final的数组,它把整个数组当做一个对象,然后根据下标index计算法元素偏移量,再调用UNSAFE.compareAndSetReference进行原子操作。数组并没被volatile修饰,为了保证元素类型在不同线程的可见,获取元素使用到了UNSAFEpublic native Object getReferenceVolatile(Object o, long offset)方法来获取实时的元素值
  • 使用示例
  1. //元素默认初始化为0
  2. AtomicIntegerArray array = new AtomicIntegerArray(2);
  3. // 下标为0的元素,期待值是0,更新值是1
  4. array.compareAndSet(0,0,1);
  5. System.out.println(array.get(0));
  6. ---------------输出结果------------------
  7. 1
  1. AtomicIntegerFieldUpdater 
  2. AtomicLongFieldUpdater
  3. AtomicReferenceFieldUpdater
  • 如果操作对象是某一类型的属性,可以使用AtomicIntegerFieldUpdater原子更新,不过类的属性需要定义成volatile修饰的变量,保证该属性在各个线程的可见性,否则会报错
  • 使用示例
  1. public class Main {
  2. public static void main(String[] args) {
  3. AtomicReferenceFieldUpdater<Test,String> fieldUpdater = AtomicReferenceFieldUpdater.newUpdater(Test.class,String.class,"name");
  4. Test test = new Test("hello world");
  5. fieldUpdater.compareAndSet(test,"hello world","siting");
  6. System.out.println(fieldUpdater.get(test));
  7. System.out.println(test.name);
  8. }
  9. }
  10. class Test{
  11. Test(String name){ this.name = name; }
  12. public volatile String name;
  13. }
  14. ---------------输出结果------------------
  15. siting
  16. siting
  1. Striped64
  2. LongAccumulator
  3. LongAdder
  4. //accumulatorFunction:运算规则,identity:初始值
  5. public LongAccumulator(LongBinaryOperator accumulatorFunction,long identity)
  • LongAccumulator和LongAdder都继承于Striped64,Striped64的主要思想是和ConcurrentHashMap有点类似,分段计算,单个变量计算并发性能慢时,我们可以把数学运算分散在多个变量,而需要计算总值时,再一一累加起来
  • LongAdder相当于LongAccumulator一个特例实现
  • LongAccumulator的示例
  1. public static void main(String[] args) throws Exception {
  2. LongAccumulator accumulator = new LongAccumulator(Long::sum, 0);
  3. for(int i=0;i<100000;i++){
  4. CompletableFuture.runAsync(() -> accumulator.accumulate(1));
  5. }
  6. Thread.sleep(1000); //等待全部CompletableFuture线程执行完成,再获取
  7. System.out.println(accumulator.get());
  8. }
  9. ---------------输出结果------------------
  10. 100000
  • java的多数同步组件会在内部维护一个状态值,和原子组件一样,修改状态值时一般也是通过cas来实现。而状态修改的维护工作被Doug Lea抽象出AbstractQueuedSynchronizer(AQS)来实现
  • AQS的原理可以看下之前写的一篇文章:详解锁原理,synchronized、volatile+cas底层实现
  • ReentrantLock、ReentrantReadWriteLock都是基于AQS(AbstractQueuedSynchronizer)实现的。因为它们有公平锁和非公平锁的区分,因此没直接继承AQS,而是使用内部类去继承,公平锁和非公平锁各自实现AQS,ReentrantLock、ReentrantReadWriteLock再借助内部类来实现同步
  • ReentrantLock的使用示例
  1. ReentrantLock lock = new ReentrantLock();
  2. if(lock.tryLock()){
  3. //业务逻辑
  4. lock.unlock();
  5. }
  • ReentrantReadWriteLock的使用示例
  1. public static void main(String[] args) throws Exception {
  2. ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
  3. if(lock.readLock().tryLock()){ //读锁
  4. //业务逻辑
  5. lock.readLock().unlock();
  6. }
  7. if(lock.writeLock().tryLock()){ //写锁
  8. //业务逻辑
  9. lock.writeLock().unlock();
  10. }
  11. }
  • Semaphore和ReentrantLock一样,也有公平和非公平竞争锁的策略,一样也是通过内部类继承AQS来实现同步
  • 通俗解释:假设有一口井,最多有三个人的位置打水。每有一个人打水,则需要占用一个位置。当三个位置全部占满时,第四个人需要打水,则要等待前三个人中一个离开打水位,才能继续获取打水的位置
  • 使用示例
  1. public static void main(String[] args) throws Exception {
  2. Semaphore semaphore = new Semaphore(2);
  3. for (int i = 0; i < 3; i++)
  4. CompletableFuture.runAsync(() -> {
  5. try {
  6. System.out.println(Thread.currentThread().toString() + " start ");
  7. if(semaphore.tryAcquire(1)){
  8. Thread.sleep(1000);
  9. semaphore.release(1);
  10. System.out.println(Thread.currentThread().toString() + " 无阻塞结束 ");
  11. }else {
  12. System.out.println(Thread.currentThread().toString() + " 被阻塞结束 ");
  13. }
  14. } catch (Exception e) {
  15. throw new RuntimeException(e);
  16. }
  17. });
  18. //保证CompletableFuture 线程被执行,主线程再结束
  19. Thread.sleep(2000);
  20. }
  21. ---------------输出结果------------------
  22. Thread[ForkJoinPool.commonPool-worker-19,5,main] start
  23. Thread[ForkJoinPool.commonPool-worker-5,5,main] start
  24. Thread[ForkJoinPool.commonPool-worker-23,5,main] start
  25. Thread[ForkJoinPool.commonPool-worker-23,5,main] 被阻塞结束
  26. Thread[ForkJoinPool.commonPool-worker-5,5,main] 无阻塞结束
  27. Thread[ForkJoinPool.commonPool-worker-19,5,main] 无阻塞结束
  • 可以看出三个线程,因为信号量设定为2,第三个线程是无法获取信息成功的,会打印阻塞结束
  • CountDownLatch也是靠AQS实现的同步操作
  • 通俗解释:玩游戏时,假如主线任务需要靠完成五个小任务,主线任务才能继续进行时。此时可以用CountDownLatch,主线任务阻塞等待,每完成一小任务,就done一次计数,直到五个小任务全部被执行才能触发主线
  • 使用示例
  1. public static void main(String[] args) throws Exception {
  2. CountDownLatch count = new CountDownLatch(2);
  3. for (int i = 0; i < 2; i++)
  4. CompletableFuture.runAsync(() -> {
  5. try {
  6. Thread.sleep(1000);
  7. System.out.println(" CompletableFuture over ");
  8. count.countDown();
  9. } catch (Exception e) {
  10. throw new RuntimeException(e);
  11. }
  12. });
  13. //等待CompletableFuture线程的完成
  14. count.await();
  15. System.out.println(" main over ");
  16. }
  17. ---------------输出结果------------------
  18. CompletableFuture over
  19. CompletableFuture over
  20. main over
  • CyclicBarrier则是靠ReentrantLock lockCondition trip属性来实现同步
  • 通俗解释:CyclicBarrier需要阻塞全部线程到await状态,然后全部线程再全部被唤醒执行。想象有一个栏杆拦住五只羊,需要当五只羊一起站在栏杆时,栏杆才会被拉起,此时所有的羊都可以飞跑出羊圈
  • 使用示例
  1. public static void main(String[] args) throws Exception {
  2. CyclicBarrier barrier = new CyclicBarrier(2);
  3. CompletableFuture.runAsync(()->{
  4. try {
  5. System.out.println("CompletableFuture run start-"+ Clock.systemUTC().millis());
  6. barrier.await(); //需要等待main线程也执行到await状态才能继续执行
  7. System.out.println("CompletableFuture run over-"+ Clock.systemUTC().millis());
  8. }catch (Exception e){
  9. throw new RuntimeException(e);
  10. }
  11. });
  12. Thread.sleep(1000);
  13. //和CompletableFuture线程相互等待
  14. barrier.await();
  15. System.out.println("main run over!");
  16. }
  17. ---------------输出结果------------------
  18. CompletableFuture run start-1609822588881
  19. main run over!
  20. CompletableFuture run over-1609822589880
  • StampedLock不是借助AQS,而是自己内部维护多个状态值,并配合cas实现的
  • StampedLock具有三种模式:写模式、读模式、乐观读模式
  • StampedLock的读写锁可以相互转换
  1. //获取读锁,自旋获取,返回一个戳值
  2. public long readLock()
  3. //尝试加读锁,不成功返回0
  4. public long tryReadLock()
  5. //解锁
  6. public void unlockRead(long stamp)
  7. //获取写锁,自旋获取,返回一个戳值
  8. public long writeLock()
  9. //尝试加写锁,不成功返回0
  10. public long tryWriteLock()
  11. //解锁
  12. public void unlockWrite(long stamp)
  13. //尝试乐观读读取一个时间戳,并配合validate方法校验时间戳的有效性
  14. public long tryOptimisticRead()
  15. //验证stamp是否有效
  16. public boolean validate(long stamp)
  • 使用示例
  1. public static void main(String[] args) throws Exception {
  2. StampedLock stampedLock = new StampedLock();
  3. long stamp = stampedLock.tryOptimisticRead();
  4. //判断版本号是否生效
  5. if (!stampedLock.validate(stamp)) {
  6. //获取读锁,会空转
  7. stamp = stampedLock.readLock();
  8. long writeStamp = stampedLock.tryConvertToWriteLock(stamp);
  9. if (writeStamp != 0) { //成功转为写锁
  10. //fixme 业务操作
  11. stampedLock.unlockWrite(writeStamp);
  12. } else {
  13. stampedLock.unlockRead(stamp);
  14. //尝试获取写读
  15. stamp = stampedLock.tryWriteLock();
  16. if (stamp != 0) {
  17. //fixme 业务操作
  18. stampedLock.unlockWrite(writeStamp);
  19. }
  20. }
  21. }
  22. }

版权声明:本文为cscw原创文章,遵循 CC 4.0 BY-SA 版权协议,转载请附上原文出处链接和本声明。
本文链接:https://www.cnblogs.com/cscw/p/14244073.html