《Java面试题集中营》- Java并发

阿提说说

发布于 2024-07-14 08:33:30

820

发布于 2024-07-14 08:33:30

文章被收录于专栏：Java技术进阶

《Java并发编程的艺术》、《Java并发编程之美》

运行中的线程能否强制杀死

Jdk提供了stop()方法用于强制停止线程，但官方并不建议使用，因为强制停止线程会导致线程使用的资源，比如文件描述符、网络连接处于不正常的状态。建议使用标志位的方式来终止线程，如果线程中有使用无限期的阻塞方式，比如wait()没有设置超时时间，就只能使用interrupt()方法来终止线程

@SneakyThrows
@Test
public void stack() {
    Thread1 thread1 = new Thread1();
    thread1.start();
    TimeUnit.MILLISECONDS.sleep(1);
    thread1.setStop();
}

class Thread1 extends Thread{

    private volatile boolean isStop = false;

    @SneakyThrows
    @Override
    public void run() {
        while (!isStop) {
            System.out.println(Thread.currentThread().getName() + " run...");
        }
    }

    public void setStop() {
        isStop = true;
    }
}

ThreadLocal 子类及原理, OOM产生原因及防治

InheritableThreadLocal 继承了ThreadLocal，并重写childValue、getMap、createMap，对该类的操作实际是对线程ThreadLocalMap的操作子线程能够读取父线程数据，实际原因是新建子线程的时候，会从父线程copy数据
OOM原因及防治 ThreadLocal只是一个工具类，具体存放变量的是线程的threadLocals变量，threadLocals是一个ThreadLocalMap类型的变量，内部是一个Entry数组，Entry继承自WeakReference,Entry内部的value用来存放通过ThreadLocal的set方法传递的值,key是ThreadLocal的弱引用，key虽然会被GC回收，但value不能被回收，这时候ThreadLocalMap中会存在key为null，value不为null的entry项，如果时间长了就会存在大量无用对象，造成OOM。虽然set,get也提供了一些对Entry项清理的时机，但不及时，所以在使用完毕后需要及时调用remove https://www.cnblogs.com/micrari/p/6790229.html

有哪些并发队列

ConcurrentLinkedQueue : 无界非阻塞队列，底层使用单向链表实现，对于出队和入队使用CAS来实现线程安全 LinkedBlockingQueue: 有界阻塞队列，使用单向链表实现，通过ReentrantLock实现线程安全，阻塞通过Condition实现，出队和入队各一把锁，不存在互相竞争 ArrayBlockingQueue: 有界数组方式实现的阻塞队列 , 通过ReentrantLock实现线程安全，阻塞通过Condition实现，出队和入队使用同一把锁 PriorityBlockingQueue: 带优先级的无界阻塞队列，内部使用平衡二叉树堆实现，遍历保证有序需要自定排序 DelayQueue: 无界阻塞延迟队列，队列中的每个元素都有个过期时间，当从队列获取元素时，只有过期元素才会出队列，队列头元素是最快要过期的元素 SynchronousQueue: 任何一个写需要等待一个读的操作，读操作也必须等待一个写操作，相当于数据交换 https://www.cnblogs.com/dwlsxj/p/Thread.html LinkedTransferQueue: 由链表组成的无界阻塞队列，多了tryTransfer 和 transfer方法。transfer方法，能够把生产者元素立刻传输给消费者，如果没有消费者在等待，那就会放入队列的tail节点，并阻塞等待元素被消费了返回，可以使用带超时的方法。tryTransfer方法，会在没有消费者等待接收元素的时候马上返回false

LinkedBlockingDeque: 由链表组成的双向阻塞队列，可以从队列的两端插入和移除元素

ThreadPoolExecutor构造函数有哪几个参数，实现原理，创建线程池的方式

构造参数： corePoolSize: 线程池核心线程个数 maximunPoolSize: 线程池最大线程数量 keeyAliveTime: 空闲线程存活时间 TimeUnit: 存活时间单位 workQueue: 用于保存等待执行任务的阻塞队列 ThreadFactory: 创建线程的工厂 RejectedExecutionHandler: 队列满，并且线程达到最大线程数量的时候，对新任务的处理策略，AbortPolicy（抛出异常）、CallerRunsPolicy(使用调用者所在线程执行)、DiscardOldestPolicy(调用poll丢弃一个任务，执行当前任务)、DiscardPolicy(默默丢弃、不抛异常)
原理：线程池主要是解决两个问题：一个是当执行大量异步任务时能够提供较好的性能，能复用线程处理任务；二是能够对线程池进行资源限制和管理。一个任务提交的线程池，首先会判断核心线程池是否已满，未满就会创建worker线程执行任务，已满判断阻塞队列是否已满，阻塞队列未满加入阻塞队列，已满就判断线程池线程数量是否已经达到最大值，没有就新建线程执行任务，达到最大值的话执行拒绝策略。拒绝策略有：直接抛出异常、使用调用者所在线程执行、丢弃一个旧任务，执行当前任务、直接丢弃什么都不做。
创建线程池的方式：直接new ThreadPoolExecutor 或者通过Executors工具类创建

Executors 可以创建的线程池类型

newFixedThreadPool 创建一个核心线程数跟最大线程数相同的线程池，因此池中的线程数量既不会增加也不会变少，如果有空闲线程任务就会被执行，如果没有就放入任务队列，等待空闲线程
newSingleThreadExecutor 创建一个只有一个线程的线程池，能够串行执行任务，如果线程因为异常而停止，会自动新建一个线程补充
newCachedThreadPool 创建一个核心线程数为0，最大线程为Inter.MAX_VALUE的线程池，也就是说没有限制,线程池中的线程数量不确定，但如果有空闲线程可以复用，则优先使用，如果没有空闲线程，则创建新线程处理任务，处理完放入线程池
newSingleThreadScheduledExecutor 创建只有一个线程的可以定时执行的线程池
newScheduledThreadPool 创建一个没有最大线程数限制的可以定时执行线程池
newWorkStealingPool 创建一个含有足够多线程的线程池，能够调用闲置的CPU去处理其他的任务，使用ForkJoinPool实现，jdk8新增

线程池的阻塞队列为什么都用LinkedBlockingQueue，而不用ArrayBlockingQueue

LinkedBlockingQueue 使用单向链表实现，在声明LinkedBlockingQueue的时候，可以不指定队列长度，长度为Integer.MAX_VALUE, 并且新建了一个Node对象,Node对象具有item，next变量，item用于存储元素，next指向链表下一个Node对象，在刚开始的时候链表的head,last都指向该Node对象，item、next都为null,新元素放在链表的尾部，并从头部取元素。取元素的时候只是一些指针的变化，LinkedBlockingQueue给put(放入元素),take(取元素)都声明了一把锁，放入和取互不影响，效率更高

ArrayBlockingQueue 使用数组实现，在声明的时候必须指定长度，如果长度太大，造成内存浪费，长度太小，并发性能不高，如果数组满了，就无法放入元素，除非有其他线程取出元素，放入和取出都使用同一把锁，因此存在竞争，效率比LinkedBlockingQueue低

为什么建议在不用线程池的时候，关闭线程池，线程池的作用就是复用线程

线程池的作用确实是为了减少频繁创建线程，使线程达到复用。但如果在不用线程池的情况下，线程池中的核心线程会一直存在，浪费资源，所以建议在不用的情况下调用shutdown方法关闭线程池。在需要的时候再调用创建线程池。

如何合理的配置Java线程池

如CPU密集型的任务，基本线程池应该配置多大？IO密集型的任务，基本线程池应该配置多大？用有界队列好还是无界队列好？任务非常多的时候，使用什么阻塞队列能获取最好的吞吐量？

CPU密集型，为了充分使用CPU，减少上下文切换，线程数配置成CPU个数+1个即可

IO密集型，由于可能大部分线程在处理IO,IO都比较耗时，因此可以配置成 2*CPU个数的线程，去处理其他任务

Timer 和 ScheduledThreadPoolExecutor 区别

Timer是固定的多线程生产者单线程消费，如果其中一个任务报错，其他任务也会失效;

但后者是可以配置的，既可以是多线程生产单线程消费也可以是多线程生产多线程消费

说说CopyOnWriteArrayList

CopyOnWriteArrayList是一个线程安全的ArrayList，对其的修改操作是在一个复制的数组上进行的，不影响其他线程的读操作

其中通过ReentrantLock独占锁保证只有一个线程对底层数组进行修改

由于在进行修改操作的时候，底层会复制一个新的数组，而读是在原数组上进行的，因此在多线程环境下这里会产生数据不一致的情况，称为弱一致性

适用于多读少写场景

public class CopyOnWriteArrayListTest {

    //模拟测试CopyOnWriteArrayList 的弱一致性
    @Test
    public void ListrayTest() throws InterruptedException {

        AryTest aryTest = new AryTest();

        StrClass strClass = new StrClass(aryTest.str1);
        String[] str2 = (String[]) strClass.getObjects();

        Thread thread = new Thread(() -> {
            aryTest.add();
            System.out.println(Arrays.toString(aryTest.str1));
        });

        thread.start();
        thread.join();

        System.out.println("str2=" + Arrays.toString(str2));
    }

    static class AryTest {
        String[] str1 = new String[]{"a", "b"};

        public void add() {
            String[] str2 = Arrays.copyOf(str1, str1.length + 1);
            str2[str2.length - 1] = "c";
            str1 = str2;
        }
    }

    static class StrClass {
        final Object[] objects;

        public StrClass(Object [] objects) {
            this.objects = objects;
        }

        public Object[] getObjects() {
            return objects;
        }
    }


    //copyOnWriteArrayList测试
    @Test
    public void copyOnWriteArrayList() throws InterruptedException {
        String[] str1 = new String[]{"a", "b"};
        List<String> copyOnWriteArrayList = new CopyOnWriteArrayList<>(str1);
        Thread thread = new Thread(() -> {
           copyOnWriteArrayList.add("c");
            System.out.println(copyOnWriteArrayList);
        });
        thread.start();
        System.out.println(copyOnWriteArrayList);
        thread.join();
    }
}

[a, b]
[a, b, c]

CountDownLatch原理、CyclicBarrier原理，两者区别

CountDownLatch: 使用AQS实现，通过AQS的状态变量state来作为计数器值,当多个线程调用countdown方法时实际是原子性递减AQS的状态值，当线程调用await方法后当前线程会被放入AQS阻塞队列等待计数器为0再返回

public class CountDownLatchTest {

    public static final CountDownLatch countDownLatch = new CountDownLatch(2);

    ExecutorService executor = Executors.newFixedThreadPool(2);

    @Test
    public void test1() throws InterruptedException {
        executor.submit(() -> {
            System.out.println(Thread.currentThread().getName() + " step1");
            countDownLatch.countDown();
        });
        executor.submit(() -> {
            System.out.println(Thread.currentThread().getName() + " step2");
            countDownLatch.countDown();
        });

        countDownLatch.await();
        System.out.println("thread end");
    }
}

pool-1-thread-1 step1
pool-1-thread-2 step2
thread end

CyclicBarrier: 区别：CountDownLatch计数器是一次性的，变为0后就起不到线程同步的作用了。而CyclicBarrier(撒克里克巴瑞儿)在计数器变为0后重新开始，通过调用await方法，能在所有线程到达屏障点后统一执行某个任务，再执行完后继续执行子线程，通过ReentrantLock实现

public class CyclicBarrierTest {

    public static final CyclicBarrier cycle = new CyclicBarrier(3);

    ExecutorService executorService = Executors.newFixedThreadPool(2);

    @Test
    public void test1() throws BrokenBarrierException, InterruptedException {
        executorService.submit(() -> {
            System.out.println(Thread.currentThread().getName());
            try {
                cycle.await();
                System.out.println(Thread.currentThread().getName() + "，执行结束");
            } catch (InterruptedException | BrokenBarrierException e) {
                e.printStackTrace();
            }
        });

        executorService.submit(() -> {
            System.out.println(Thread.currentThread().getName());
            try {
                Thread.sleep(3000);
                cycle.await();
                System.out.println(Thread.currentThread().getName() + "，执行结束");
            } catch (InterruptedException | BrokenBarrierException e) {
                e.printStackTrace();
            }
        });
        cycle.await();
    }
}

pool-1-thread-1
pool-1-thread-2
pool-1-thread-2，执行结束
pool-1-thread-1，执行结束

Phaser 的实现

Phaser可以替代CountDownLatch 和CyclicBarrier，但比两者更加强大，可以动态调整需要的线程个数，可以通过构造函数传入父Phaser实现层次Phaser

源码参考：https://www.cnblogs.com/tong-yuan/p/11614755.html

public class PhaserTest {

    Phaser phaser = new Phaser(2);

    ExecutorService executor = Executors.newFixedThreadPool(2);

   @Test
   public void test1() {
       executor.submit(() -> {
           System.out.println(Thread.currentThread().getName() + " step1");
           //等同 countDown()
           phaser.arrive();
       });

       executor.submit(() -> {
           System.out.println(Thread.currentThread().getName() + " step2");
           phaser.arrive(d);
       });

       //等同await()
       phaser.awaitAdvance(phaser.getPhase());
       System.out.println("thread end");
   }
}

Semaphore 原理

Semaphore 可以用来控制同时访问特定资源的线程数量，它通过协调各个线程，以保证合理的使用公共资源,使用AQS实现，AQS的状态变量state做为许可证数量，每次通过acquire()/tryAcquire()，许可证数量通过CAS原子性递减，调用release()释放许可证，原子性递增，只要有许可证就可以重复使用

@Test
public void semaphoreTest() throws InterruptedException {
    Semaphore semaphore = new Semaphore(3);
    for (int i = 0; i < 5; i++) {
        executor.execute(() -> {
            try {
                semaphore.acquire();
                System.out.println("输出");
                //semaphore.release();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
    }

    TimeUnit.SECONDS.sleep(3);
    System.out.println("释放许可证===");
    semaphore.release(2);
}

输出
输出
输出
释放许可证===
输出
输出

Exchanger 原理

用于进行线程间的数据交换，它提供一个同步点，在这个同步点两个线程可以交换彼此的数据。如果第一个线程先执行exchange()方法，它会一直等待第二个线程也执行exchange方法，当都达到同步点时，这两个线程可以交换数据。 https://blog.csdn.net/u014634338/article/details/78385521

 @Test
public void exchangerTest() throws InterruptedException {
    Exchanger<Integer> exchanger = new Exchanger<> ();

    Thread thread = new Thread(() -> {
        Integer a = 1;
        try {
            Integer b = exchanger.exchange(a);
            System.out.println("Thread1：" + b);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    });

    thread.start();
    Integer c = exchanger.exchange(2);
    Thread.sleep(3000);
    System.out.println("Thread2:" + c);
}

Thread1：2
Thread2:1

volatile的特点

修改可见性，通过volatile修饰的变量(包括对象和数组)，线程对变量的修改对另一个线程即时可见，因此另外一个线程在使用前会去判断该变量是否有修改；
禁止指令重排序 long 和double的读写操作分为两次32位操作进行，其他基本数据类型的读写操作都是原子性的 实现：加入volatile关键字的代码生成汇编代码，会发现多了一个lock前缀指令，这个指令相当于一个内存屏障，可以防止重排序，通过空的写入操作将变量的修改写入到主内存中，这就实现了可见性

volatile修饰对象分析可见：https://www.jianshu.com/p/8f2f833dc9b0

什么时候使用volatile?

写入变量值不依赖变量的当前值。因为如果依赖当前值，将是获取-计算-写入三步操作，这三步操作不是原子性的，而volatile不保证原子性

读写变量值时没有加锁。因为加锁已经保证了内存可见性，没必要再使用volatile

volatile不能保证原子性

public class VolatileTest {
    
    public volatile int inc = 0;

    public void increase() {
        inc++;
    }
    
    public static void main(String[] args) {
        final VolatileTest test = new VolatileTest();
        
        for(int i = 0; i < 10; i++) {
            new Thread(() -> {
                for(int j = 0; j < 1000; j++)
                    test.increase();
            }).start();
        }
        //保证前面的线程都执行完
        while (Thread.activeCount() > 1)
            Thread.yield();

        System.out.println(test.inc);
    }
}
//输出：<=10000

伪共享

在CPU和主内存之间添加一级或者多级高速缓冲存储器（Cache），这个缓存被集成到CPU内部，在Cache内部是按行存储的，每一行称为一个Cache行，大小一般为2的幂次数字节，当访问某个变量时，首先会去看CPU Cache内是否有该变量，如果有则直接从中获取，否则就去主内存里面获取该变量，然后把该变量所在的内存区域的一个Cache行的内存复制到Cache中。由于存放到Cache行的是内存块而不是单个变量，所以可能会把多个变量存放到一个Cache行中。根据缓存一致性协议，CPU在修改缓存行中的变量时，同一缓存行的数据将失效，这时候其他CPU需要从二级缓存或者主存中加载数据，这就导致了性能问题，称为伪共享

伪共享解决：字节填充；使用sun.misc.Contended注解

@Contended注解只用于Java核心类，如果用户类路径下的类要使用这个注解，需要添加JVM参数：-XX:-RestrictContended。默认填充宽度为128，需要自定义宽度设置 -XX:ContendedPaddingWidth参数
CPU缓存行详细说明：https://mp.weixin.qq.com/s/yosnZr0bDdLrhmpnX8TY5A

原子操作类

AtomicBoolean

布尔类型的原子操作类，内部使用int型存储布尔值，0表示false,1表示true

AtomicInteger

整型的原子操作类，1.8后提供函数式操作的方法

int getAndUpdate(IntUnaryOperator updateFunction) 使用指定函数计算并更新，返回计算前结果
int updateAndGet(IntUnaryOperator updateFunction) 使用指定函数计算并更新，返回计算后的结果
int getAndAccumulate(int x,IntBinaryOperator accumulatorFunction) 使用指定的函数计算x值和当前值，返回计算前结果
int accumulateAndGet(int x,IntBinaryOperator accumulatorFunction) 使用指定的函数计算x值和当前值，返回结算后的结果

@Test
public void atomicIntegerTest() {
    AtomicInteger atomicInteger = new AtomicInteger(1);

    int result = atomicInteger.getAndUpdate(e -> e + 3);
    返回计算前结果
    assert result == 1;
    //1 + 3
    assert atomicInteger.get() == 4;


    result = atomicInteger.updateAndGet(e -> e + 3);
    //返回计算后结果
    assert result == 7;
    // 4 + 3
    assert atomicInteger.get() == 7;


    result = atomicInteger.getAndAccumulate(10, (x, y) -> x + y);
    //返回计算前结果
    assert  result == 7;
    // 7 + 10
    assert atomicInteger.get() == 17;

    result = atomicInteger.accumulateAndGet(10, (x, y) -> x + y);
    //返回计算后的结果
    assert result == 27;
    assert atomicInteger.get() == 27;
}

AtomicIntegerArray

提供了原子性更新整型数组元素的方式

int getAndUpdate(int i, IntUnaryOperator updateFunction) 使用指定函数计算i索引的值，返回计算前结果
int updateAndGet(int i, IntUnaryOperator updateFunction) 使用指定函数计算i索引的值，返回计算后结果
int getAndAccumulate(int i, int x, IntBinaryOperator accumulatorFunction) 使用指定的函数计算x值和i索引的值，返回计算前结果
int accumulateAndGet(int i, int x, IntBinaryOperator accumulatorFunction)
使用指定的函数计算x值和i索引的值，返回计算后结果

@Test
public void atomicIntegerArrayTest() {
    AtomicIntegerArray atomicIntegerArray = new AtomicIntegerArray(10);
    int result = atomicIntegerArray.getAndUpdate(0, e -> e + 1);
    assert result == 0;
    // 0 + 1
    assert atomicIntegerArray.get(0) == 1;

    result = atomicIntegerArray.updateAndGet(0, e -> e + 1);
    assert result == 2;
    //1 + 1
    assert atomicIntegerArray.get(0) == 2;

    result = atomicIntegerArray.getAndAccumulate(0, 3, (x, y) -> x + y);
    assert result == 2;
    // 2 + 3
    assert atomicIntegerArray.get(0) == 5;
}

AtomicIntegerFieldUpdater

通过反射实现，可以对指定类的指定字段（被volatile修饰）的int型字段进行原子更新，更新字段类型必须为 volatile int

源码分析 > AtomicIntegerFieldUpdater源码分析

@Test
public void fieldUpdaterTest() {
    //第一个参数 持有给定字段的目标对象类  第二个参数 要更新的字段名称，必须在给定的目标对象中
    AtomicIntegerFieldUpdater fieldUpdater = AtomicIntegerFieldUpdater.newUpdater(Person.class, "count");
    Person person = new Person("小明", "男");
    fieldUpdater.addAndGet(person, 10);
    //10
    System.out.println(fieldUpdater.get(person));
    //Person{name='小明', sex='男', count=10}
    System.out.println(person);
}

private static class Person{
    String name;
    String sex;
    volatile int count;

    public Person(String name, String sex) {
        this.name = name;
        this.sex = sex;
    }

    @Override
    public String toString() {
        return "Person{" +
            "name='" + name + '\'' +
            ", sex='" + sex + '\'' +
            ", count=" + count +
            '}';
    }
}

AtomicLong

long型原子操作类

AtomicLongArray

提供了原子性更新long型数组元素的方式

AtomicLongFieldUpdater

通过反射实现，可以对指定类的指定字段（被volatile修饰）的long型字段进行原子更新，更新字段类型必须为 volatile long

AtomicMarkableReference

通过布尔类型做为标记，原子更新引用变量

//构造参数，提供了泛型，而不是像AtomicInteger,只能是int型
AtomicStampedReference(V initialRef, int initialStamp)

AtomicReference

同样提供了泛型变量，原子性更新变量，但没有标识，会产生ABA问题

AtomicReferenceArray

提供原型性更新泛型类数组元素的方式，内部使用Object[]数组保存引用变量

AtomicReferenceFieldUpdater

引用类型的，可以对指定类的指定字段（被volatile修改，基本类型）进行原子更新

AtomicStampedReference

通过维护一个版本号，原子更新引用变量，解决了ABA问题

DoubleAccumulator

double类型的原子更新类，提供自定义函数接口，通过不同的Cell资源，减少了竞争

DoubleAdder

double类型的原子更新类，只提供累计操作

LongAccumulator

LongAdder

ABA问题

CAS操作中会ABA问题，指线程1使用CAS修改初始值为A的变量X的时候，需要先获取变量X的值，但这时候线程2已经将变量X的值修改为了B,然后又修改成了A，虽然字面值还是A，但这个A已经是修改后的A了，对于线程1则还是认为是原来的A,而继续修改变量X的值，这就是ABA问题。

解决：JDK中提供了AtomicStampedReference类解决了该问题，其给每个变量的状态值都提供了一个版本号

ABA问题

@SneakyThrows
@Test
public void test1() {
    AtomicInteger atomicInteger = new AtomicInteger(10);

    CountDownLatch countDownLatch = new CountDownLatch(2);

    new Thread(() -> {
        atomicInteger.compareAndSet(10, 11);
        atomicInteger.compareAndSet(11,10);
        System.out.println(Thread.currentThread().getName() + "：10->11->10");
        countDownLatch.countDown();
    }).start();

    new Thread(() -> {
        try {
            TimeUnit.SECONDS.sleep(1);
            boolean isSuccess = atomicInteger.compareAndSet(10,12);
            System.out.println("设置是否成功：" + isSuccess + ",设置的新值：" + atomicInteger.get());
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        countDownLatch.countDown();
    }).start();

    countDownLatch.await();
}
//输出：线程2并没有发现初始值已经被修改
//Thread-0：10->11->10
//设置是否成功：true,设置的新值：12

AtomicStampedReference解决ABA问题,通过维护一个版本号

@SneakyThrows
@Test
public void test2() {
    AtomicStampedReference<Integer> atomicStampedReference = new AtomicStampedReference(10,1);

    CountDownLatch countDownLatch = new CountDownLatch(2);

    new Thread(() -> {
        System.out.println(Thread.currentThread().getName() + " 第一次版本：" + atomicStampedReference.getStamp());
        atomicStampedReference.compareAndSet(10, 11, atomicStampedReference.getStamp(), atomicStampedReference.getStamp() + 1);
        System.out.println(Thread.currentThread().getName() + " 第二次版本：" + atomicStampedReference.getStamp());
        atomicStampedReference.compareAndSet(11, 10, atomicStampedReference.getStamp(), atomicStampedReference.getStamp() + 1);
        System.out.println(Thread.currentThread().getName() + " 第三次版本：" + atomicStampedReference.getStamp());
        countDownLatch.countDown();
    }).start();

    new Thread(() -> {
        System.out.println(Thread.currentThread().getName() + " 第一次版本：" + atomicStampedReference.getStamp());
        try {
            TimeUnit.SECONDS.sleep(2);
            boolean isSuccess = atomicStampedReference.compareAndSet(10,12, atomicStampedReference.getStamp(), atomicStampedReference.getStamp() + 1);
            System.out.println(Thread.currentThread().getName() + " 修改是否成功：" + isSuccess + " 当前版本：" + atomicStampedReference.getStamp() + " 当前值：" + atomicStampedReference.getReference());
            countDownLatch.countDown();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }).start();

    countDownLatch.await();
}

//输出
Thread-0 第一次版本：1
Thread-0 第二次版本：2
Thread-0 第三次版本：3
Thread-1 第一次版本：3
Thread-1 修改是否成功：true 当前版本：4 当前值：12

AtomicMarkableReference 通过标志位,由于其标志位只有true和false，如果每次更新都变更标志位，在第三次的时候标志位还是跟第一次一样，并没有解决ABA问题

@SneakyThrows
@Test
public void test3() {
    AtomicMarkableReference<Integer> markableReference = new AtomicMarkableReference<>(10, false);

    CountDownLatch countDownLatch = new CountDownLatch(2);

    new Thread(() -> {
        System.out.println(Thread.currentThread().getName() + " 第一次标记：" + markableReference.isMarked());
        markableReference.compareAndSet(10, 11, markableReference.isMarked(), true);
        System.out.println(Thread.currentThread().getName() + " 第二次标记：" + markableReference.isMarked());
        markableReference.compareAndSet(11, 10, markableReference.isMarked(), false);
        System.out.println(Thread.currentThread().getName() + " 第三次标记：" + markableReference.isMarked());
        countDownLatch.countDown();
    }).start();

    new Thread(() -> {
        System.out.println(Thread.currentThread().getName() + " 第一次标记：" + markableReference.isMarked());
        try {
            TimeUnit.SECONDS.sleep(2);
            boolean isSuccess = markableReference.compareAndSet(10,12, false, true);
            System.out.println(Thread.currentThread().getName() + " 修改是否成功：" + isSuccess + " 当前标记：" + markableReference.isMarked() + " 当前值：" + markableReference.getReference());
            countDownLatch.countDown();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }).start();

    countDownLatch.await();
}

Thread-0 第一次标记：false
Thread-0 第二次标记：true
Thread-0 第三次标记：false
Thread-1 第一次标记：false
Thread-1 修改是否成功：true 当前标记：true 当前值：12

该类的标记更多的用于表示引用值是否已逻辑删除

@SneakyThrows
@Test
public void test4() {
    AtomicMarkableReference<Integer> markableReference = new AtomicMarkableReference<>(10,true);

    CountDownLatch countDownLatch = new CountDownLatch(2);

    System.out.println("初始标记: " + markableReference.isMarked());

    //该线程将10设置为逻辑删除
    new Thread(() -> {
        boolean isSuccess = markableReference.attemptMark(10,false);
        System.out.println("设置标记为flase: " + isSuccess + "，当前标记："+ markableReference.isMarked());
        countDownLatch.countDown();
    }).start();

    //该线程想要更新10->11，但标志已经变为false，与预期不符
    new Thread(() -> {
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        boolean isSuccess = markableReference.compareAndSet(10, 11, true, false);
        System.out.println("设置值: " + isSuccess + "期望标记：true"  + "，当前标记："+ markableReference.isMarked());
        countDownLatch.countDown();
    }).start();
    countDownLatch.await();
}

初始标记: true
设置标记为flase: true，当前标记：false
设置值: false期望标记：true，当前标记：false

Java8新增的原子操作类

LongAdder 由于AtomicLong通过CAS提供非阻塞的原子性操作，性能已经很好，在高并发下大量线程竞争更新同一个原子量，但只有一个线程能够更新成功，这就造成大量的CPU资源浪费。 LongAdder 通过让多个线程去竞争多个Cell资源，来解决，再很高的并发情况下,线程操作的是Cell数组，并不是base，在cell元素不足时进行2倍扩容，在高并发下性能高于AtomicLong

LongAdder的真实值是base的值与Cell数组里面所有Cell元素中value值的累加

LongAdder示例

@Test
public void longAdderTest() {
    LongAdder longAdder = new LongAdder();
    longAdder.add(101);
    longAdder.add(102);
    //203
    System.out.println(longAdder.sumThenReset());
    //0
    System.out.println(longAdder.longValue());
}

LongAdder和AtomicLong多线程累加性能测试

@Test
public void multiplyThreadLongAdderTest() throws InterruptedException {
    LongAdder longAdder = new LongAdder();
    AtomicLong atomicLong = new AtomicLong();

    AtomicLong time1 = new AtomicLong();
    AtomicLong time2 = new AtomicLong();

    int threadNum = 5;
    int cycleNums = 500000;
    CountDownLatch countDownLatch1 = new CountDownLatch(threadNum);
    for (int a = 0; a < threadNum; a++) {
        executor.execute(() -> {
            long start = System.nanoTime();
            for (int i = 0; i < cycleNums; i++) {
                longAdder.increment();
            }
            //System.out.println(longAdder.longValue());
            time1.addAndGet(System.nanoTime() - start);
            countDownLatch1.countDown();
        });
    }
    countDownLatch1.await();

    CountDownLatch countDownLatch2 = new CountDownLatch(threadNum);
    for (int a = 0; a < threadNum ; a++) {
        executor.execute(() -> {
            long start = System.nanoTime();
            for (int i = 0; i < cycleNums; i++) {
                atomicLong.incrementAndGet();
            }
            //System.out.println(atomicLong.longValue());
            time2.addAndGet(System.nanoTime() - start);
            countDownLatch2.countDown();
        });
    }
    countDownLatch2.await();

    System.out.println("data=" + longAdder.longValue() + " time1 = " + time1.longValue());
    System.out.println("data=" + atomicLong.longValue() + " time2 = " + time2.longValue());
}

data=2500000 LongAdder time1 = 112762041
data=2500000 AtomicLong time2 = 292448392

LongAccumulator 是LongAdder的增强，提供了一个函数式接口，可以自定义运算规则

@Test
public void LongAccumulatorTest() {
    LongAccumulator longAccumulator = new LongAccumulator((x,y) -> x * y, 2);
    //2 * 10
    longAccumulator.accumulate(10);
    assert  longAccumulator.get() == 20;
    //重置为2
    longAccumulator.reset();
    assert longAccumulator.get() == 2;
}

如何实现一个生产者与消费者模型

一共5种方法

同步对象的 wait() / notify() 方法
ReetrantLock Condition 的 await() / signal()方法
BlockingQueue阻塞队列 put() 和take方法
Semaphore 基于计数的信号量
PipedInputStream / PipedOutputStream 管道输入输出流 https://blog.csdn.net/ldx19980108/article/details/81707751