编程导航会员会员
登录/注册

Day23 CAS与synchronized

CAS(Compare And Swap)

Compare And Swap 比较与交换，CPU硬件层面的一种指令，是非阻塞同步的实现原理。

CAS指令操作包括三个参数：内存值(内存地址值)V、预期值E、新值N，当CAS指令执行时，当且仅当预期值E和内存值V相同时，才更新内存值为N，否则就不执行更新，无论更新与否都会返回否会返回旧的内存值V

image.png

CAS是一种无锁算法，在不使用锁的情况下实现多线程之间的变量同步。在Java中，CAS操作是由Unsafe类提供支持的(构造器私有，需要通过反射访问)

▼
java
复制代码
public class CASTest {

    public static void main(String[] args) {
        Entity entity = new Entity();
			
        Unsafe unsafe = UnsafeFactory.getUnsafe();

        long offset = UnsafeFactory.getFieldOffset(unsafe, Entity.class, "x");

        boolean successful;

        // 4个参数分别是：对象实例、字段的内存偏移量、字段期望值、字段新值
        successful = unsafe.compareAndSwapInt(entity, offset, 0, 3);
        System.out.println(successful + "\t" + entity.x);

        successful = unsafe.compareAndSwapInt(entity, offset, 3, 5);
        System.out.println(successful + "\t" + entity.x);

        successful = unsafe.compareAndSwapInt(entity, offset, 3, 8);
        System.out.println(successful + "\t" + entity.x);
    }
}

public class UnsafeFactory {

    /**
     * 获取 Unsafe 对象
     * @return
     */
    public static Unsafe getUnsafe() {
        try {
            Field field = Unsafe.class.getDeclaredField("theUnsafe");
            field.setAccessible(true);
            return (Unsafe) field.get(null);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    /**
     * 获取字段的内存偏移量
     * @param unsafe
     * @param clazz
     * @param fieldName
     * @return
     */
    public static long getFieldOffset(Unsafe unsafe, Class clazz, String fieldName) {
        try {
            return unsafe.objectFieldOffset(clazz.getDeclaredField(fieldName));
        } catch (NoSuchFieldException e) {
            throw new Error(e);
        }
    }
}
  • CAS缺陷

    • 自旋开销 失败时循环重试，高并发下CPU占用高

    • ABA问题 值从A→B→A CAS误以为没变 使可用AtomicStampedReference解决

      构造方法中有两个参数，reference即我们实际存储的变量，stamp是版本，相当于加了乐观锁

      ▼
      java
      复制代码
      @Slf4j
public class AtomicStampedReferenceTest {

    public static void main(String[] args) {
        // 定义AtomicStampedReference    Pair.reference值为1, Pair.stamp为1
        AtomicStampedReference atomicStampedReference = new AtomicStampedReference(1,1);

        new Thread(()->{
            int[] stampHolder = new int[1];
            int value = (int) atomicStampedReference.get(stampHolder);
            int stamp = stampHolder[0];
            log.debug("Thread1 read value: " + value + ", stamp: " + stamp);

            // 阻塞1s
            LockSupport.parkNanos(1000000000L);
            // Thread1通过CAS修改value值为3
            if (atomicStampedReference.compareAndSet(value, 3,stamp,stamp+1)) {
                log.debug("Thread1 update from " + value + " to 3");
            } else {
                log.debug("Thread1 update fail!");
            }
        },"Thread1").start();

        new Thread(()->{
            int[] stampHolder = new int[1];
            int value = (int)atomicStampedReference.get(stampHolder);
            int stamp = stampHolder[0];
            log.debug("Thread2 read value: " + value+ ", stamp: " + stamp);
            // Thread2通过CAS修改value值为2
            if (atomicStampedReference.compareAndSet(value, 2,stamp,stamp+1)) {
                log.debug("Thread2 update from " + value + " to 2");

                value = (int) atomicStampedReference.get(stampHolder);
                stamp = stampHolder[0];
                log.debug("Thread2 read value: " + value+ ", stamp: " + stamp);
                // Thread2通过CAS修改value值为1
                if (atomicStampedReference.compareAndSet(value, 1,stamp,stamp+1)) {
                    log.debug("Thread2 update from " + value + " to 1");
                }
            }
        },"Thread2").start();
    }
}
  • CAS 在Java中的应用——Atomic原子类

    • 基本类型：AtomicInteger、AtomicLong、AtomicBoolean；

      ▼
      java
      复制代码
      public class AtomicIntegerTest {
    static AtomicInteger sum = new AtomicInteger(0);

    public static void main(String[] args) {

        for (int i = 0; i < 10; i++) {
            Thread thread = new Thread(() -> {
                for (int j = 0; j < 10000; j++) {
                    // 原子自增  CAS
                    sum.incrementAndGet();
                    //TODO
                }
            });
            thread.start();
        }

        try {
            Thread.sleep(3000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(sum.get());
    }
}

    • 引用类型：AtomicReference、AtomicStampedRerence、AtomicMarkableReference；

      ▼
      java
      复制代码
      public class AtomicReferenceTest {
    public static void main( String[] args ) {
        User user1 = new User("张三", 23);
        User user2 = new User("李四", 25);
        User user3 = new User("王五", 20);

        //初始化为 user1
        AtomicReference<User> atomicReference = new AtomicReference<>();
        atomicReference.set(user1);

        //把 user2 赋给 atomicReference
        atomicReference.compareAndSet(user1, user2);
        System.out.println(atomicReference.get());

        //把 user3 赋给 atomicReference
        atomicReference.compareAndSet(user1, user3);
        System.out.println(atomicReference.get());   
    }
}

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

    • 数组类型：AtomicIntegerArray、AtomicLongArray、AtomicReferenceArray

      ▼
      java
      复制代码
      public class AtomicIntegerArrayTest {

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

    public static void main(String[] args) throws InterruptedException {

        //设置索引0的元素为100
        atomicIntegerArray.set(0, 100);
        System.out.println(atomicIntegerArray.get(0));
        //以原子更新的方式将数组中索引为1的元素与输入值相加
        atomicIntegerArray.getAndAdd(1,5);

        System.out.println(atomicIntegerArray);
    }
}

    • 对象属性原子修改器：AtomicIntegerFieldUpdater、AtomicLongFieldUpdater、AtomicReferenceFieldUpdater 使用约束

      • 字段必须是volatile修饰的，确保线程之间共享变量立即可见
      • 只能是操作当前实例变量，不能是类的静态变量或者父类的变量
      • 一定要是可变的变量，不能用final修饰
      • 对于AtomicIntegerFieldUpdater和AtomicLongFieldUpdater只能修改int/long类型的字段，不能修改其包装类型（Integer/Long）。如果要修改包装类型就需要使用AtomicReferenceFieldUpdater。
      ▼
      java
      复制代码
      public class AtomicIntegerFieldUpdaterTest {

    public static class Candidate {
        volatile int score = 0;
        AtomicInteger score2 = new AtomicInteger();
    }

    public static final AtomicIntegerFieldUpdater<Candidate> scoreUpdater =
            AtomicIntegerFieldUpdater.newUpdater(Candidate.class, "score");

    public static AtomicInteger realScore = new AtomicInteger(0);

    public static void main(String[] args) throws InterruptedException {

        final Candidate candidate = new Candidate();

        Thread[] t = new Thread[10000];
        for (int i = 0; i < 10000; i++) {
            t[i] = new Thread(new Runnable() {
                @Override
                public void run() {
                    if (Math.random() > 0.4) {
                        candidate.score2.incrementAndGet();
                        scoreUpdater.incrementAndGet(candidate);
                        realScore.incrementAndGet();
                    }
                }
            });
            t[i].start();
        }
        for (int i = 0; i < 10000; i++) {
            t[i].join();
        }
        System.out.println("AtomicIntegerFieldUpdater Score=" + candidate.score);
        System.out.println("AtomicInteger Score=" + candidate.score2.get());
        System.out.println("realScore=" + realScore.get());

    }
}

    • 原子类型累加器（jdk1.8增加的类）：DoubleAccumulator、DoubleAdder、LongAccumulator、LongAdder、Striped64

      ▼
      java
      复制代码
      public class LongAdderTest {

    public static void main(String[] args) {
        testAtomicLongVSLongAdder(10, 10000);
        System.out.println("==================");
        testAtomicLongVSLongAdder(10, 200000);
        System.out.println("==================");
        testAtomicLongVSLongAdder(100, 200000);
    }

    static void testAtomicLongVSLongAdder(final int threadCount, final int times) {
        try {
            long start = System.currentTimeMillis();
            testLongAdder(threadCount, times);
            long end = System.currentTimeMillis() - start;
            System.out.println("条件>>>>>>线程数:" + threadCount + ", 单线程操作计数" + times);
            System.out.println("结果>>>>>>LongAdder方式增加计数" + (threadCount * times) + "次,共计耗时:" + end);

            long start2 = System.currentTimeMillis();
            testAtomicLong(threadCount, times);
            long end2 = System.currentTimeMillis() - start2;
            System.out.println("条件>>>>>>线程数:" + threadCount + ", 单线程操作计数" + times);
            System.out.println("结果>>>>>>AtomicLong方式增加计数" + (threadCount * times) + "次,共计耗时:" + end2);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    static void testAtomicLong(final int threadCount, final int times) throws InterruptedException {
        CountDownLatch countDownLatch = new CountDownLatch(threadCount);
        AtomicLong atomicLong = new AtomicLong();
        for (int i = 0; i < threadCount; i++) {
            new Thread(new Runnable() {
                @Override
                public void run() {
                    for (int j = 0; j < times; j++) {
                        atomicLong.incrementAndGet();
                    }
                    countDownLatch.countDown();
                }
            }, "my-thread" + i).start();
        }
        countDownLatch.await();
    }

    static void testLongAdder(final int threadCount, final int times) throws InterruptedException {
        CountDownLatch countDownLatch = new CountDownLatch(threadCount);
        LongAdder longAdder = new LongAdder();
        for (int i = 0; i < threadCount; i++) {
            new Thread(new Runnable() {
                @Override
                public void run() {
                    for (int j = 0; j < times; j++) {
                        longAdder.add(1);
                    }
                    countDownLatch.countDown();
                }
            }, "my-thread" + i).start();
        }

        countDownLatch.await();
    }
}

并发锁

一段代码块内如果存在对共享资源的多线程读写操作，称这段代码块为临界区，其共享资源为临界资源。多个线程在临界区内执行，由于代码的执行序列不同而导致结果无法预测，称之为发生了竞态条件，为了避免临界区的竞态条件发生，可使用两种解决方案：

  • 阻塞式方案： synchronized、Lock

  • 非阻塞式方案：原子变量

  • synchronized

    synchronized 同步块是 Java 提供的一种原子性内置锁，Java 中的每个对象都可以把它当作一个同步锁来使用，这些 Java 内置的使用者看不到的锁被称为内置锁，也叫作监视器锁。

    • 加锁方式

      • 修饰实例方法 锁住的是该类的实例对象
      • 修饰静态方法 锁住的是类对象

      行内代码块

      • 修饰实例对象 synchronized(this) 锁住实例对象
      • 修饰class synchronized(Demo.class) 锁住类对象
      • 修饰任意对象Object 锁住实例对象Object

      image.png

    • 实现原理

      • synchronized是JVM内置锁，基于Monitor机制实现，依赖底层操作系统的Mutex（互斥量），内部会有等待队列(cxq 和 EntryList)和条件等待队列(waitSet)来存放相应阻塞的线程。未竞争到锁的线程存储到等待队列中，获得锁的线程调用 wait 后便存放在条件等待队列中，解锁和 notify 都会唤醒相应队列中的等待线程来争抢锁。它是一个重量级锁，性能较低 同步方法是通过方法中的access_flags中设置ACC_SYNCHRONIZED标志来实现；同步代码块是通过monitorenter和monitorexit来实现。

      image.png

      在获取锁时，是将当前线程插入到cxq的头部，而释放锁时，默认策略（QMode=0）是：如果EntryList为空，则将cxq中的元素按原有顺序插入到EntryList，并唤醒第一个线程，也就是当EntryList为空时，是后来的线程先获取锁。_EntryList不为空，直接从_EntryList中唤醒线程。 为什么会有_cxq 和 _EntryList 两个列表来放线程？ 因为会有多个线程会同时竞争锁，所以搞了个 _cxq 这个单向链表基于 CAS 来 hold 住这些并发，然后另外搞一个 _EntryList 这个双向链表，来在每次唤醒的时候搬迁一些线程节点，降低 _cxq 的尾部竞争。

      • Monitor机制

        Monitor，直译为“监视器”，而操作系统领域一般翻译为“管程”。管程是指管理共享变量以及对共享变量操作的过程，让它们支持并发，java内置的管程里只有一个条件变量

        Java管程模型

        java.lang.Object 类定义了 wait()，notify()，notifyAll() 方法，这些方法的具体实现，依赖于 ObjectMonitor 实现，这是 JVM 内部基于 C++ 实现的一套机制

      • 四种锁状态

        • 重量级锁 Heavyweight Locking

          synchronized 由于阻塞和唤醒依赖于底层的操作系统实现，系统调用存在用户态与内核态之间的切换，所以有较高的开销，因此称之为重量级锁。在JDK6+优化后，JVM会根据竞争情况动态升级锁状态（锁只能升级不能降级），synchronized就不再是“重量级锁”的代名词，并发性能基本与Lock持平

          重量级锁的优化策略

          • 锁粗化 Lock Coarsening 就是将多个连续的锁扩展为一个更大范围的锁。 粒度越细，加锁和解锁的开销越多，粗化可以减少开销从而提高效率

            ▼
            java
            复制代码
            StringBuffer buffer = new StringBuffer();
// 线程安全类，append操作会加锁，JVM识别到如下操作时会合并成一把范围更大的锁
buffer.append("aaa").append(" bbb").append(" ccc");

          • 锁消除 Lock Elimination 当一个数据仅在一个线程中使用，或者说这个数据的作用域仅限于一个线程时，这个线程对该数据的所有操作都不需要加锁。 去除不必要的锁竞争，只有自己用，加锁反而增加开销，浪费性能。

            ▼
            java
            复制代码
            public class LockEliminationTest {
    /**
     * 锁消除
     * -XX:+EliminateLocks 开启锁消除(jdk8默认开启）//执行时间2601 ms
     * -XX:-EliminateLocks 关闭锁消除 // 执行时间4688 ms   
     */
    public void append(String str1, String str2) {
        StringBuffer stringBuffer = new StringBuffer();
        stringBuffer.append(str1).append(str2);
    }
    public static void main(String[] args) throws InterruptedException {
        LockEliminationTest demo = new LockEliminationTest();
        long start = System.currentTimeMillis();
        for (int i = 0; i < 100000000; i++) {
            demo.append("aaa", "bbb");
        }
        long end = System.currentTimeMillis();
        System.out.println("执行时间：" + (end - start) + " ms");
    }

}

          • CAS自适应自旋 Adaptive Spinning 自旋的目的是为了减少线程挂起的次数，尽量避免直接挂起线程（挂起操作涉及系统调用，存在用户态和内核态切换，这才是重量级锁最大的开销）

        • 轻量级锁 Lightweight Locking 多个线程都是在不同的时间段来请求同一把锁，此时不存在竞争，根本就用不需要阻塞线程，连 monitor 对象都不需要。

          线程在自己的栈帧中创建一个Lock Record，把对象的Mark Word拷贝进去，然后CAS把对象头指向这个Lock Record。成功就拿到锁，不成功就膨胀成重量级锁

        • 偏向锁 Biased Locking（JDK15+默认禁用，都是有并发的）始终只有一个线程访问，不存在其他线程竞争，此时CAS也不需要，偏向锁可以消除锁重入（CAS）的开销

        • 无锁

        image.png

JUC
0个评论
全部评论
点击登录，快来和大家讨论吧～
表情
图片
暂无评论
麋鹿
麋鹿
等级vip
内容推荐
更多
Day 21 线程知识补充
2
Day22 CompletableFuture与ThreadLocal
2
Day24 ReentrantLock、并发容器
3
Day25 线程池
3
Day28 JMM内存模型
3
作者分享
更多
面经N篇
5
面经三篇
4
面经三篇
17
Day 35 被拷打得体无完肤，终究还是自己太菜了
8
Day34 笔试+面经
8
下载 APP