[Java知识库]Java JUC并发编程详解

1. JUC概述

1.1 JUC简介

JUC就是 java.util .concurrent 工具包的简称。
这是一个处理线程的工具包， JDK1.5 开始出现的。

1.2 进程与线程

比较官方的说法，
进程（Process） 是计算机中的程序关于某数据集合上的一次运行活动，是系统进行资源分配和调度的基本单位，是操作系统结构的基础。
线程（thread） 是操作系统能够进行运算调度的最小单位。

太官方了，举个简单的例子，进程就是你微信运行的那个exe进程，线程就是你作为一个时间管理者一次性打开了100个聊天窗口，特别类似并发与并行的区别，这里顺带说一下，并行就是虽然你开了100个窗口，但是很遗憾你是单核的CPU，给你的感觉好像同时操作100个窗口，对不起实际上只有一个CPU在做时间片轮转，只是因为时间间隔很短你没感知出来罢了，而并发100核CPU，开100个窗口，同时为你处理请求。

1.2 并发与并行

并行意味着可以同时取得多个任务，并同时去执行所取得的这些任务。并行模式相当于将长长的一条队列，划分成了多条短队列，所以并行缩短了任务队列的长度。并行的效率从代码层次上强依赖于多进程/多线程代码，从硬件角度上则依赖于多核 CPU。

并发(concurrent)指的是多个程序可以同时运行的现象，更细化的是多进程可以同时运行或者多指令可以同时运行。

1.3 用户线程和守护线程

用户线程:平时用到的普通线程,自定义线程
守护线程:运行在后台,是一种特殊的线程,比如垃圾回收
当主线程结束后,用户线程还在运行,JVM 存活
如果没有用户线程,都是守护线程,JVM 结束

2. Lock接口

2.1 Synchronized

synchronized实现同步的3种形式：
对于普通同步方法，锁的是当前实例对象。
对于静态同步方法，锁的是当前类的Class对象。
对于同步方法快，锁的是synchronized括号里的对象。

2.2 什么是 Lock

Lock 锁实现提供了比使用同步方法和语句可以获得的更广泛的锁操作。它们允许更灵活的结构，可能具有非常不同的属性，并且可能支持多个关联的条件对象。 Lock 提供了比 synchronized 更多的功能。

lock方法
lock()方法是平常使用得最多的一个方法，就是用来获取锁。如果锁已被其他线程获取，则进行等待。采用 Lock，必须主动去释放锁，并且在发生异常时，不会自动释放锁。因此一般来说，使用 Lock 必须在try{}catch{}块中进行，并且将释放锁的操作放在finally 块中进行，以保证锁一定被被释放，防止死锁的发生。通常使用 Lock来进行同步的话。

newCondition
关键字 synchronized 与 wait()/notify()这两个方法一起使用可以实现等待/通知模式， Lock 锁的 newContition()方法返回 Condition 对象，Condition 类也可以实现等待/通知模式。用 notify()通知时， JVM 会随机唤醒某个等待的线程， 使用 Condition 类可以进行选择性通知， Condition 比较常用的两个方法：
? await()会使当前线程等待,同时会释放锁,当其他线程调用 signal()时,线程会重新获得锁并继续执行。
? signal()用于唤醒一个等待的线程。

注意：在调用 Condition 的 await()/signal()方法前，也需要线程持有相关的 Lock 锁，调用 await()后线程会释放这个锁，在 singal()调用后会从当前Condition 对象的等待队列中，唤醒 一个线程，唤醒的线程尝试获得锁， 一旦获得锁成功就继续执行。

2.3 ReentrantLock

ReentrantLock，意思是“可重入锁” ，关于可重入锁的概念将在后面讲述。ReentrantLock 是唯一实现了 Lock 接口的类，并且 ReentrantLock提供了更多的方法。下面通过一些实例看具体看一下如何使用。

2.4 ReadWriteLock

ReadWriteLock 也是一个接口，在它里面只定义了两个方法：
Lock readLock();
Lock writeLock();

一个用来获取读锁，一个用来获取写锁。也就是说将文件的读写操作分开，分成 2 个锁来分配给线程，从而使得多个线程可以同时进行读操作。下面的ReentrantReadWriteLock 实现了 ReadWriteLock 接口。ReentrantReadWriteLock 里面提供了很多丰富的方法，不过最主要的有两个方法： readLock()和 writeLock()用来获取读锁和写锁。

package com.zrj.unit.juc;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import lombok.extern.slf4j.Slf4j;
import org.junit.Test;

import java.util.concurrent.*;
import java.util.concurrent.locks.ReentrantReadWriteLock;

/**
 * 可重入锁
 *
 * @author zrj
 * @since 2021/8/19
 **/
@Slf4j
public class ReentrantLockTest {
    // 自定义线程名称
    private static ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat("thread-local-pool-%d").build();
    // 定义线程池
    private static ExecutorService pool = new ThreadPoolExecutor(30, 200, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(1024), namedThreadFactory, new ThreadPoolExecutor.AbortPolicy());
    // 可重入读写锁
    private static ReentrantReadWriteLock reentrantLock = new ReentrantReadWriteLock();

    @Test
    public void reentrantLockTest() {
        for (int i = 0; i < 2; i++) {
            pool.execute(() -> {
                //getSync(Thread.currentThread());
                getReentrantLock(Thread.currentThread());
            });
        }
        log.info("执行完成");
    }

    /**
     * 可重入读写锁
     */
    public void getReentrantLock(Thread thread) {
        reentrantLock.readLock().lock();
        try {
            long start = System.currentTimeMillis();
            while (System.currentTimeMillis() - start <= 1) {
                log.info(thread.getName() + "：[ReentrantLock]正在进行读操作");
            }
            log.info(thread.getName() + "：[ReentrantLock]读操作完毕");
        } catch (Exception e) {
            log.error("[ReentrantLock]系统异常：" + e);
        } finally {
            reentrantLock.readLock().unlock();
        }
    }

    /**
     * 线程同步
     */
    public synchronized void getSync(Thread thread) {
        long start = System.currentTimeMillis();
        while (System.currentTimeMillis() - start <= 1) {
            log.info(thread.getName() + "：[synchronized]正在进行读操作");
        }
        log.info(thread.getName() + "：[synchronized]读操作完毕");
    }

}

2.5 Lock与Synchronized区别

1.Lock 是一个接口，而 synchronized 是 Java 中的关键字，synchronized 是内置的语言实现；
2.synchronized 在发生异常时，会自动释放线程占有的锁，因此不会导致死锁现象发生；而 Lock 在发生异常时，如果没有主动通过unLock()去释放锁，则很可能造成死锁现象，因此使用 Lock 时需要在 finally 块中释放锁；
3.Lock 可以让等待锁的线程响应中断，而 synchronized 却不行，使用synchronized 时，等待的线程会一直等待下去，不能够响应中断。
4.通过 Lock 可以知道有没有成功获取锁，而 synchronized 却无法办到。
5.Lock 可以提高多个线程进行读操作的效率。在性能上来说，如果竞争资源不激烈，两者的性能是差不多的，而当竞争资源非常激烈时（即有大量线程同时竞争），此时 Lock 的性能要远远优于synchronized。

3. 线程间通信

线程间通信的模型有两种：共享内存和消息传递，以下方式都是基本这两种模型来实现的。
我们来基本一道面试常见的题目来分析场景。
两个线程，一个线程对当前数值加 1，另一个线程对当前数值减 1,要求用线程间通信。

3.1 synchronized方案

3.2 lock方案

线程通信 = synchronized方案+ lock方案

package com.zrj.unit.juc;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.junit.Test;

import java.util.concurrent.*;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 线程通信
 *
 * @author zrj
 * @since 2021/8/19
 **/
public class ThreadIncrease {
    // 自定义线程名称
    private static ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat("thread-%d").build();
    // 定义线程池
    private static ExecutorService pool = new ThreadPoolExecutor(30, 200, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(1024), namedThreadFactory, new ThreadPoolExecutor.AbortPolicy());

    @Test
    @SneakyThrows
    public void ThreadComm() {
        IncreaseSyn increaseSyn = new IncreaseSyn();
        IncreaseLock increaseLock = new IncreaseLock();

        pool.execute(() -> {
            for (int i = 0; i < 5; i++) {
                //increaseSyn.increment();
                increaseLock.increment();
            }
        });
        pool.execute(() -> {
            for (int i = 0; i < 5; i++) {
                //increaseSyn.decrement();
                increaseLock.decrement();
            }
        });
        Thread.sleep(10000);
    }

}

/**
 * 加锁自增减
 */
@Slf4j
class IncreaseLock {
    // 加减变量
    private int number = 0;
    // 可重入锁
    private ReentrantLock reentrantLock = new ReentrantLock();
    // 声明钥匙
    private Condition condition = reentrantLock.newCondition();

    /**
     * 加一
     */
    public void increment() {
        reentrantLock.lock();
        try {
            while (number != 0) {
                condition.await();
            }
            number++;
            log.info("[IncreaseLock] " + Thread.currentThread().getName() + "：加1成功，值为：" + number);

            condition.signalAll();
        } catch (Exception e) {
            log.error("[IncreaseLock] 系统异常：" + e);
        } finally {
            reentrantLock.unlock();
        }
    }

    /**
     * 减一
     */
    public synchronized void decrement() {
        reentrantLock.lock();
        try {
            while (number == 0) {
                condition.await();
            }
            number--;
            log.info("[IncreaseLock] " + Thread.currentThread().getName() + "：减1成功，值为：" + number);

            condition.signalAll();
        } catch (Exception e) {
            log.error("[IncreaseLock] 系统异常：" + e);
        } finally {
            reentrantLock.unlock();
        }
    }
}

/**
 * 同步自增减
 */
@Slf4j
class IncreaseSyn {
    private int number = 0;

    /**
     * 加一
     */
    public synchronized void increment() {
        try {
            while (number != 0) {
                this.wait();
            }
            number++;
            log.info("[IncreaseSyn] " + Thread.currentThread().getName() + "：加1成功，值为：" + number);
            notifyAll();
        } catch (Exception e) {
            log.error("[IncreaseSyn] 系统异常：" + e);
        }
    }

    /**
     * 减一
     */
    public synchronized void decrement() {
        try {
            while (number == 0) {
                this.wait();
            }
            number--;
            log.info("[IncreaseSyn] " + Thread.currentThread().getName() + "：减1成功，值为：" + number);
            notifyAll();
        } catch (Exception e) {
            log.error("[IncreaseSyn] 系统异常：" + e);
        }
    }
}

3.4 线程间定制化通信

问题: A 线程打印 5 次 A， B 线程打印 10 次 B， C 线程打印 15 次 C,按照此顺序循环 10 轮

package com.zrj.unit.juc;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.junit.Test;

import java.util.concurrent.*;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 线程定制化通信
 *
 * @author zrj
 * @since 2021/8/19
 **/
@Slf4j
public class ThreadCondition {
    // 通信对象：0-打印A，1-打印B，2-打印C
    private int number = 0;

    // 自定义线程名称
    private static ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat("thread-%d").build();
    // 定义线程池
    private static ExecutorService pool = new ThreadPoolExecutor(30, 200, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(1024), namedThreadFactory, new ThreadPoolExecutor.AbortPolicy());

    //声明锁
    private ReentrantLock lock = new ReentrantLock();
    private Condition conditionA = lock.newCondition();
    private Condition conditionB = lock.newCondition();
    private Condition conditionC = lock.newCondition();

    /**
     * 问题:
     * A 线程打印 5 次 A， B 线程打印 10 次 B， C 线程打印 15 次 C,
     * 按照此顺序循环 10 轮
     */
    @Test
    @SneakyThrows
    public void threadCustomized() {
        pool.execute(() -> {
            for (int i = 0; i < 10; i++) {
                printA(i);
            }
        });
        pool.execute(() -> {
            for (int i = 0; i < 10; i++) {
                printB(i);
            }
        });
        pool.execute(() -> {
            for (int i = 0; i < 10; i++) {
                printC(i);
            }
        });
        Thread.sleep(10000);
        log.info("输出完成");
    }

    /**
     * 打印A
     */
    public void printA(int j) {
        lock.lock();

        try {
            while (number != 0) {
                conditionA.await();
            }
            log.info(Thread.currentThread().getName() + " 输出A，第" + j + "次开始");

            for (int i = 0; i < 5; i++) {
                log.info("A");
            }

            // 开始打印B
            number = 1;
            // 唤醒B
            conditionB.signalAll();

        } catch (Exception e) {
            log.error("系统异常：" + e);
        } finally {
            lock.unlock();
        }
    }

    /**
     * 打印B
     */
    public void printB(int j) {
        lock.lock();

        try {
            while (number != 1) {
                conditionB.await();
            }
            log.info(Thread.currentThread().getName() + " 输出B，第" + j + "次开始");

            for (int i = 0; i < 10; i++) {
                log.info("B");
            }

            // 开始打印C
            number = 2;
            // 唤醒C
            conditionC.signalAll();

        } catch (Exception e) {
            log.error("系统异常：" + e);
        } finally {
            lock.unlock();
        }
    }


    /**
     * 打印C
     */
    public void printC(int j) {
        lock.lock();

        try {
            while (number != 2) {
                conditionC.await();
            }
            log.info(Thread.currentThread().getName() + " 输出C，第" + j + "次开始");

            for (int i = 0; i < 15; i++) {
                log.info("C");
            }

            // 开始打印A
            number = 0;
            // 唤醒A
            conditionA.signalAll();

        } catch (Exception e) {
            log.error("系统异常：" + e);
        } finally {
            lock.unlock();
        }
    }

}

4. 集合线程安全

4.1 ArrayList线程不安全

4.2 Vector线程安全

Vector 是矢量队列，它是 JDK1.0 版本添加的类。继承于AbstractList，实现了 List, RandomAccess, Cloneable 这些接口。 Vector 继承了 AbstractList，实现了 List；所以， 它是一个队列，支持相关的添加、删除、修改、遍历等功能。
Vector 实现了 RandmoAccess 接口，即提供了随机访问功能。RandmoAccess 是 java 中用来被 List 实现，为 List 提供快速访问功能的。在Vector 中，我们即可以通过元素的序号快速获取元素对象；这就是快速随机访问。 Vector 实现了Cloneable 接口，即实现 clone()函数。它能被克隆。

4.3 Collections.synchronizedList

Collections 提供了方法 synchronizedList 保证 list 是同步线程安全的。

4.4 CopyOnWriteArrayList(重点)

CopyOnWriteArrayList它相当于线程安全的 ArrayList。和 ArrayList 一样，它是个可变数组；但是和ArrayList 不同的时，它具有以下特性：
1.它最适合于具有以下特征的应用程序： List 大小通常保持很小，只读操作远多于可变操作，需要在遍历期间防止线程间的冲突。
2. 它是线程安全的。
3. 因为通常需要复制整个基础数组，所以可变操作（add()、set() 和 remove()等等）的开销很大。
4. 迭代器支持 hasNext(), next()等不可变操作，但不支持可变 remove()等操作。
5. 使用迭代器进行遍历的速度很快，并且不会与其他线程发生冲突。在构造迭代器时，迭代器依赖于不变的数组快照。

6.独占锁效率低：采用读写分离思想解决
7. 写线程获取到锁，其他写线程阻塞
8. 复制思想：当我们往一个容器添加元素的时候，不直接往当前容器添加，而是先将当前容器进行 Copy，复制出一个新的容器，然后新的容器里添加元素，添加完元素之后，再将原容器的引用指向新的容器。这时候会抛出来一个新的问题，也就是数据不一致的问题。如果写线程还没来得及写会内存，其他的线程就会读到了脏数据。

4.5 实现验证

package com.zrj.unit.juc;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import lombok.extern.slf4j.Slf4j;
import org.junit.Test;

import java.util.*;
import java.util.concurrent.*;

/**
 * 集合线程安全
 *
 * @author zrj
 * @since 2021/8/18
 **/
@Slf4j
public class ThreadCollectionTest {
    // 线程数量，也是循环次数
    private static final int threadCount = 1000;
    // 自定义线程名称
    private static ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat("thread-local-pool-%d").build();
    // 定义线程池
    private static ExecutorService pool = new ThreadPoolExecutor(30, 200, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(1024), namedThreadFactory, new ThreadPoolExecutor.AbortPolicy());
    // 倒计时锁存器 CountDownLatch
    private static final CountDownLatch latch = new CountDownLatch(threadCount);

    /**
     * ArrayList多线程不安全
     * 新增数据不到1000条
     */
    @Test
    public void arrayListTest() {
        List<String> array = new ArrayList<>(1100);

        // 开启100个线程递增
        for (int i = 0; i < threadCount; i++) {
            pool.execute(() -> {
                try {
                    array.add(UUID.randomUUID().toString());
                } catch (Exception e) {
                    log.error("多线程递增异常，" + e);
                } finally {
                    latch.countDown();
                }
            });
        }

        // 倒计时锁存器 CountDownLatch
        try {
            latch.await();
        } catch (Exception e) {
            log.error("多线程递增异常，" + e);
        }

        log.info("子线程都执行完毕，继续执行主线程");
        log.info("array=" + array.size());
    }

    /**
     * vector多线程安全
     * 成功放入1000条数据
     */
    @Test
    public void arrayListRunable() throws InterruptedException {
        List<String> vector = new Vector<>(1000);

        for (int i = 0; i < 1000; i++) {
            pool.execute(() -> {
                vector.add(UUID.randomUUID().toString());
            });
        }
        Thread.sleep(10000);
        log.info("vector=" + vector.size());
    }

    /**
     * 多线程操作集合
     * synList=1000
     */
    @Test
    public void synList() throws InterruptedException {
        List<String> synList = Collections.synchronizedList(new ArrayList<>());

        for (int i = 0; i < 1000; i++) {
            pool.execute(() -> synList.add(UUID.randomUUID().toString()));
        }
        Thread.sleep(5000);
        log.info("synList=" + synList.size());
    }

    /**
     * 多线程操作集合
     * synList=1000
     */
    @Test
    public void copyList() throws InterruptedException {
        List<String> copyList = new CopyOnWriteArrayList();

        for (int i = 0; i < 1000; i++) {
            pool.execute(() -> copyList.add(UUID.randomUUID().toString()));
        }
        Thread.sleep(5000);
        log.info("copyList=" + copyList.size());
    }

}

5. 多线程锁

5.1 锁的八个问题

一个对象里面如果有多个 synchronized 方法，某一个时刻内，只要一个线程去调用其中的一个 synchronized 方法了，其它的线程都只能等待，换句话说，某一个时刻内，只能有唯一一个线程去访问这些synchronized 方法锁的是当前对象this，被锁定后，其它的线程都不能进入到当前对象的其它的synchronized 方法加个普通方法后发现和同步锁无关换成两个对象后，不是同一把锁了，情况立刻变化。
synchronized 实现同步的基础： Java 中的每一个对象都可以作为锁。

具体表现为以下 3 种形式。
对于普通同步方法，锁是当前实例对象。
对于静态同步方法，锁是当前类的 Class 对象。
对于同步方法块，锁是 Synchonized 括号里配置的对象

当一个线程试图访问同步代码块时，它首先必须得到锁，退出或抛出异常时必须释放锁。也就是说如果一个实例对象的非静态同步方法获取锁后，该实例对象的其他非静态同步方法必须等待获取锁的方法释放锁后才能获取锁，可是别的实例对象的非静态同步方法因为跟该实例对象的非静态同步方法用的是不同的锁，所以毋须等待该实例对象已获取锁的非静态同步方法释放锁就可以获取他们自己的锁。所有的静态同步方法用的也是同一把锁——类对象本身，这两把锁是两个不同的对象，所以静态同步方法与非静态同步方法之间是不会有竞态条件的。但是一旦一个静态同步方法获取锁后，其他的静态同步方法都必须等待该方法释放锁后才能获取锁，而不管是同一个实例对象的静态同步方法之间，还是不同的实例对象的静态同步方法之间，只要它们同一个类的实例对象！

package com.zrj.unit.juc;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import lombok.SneakyThrows;
import org.junit.Test;

import java.util.concurrent.*;

/**
 * 多线程同步问题
 *
 * @author zrj
 * @since 2021/8/20
 **/
public class ThreadPhoneSyn {
    // 自定义线程名称
    private static ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat("thread-%d").build();
    // 定义线程池
    private static ExecutorService pool = new ThreadPoolExecutor(30, 200, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(1024), namedThreadFactory, new ThreadPoolExecutor.AbortPolicy());

    @Test
    @SneakyThrows
    public void sendThreadTest() {
        ThreadPhoneSyn threadPhoneSyn1 = new ThreadPhoneSyn();
        ThreadPhoneSyn threadPhoneSyn2 = new ThreadPhoneSyn();

        //pool.execute(() -> threadPhoneSyn1.sendSMS());
        //pool.execute(() -> threadPhoneSyn1.sendEmail());

        // 静态方法，两个对象，SMS停留4秒，email-hello-sms
        pool.execute(() -> threadPhoneSyn1.sendSMSStatic());
        pool.execute(() -> threadPhoneSyn2.sendEmail());
        pool.execute(() -> sendHello());

        Thread.sleep(5000);
    }

    @SneakyThrows
    public static synchronized void sendSMSStatic() {
        //停留4秒
        TimeUnit.SECONDS.sleep(4);
        System.out.println("------sendSMS");
    }
    @SneakyThrows
    public synchronized void sendSMS() {
        //停留4秒
        TimeUnit.SECONDS.sleep(4);
        System.out.println("------sendSMS");
    }

    public synchronized void sendEmail() {
        System.out.println("------sendEmail");
    }

    public void sendHello() {
        System.out.println("------sendHello");
    }
}

6. Callable&Future 接口

6.1 Callable 接口

目前我们学习了有两种创建线程的方法-一种是通过创建Thread 类，另一种是通过使用 Runnable 创建线程。但是， Runnable 缺少的一项功能是，当线程终止时（即 run（）完成时），我们无法使线程返回结果。为了支持此功能，Java 中提供了 Callable 接口。

6.2 Future 接口

当 call（）方法完成时，结果必须存储在主线程已知的对象中，以便主线程可以知道该线程返回的结果。为此，可以使用 Future 对象。将 Future 视为保存结果的对象–它可能暂时不保存结果，但将来会保存（一旦Callable 返回）。 Future 基本上是主线程可以跟踪进度以及其他线程的结果的一种方式。要实现此接口，必须重写 5 种方法，这里列出了重要的方法,如下:

6.3 实现验证

package com.zrj.unit.juc;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.junit.Test;

import java.util.concurrent.*;

/**
 * Callable&Future 接口
 *
 * @author zrj
 * @since 2021/8/20
 **/
public class CallableThread {
    // 自定义线程名称
    private static ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat("thread-%d").build();
    // 定义线程池
    private static ExecutorService pool = new ThreadPoolExecutor(30, 200, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(1024), namedThreadFactory, new ThreadPoolExecutor.AbortPolicy());

    @Test
    @SneakyThrows
    public void sendThreadTest() {
        // 实现Runnable
        pool.execute(() -> System.out.println("实现Runnable"));

        // 获取future，future不阻塞，get阻塞
        Future<String> future = pool.submit(() -> new CallableThreadDemo().call());
        System.out.println("通过Future对象，不阻塞，获取get结果阻塞");

        // 实现Callable，get会一直阻塞，直到拿到结果
        String result = future.get();
        System.out.println("阻塞结果：" + result);
    }
}

/**
 * callable线程
 */
@Slf4j
class CallableThreadDemo implements Callable<String> {
    @Override
    public String call() {
        try {
            log.info(Thread.currentThread().getName() + "：线程进入call方法，进入休眠");
            Thread.sleep(3000);
        } catch (Exception e) {
            log.error("系统异常：" + e);
        }
        return "success：" + System.currentTimeMillis();
    }
}

7. JUC 三大辅助类: CountDownLatch CyclicBarrier Semaphore

JUC 中提供了三种常用的辅助类，通过这些辅助类可以很好的解决线程数量过多时 Lock 锁的频繁操作。这三种辅助类为：
? CountDownLatch: 倒计时锁存器
? CyclicBarrier: 循环栅栏
? Semaphore: 信号灯

7.1 倒计时锁存器 CountDownLatch

CountDownLatch 类可以设置一个计数器，然后通过ountDown 方法来进行减 1 的操作，使用 await 方法等待计数器不大于 0，然后继续执行 await 方法之后的语句。
? CountDownLatch 主要有两个方法，当一个或多个线程调用 await 方法时，这些线程会阻塞
? 其它线程调用 countDown 方法会将计数器减 1(调用countDown 方法的线程不会阻塞)
? 当计数器的值变为 0 时，因 await 方法阻塞的线程会被唤醒，继续执行。

package com.zrj.unit.juc;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.junit.Test;

import java.util.concurrent.*;

/**
 * CountDownLatch 倒计时锁存器
 *
 * @author zrj
 * @since 2021/8/20
 **/
public class CountDownLatchTest {
    // 线程数量，也是循环次数
    private static int threadCount = 10;
    // 自定义线程名称
    private static ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat("thread-%d").build();
    // 定义线程池
    private static ExecutorService pool = new ThreadPoolExecutor(30, 200, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(1024), namedThreadFactory, new ThreadPoolExecutor.AbortPolicy());
    // 倒计时锁存器
    private static CountDownLatch latch = new CountDownLatch(threadCount);

    @Test
    public void latchTest() {
        // 开启多线程处理
        for (int i = 0; i < threadCount; i++) {
            pool.execute(() -> {
                try {
                    String threadName = Thread.currentThread().getName();
                    if ("thread-6".equals(threadName)) {
                        Thread.sleep(2000);
                    }
                    System.out.println("线程" + threadName + "执行完成");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    latch.countDown();
                }
            });
        }
        // 等待所有线程到达放行
        try {
            latch.await();
        } catch (Exception e) {
            System.out.println("多线程递增异常，" + e);
        }
        System.out.println("所有线程执行完成，继续执行主线程");
    }
}

7.2 循环栅栏 CyclicBarrier

CyclicBarrier 看英文单词可以看出大概就是循环阻塞的意思，在使用中CyclicBarrier 的构造方法第一个参数是目标障碍数，每次执行 CyclicBarrier 一次障碍数会加一，如果达到了目标障碍数，才会执行 cyclicBarrier.await()之后的语句。可以将 CyclicBarrier 理解为加 1 操作

package com.zrj.unit.juc;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.junit.Test;

import java.util.concurrent.*;

/**
 * CyclicBarrier 循环栅栏
 *
 * @author zrj
 * @since 2021/8/20
 **/
public class CyclicBarrierTest {
    // 定义龙珠总数
    private static int dragonBallCount = 7;
    // 线程数量，也是循环次数
    private static int threadCount = 7;
    // 自定义线程名称
    private static ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat("thread-%d").build();
    // 定义线程池
    private static ExecutorService pool = new ThreadPoolExecutor(30, 200, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(1024), namedThreadFactory, new ThreadPoolExecutor.AbortPolicy());

    @Test
    public void summonShenlong() {
        // 定义循环栅栏
        CyclicBarrier cyclicBarrier = new CyclicBarrier(dragonBallCount, () -> {
            System.out.println("集齐" + dragonBallCount + "颗龙珠，召唤神龙！！！");
        });

        for (int i = 0; i < threadCount; i++) {
            pool.execute(() -> {
                try {
                    String threadName = Thread.currentThread().getName();
                    System.out.println("线程" + threadName + "开始收集龙珠");
                    // 集齐七颗龙珠召唤神龙
                    cyclicBarrier.await();
                } catch (Exception e) {
                    System.out.println("系统异常：" + e);
                }
            });
        }
    }
}

7.3 Semaphore

Semaphore 的构造方法中传入的第一个参数是最大信号量（可以看成最大线程池），每个信号量初始化为一个最多只能分发一个许可证。使用 acquire 方法获得许可证， release 方法释放许可场景: 抢车位, 6 部汽车 3 个停车位

package com.zrj.unit.juc;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import lombok.SneakyThrows;
import org.junit.Test;

import java.util.concurrent.*;

/**
 * Semaphore 信号量
 *
 * @author zrj
 * @since 2021/8/20
 **/
public class SemaphoreTest {
    // 定义3个停车位
    private static int carParking = 3;
    // 线程数量，也是循环次数
    private static int threadCount = 6;
    // 自定义线程名称
    private static ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat("thread-%d").build();
    // 定义线程池
    private static ExecutorService pool = new ThreadPoolExecutor(30, 200, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(1024), namedThreadFactory, new ThreadPoolExecutor.AbortPolicy());

    /**
     * 信号量抢车位
     */
    @Test
    @SneakyThrows
    public void semaphoreGrabParking() {
        // 定义3个停车位
        Semaphore semaphore = new Semaphore(carParking);
        // 模拟6辆汽车
        for (int i = 0; i < threadCount; i++) {
            pool.execute(() -> {
                String threadName = Thread.currentThread().getName();
                try {
                    System.out.println(threadName + "找车位ing...");
                    // 占用资源
                    semaphore.acquire();
                    System.out.println(threadName + "停车成功");

                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    System.out.println(threadName + "溜了，溜了");
                    // 释放资源
                    semaphore.release();
                }
            });
        }
        Thread.sleep(10000);
        System.out.println("执行完成");
    }
}

8. 读写锁ReentrantReadWriteLock

现实中有这样一种场景：对共享资源有读和写的操作，且写操作没有读操作那么频繁。在没有写操作的时候，多个线程同时读一个资源没有任何问题，所以应该允许多个线程同时读取共享资源；但是如果一个线程想去写这些共享资源，就不应该允许其他线程对该资源进行读和写的操作了。针对这种场景，JAVA 的并发包提供了读写锁 ReentrantReadWriteLock，它表示两个锁，一个是读操作相关的锁，称为共享锁；一个是写相关的锁，称为排他锁.

在线程持有读锁的情况下，该线程不能取得写锁(因为获取写锁的时候，如果发现当前的读锁被占用，就马上获取失败，不管读锁是不是被当前线程持有)。
? 在线程持有写锁的情况下，该线程可以继续获取读锁（获取读锁时如果发现写锁被占用，只有写锁没有被当前线程占用的情况才会获取失败）。
原因: 当线程获取读锁的时候，可能有其他线程同时也在持有读锁，因此不能把获取读锁的线程“升级” 为写锁；而对于获得写锁的线程，它一定独占了读写锁，因此可以继续让它获取读锁，当它同时获取了写锁和读锁后，还可以先释放写锁继续持有读锁，这样一个写锁就“降级” 为了读锁。

package com.zrj.unit.juc;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantReadWriteLock;

/**
 * 可重入读写锁
 * 场景: 使用 ReentrantReadWriteLock 对一个 hashmap 进行读和写操作
 * hashmap本身是线程不安全的，若需要安全集合可以采用concurrentHashMap
 * 或者是对hashmap加锁操作
 *
 * @author zrj
 * @since 2021/8/20
 **/
public class ReentrantReadWriteLockTest {
    // 创建map集合
    private volatile Map<String, Object> map = new HashMap<>(16);
    // 定义读写锁
    private ReentrantReadWriteLock rwLock = new ReentrantReadWriteLock();

    /**
     * 写操作，写锁
     * 进行写操作时，所有读与写操作都要等待
     */
    public void put(String key, Object value) {
        // 锁定资源
        rwLock.writeLock().lock();
        String threadName = Thread.currentThread().getName();
        try {
            System.out.println(threadName + "正在进行写入操作");
            TimeUnit.MICROSECONDS.sleep(300);
            map.put(key, value);
            System.out.println(threadName + "正在进行写入成功");
        } catch (Exception e) {
            System.out.println("系统异常：" + e);
        } finally {
            // 释放锁资源
            rwLock.writeLock().unlock();
        }
    }

    /**
     * 读操作,读锁
     * 可以并发读，互不影响
     */
    public void get(String key) {
        // 锁定资源
        rwLock.readLock().lock();
        String threadName = Thread.currentThread().getName();
        try {
            System.out.println(threadName + "正在进行读作");
            TimeUnit.MICROSECONDS.sleep(300);
            map.get(key);
            System.out.println(threadName + "正在进行读成功");
        } catch (Exception e) {
            System.out.println("系统异常：" + e);
        } finally {
            // 释放锁资源
            rwLock.readLock().unlock();
        }
    }


}

9. 阻塞队列

9.1 BlockingQueue 简介

Concurrent 包中， BlockingQueue 很好的解决了多线程中，如何高效安全“传输” 数据的问题。通过这些高效并且线程安全的队列类，为我们快速搭建高质量的多线程程序带来极大的便利。本文详细介绍了 BlockingQueue 家庭中的所有成员，包括他们各自的功能以及常见使用场景。阻塞队列，顾名思义，首先它是一个队列, 通过一个共享的队列，可以使得数据由队列的一端输入，从另外一端输出；

9.2 BlockingQueue 核心方法

9.3 实现验证

package com.zrj.unit.juc;

import lombok.SneakyThrows;
import org.junit.Test;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * 阻塞队列
 *
 * @author zrj
 * @since 2021/8/20
 **/
public class BlockingQueueTest {

    @Test
    @SneakyThrows
    public void queueTest() {
        //创建阻塞队列
        BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(3);

        //第一组
        //System.out.println(blockingQueue.add("a"));
        //System.out.println(blockingQueue.add("b"));
        //System.out.println(blockingQueue.add("c"));
        //System.out.println(blockingQueue.element());

        //System.out.println(blockingQueue.add("w"));// 队列满了再放数据抛异常，IllegalStateException: Queue full
        //System.out.println(blockingQueue.remove());
        //System.out.println(blockingQueue.remove());
        //System.out.println(blockingQueue.remove());
        //System.out.println(blockingQueue.remove());// 删除队列不存在数据，数据不存抛异常，NoSuchElementException

        //第二组
        //System.out.println(blockingQueue.offer("a"));
        //System.out.println(blockingQueue.offer("b"));
        //System.out.println(blockingQueue.offer("c"));
        //System.out.println(blockingQueue.offer("www"));// 队列满了再放数据放不进去，不会抛异常
        //
        //System.out.println(blockingQueue.poll());
        //System.out.println(blockingQueue.poll());
        //System.out.println(blockingQueue.poll());
        //System.out.println(blockingQueue.poll());// 删除队列不存在数据，数据不存不抛异常，返回null

        //第三组
        //blockingQueue.put("a");
        //blockingQueue.put("b");
        //blockingQueue.put("c");
        blockingQueue.put("w"); // 队列满了再放数据放不进去，不会抛异常
        //
        //System.out.println(blockingQueue.take());
        //System.out.println(blockingQueue.take());
        //System.out.println(blockingQueue.take());
        //System.out.println(blockingQueue.take());// 队列已满，再放数据，一直阻塞

        //第四组
        System.out.println(blockingQueue.offer("a"));
        System.out.println(blockingQueue.offer("b"));
        System.out.println(blockingQueue.offer("c"));
        // 队列已满，再放数据，一直阻塞，超时释放，返回false
        System.out.println(blockingQueue.offer("w", 3L, TimeUnit.SECONDS));
    }

}

10. ThreadPool 线程池

Java多线程之线程池ThreadPool详解：
https://blog.csdn.net/m0_37583655/article/details/119828433

11. Fork/Join 框架

Fork/Join 它可以将一个大的任务拆分成多个子任务进行并行处理，最后将子任务结果合并成最后的计算结果，并进行输出。 Fork/Join 框架要完成两件事情：
ForkJoin的使用需要根据实际的业务场景来判断是否会有性能提升，毕竟拆分与合并也是需要耗费性能的，如果只是简单计算1到100的和，直接计算或许性能会更好些。这个从以下验证可以看出来。

package com.zrj.unit.juc;

import lombok.SneakyThrows;
import org.junit.Test;

import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
import java.util.concurrent.RecursiveTask;

/**
 * 切分合并任务
 * ForkJoin的使用需要根据实际的业务场景来判断是否会有性能提升，毕竟拆分与合并也是需要耗费性能的，
 * 如果只是简单计算1到100的和，直接计算或许性能会更好些。这个从以下验证可以看出来
 * 场景: 生成一个计算任务，计算 1+2+3.........+100,每 10 个数切分一个子任务
 *
 * @author zrj
 * @since 2021/8/20
 **/
public class ForkJoinTest {

    /**
     * 直接计算
     * 耗时：0, 合并结果：5050
     */
    @Test
    public void calculationNumberTest() {
        long beginTime = System.currentTimeMillis();

        int begin = 0;//开始值
        int end = 100;//结束值
        int result = 0; //返回结果

        for (int i = begin; i <= end; i++) {
            result += i;
        }
        System.out.println("耗时：" + (System.currentTimeMillis() - beginTime) + ", 结果：" + result);
    }

    /**
     * 拆分合并计算
     * 耗时：2, 合并结果：5050
     */
    @Test
    @SneakyThrows
    public void forkJoinTest() {
        long beginTime = System.currentTimeMillis();

        //创建MyTask对象
        RecursiveTaskTest myTask = new RecursiveTaskTest(0, 100);
        //创建分支合并池对象
        ForkJoinPool forkJoinPool = new ForkJoinPool();
        ForkJoinTask<Integer> forkJoinTask = forkJoinPool.submit(myTask);
        //获取最终合并之后结果
        Integer result = forkJoinTask.get();
        System.out.println("耗时：" + (System.currentTimeMillis() - beginTime) + ", 合并结果：" + result);
        //关闭池对象
        forkJoinPool.shutdown();
    }
}

/**
 * 拆分合并
 * 每10个一组，拆分计算，合并
 */
class RecursiveTaskTest extends RecursiveTask<Integer> {
    //拆分差值不能超过10，计算10以内运算
    private static final Integer VALUE = 10;
    private int begin;//拆分开始值
    private int end;//拆分结束值
    private int result; //返回结果

    public RecursiveTaskTest(int begin, int end) {
        this.begin = begin;
        this.end = end;
    }

    @Override
    protected Integer compute() {
        if ((end - begin) <= VALUE) {
            for (int i = begin; i <= end; i++) {
                result = result + i;
            }
        } else {
            //获取中间值，左右拆分
            int middle = (begin + end) / 2;
            RecursiveTaskTest task1 = new RecursiveTaskTest(begin, middle);
            RecursiveTaskTest task2 = new RecursiveTaskTest(middle + 1, end);
            //调用方法拆分
            task1.fork();
            task2.fork();
            //合并结果
            result = task1.join() + task2.join();
        }
        return result;
    }
}

12. CompletableFuture

12.1 CompletableFuture简介

CompletableFuture 在 Java 里面被用于异步编程，异步通常意味着非阻塞，可以使得我们的任务单独运行在与主线程分离的其他线程中，并且通过回调可以在主线程中得到异步任务的执行状态，是否完成，和是否异常等信息。
CompletableFuture 实现了 Future, CompletionStage 接口，实现了 Future接口就可以兼容现在有线程池框架，而CompletionStage 接口才是异步编程的接口抽象，里面定义多种异步方法，通过这两者集合，从而打造出了强大的CompletableFuture 类。

12.2 Future 与 CompletableFuture

Futrue 在 Java 里面，通常用来表示一个异步任务的引用，比如我们将任务提交到线程池里面，然后我们会得到一个 Futrue，在 Future 里面有 isDone 方法来 判断任务是否处理结束，还有 get 方法可以一直阻塞直到任务结束然后获取结果，但整体来说这种方式，还是同步的，因为需要客户端不断阻塞等待或者不断轮询才能知道任务是否完成。

Future 的主要缺点如下：
（1）不支持手动完成我提交了一个任务，但是执行太慢了，我通过其他路径已经获取到了任务结果，现在没法把这个任务结果通知到正在执行的线程，所以必须主动取消或者一直等待它执行完成
（2）不支持进一步的非阻塞调用通过 Future 的 get 方法会一直阻塞到任务完成，但是想在获取任务之后执行额外的任务，因为 Future 不支持回调函数，所以无法实现这个功能
（3）不支持链式调用对于 Future 的执行结果，我们想继续传到下一个 Future 处理使用，从而形成一个链式的 pipline 调用，这在 Future 中是没法实现的。
（4）不支持多个 Future 合并比如我们有 10 个 Future 并行执行，我们想在所有的 Future 运行完毕之后，执行某些函数，是没法通过 Future 实现的。
（5）不支持异常处理Future 的 API 没有任何的异常处理的 api，所以在异步运行时，如果出了问题是不好定位。

12.3 实现验证

package com.zrj.unit.juc;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import lombok.SneakyThrows;
import org.junit.Test;

import java.util.concurrent.*;

/**
 * 异步编程 CompletableFuture
 *
 * @author zrj
 * @since 2021/8/20
 **/
public class CompletableFutureTest {
    private static Integer num = 10;
    // 自定义线程名称
    private static ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat("thread-local-pool-%d").build();
    // 定义线程池
    private static ExecutorService pool = new ThreadPoolExecutor(30, 200, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(1024), namedThreadFactory, new ThreadPoolExecutor.AbortPolicy());

    /**
     * CompletableFuture阻塞接收
     * 主线程里面创建一个 CompletableFuture，然后主线程调用 get 方法会阻塞，
     * 最后我们在一个子线程中使其终止
     */
    @Test
    @SneakyThrows
    public void futureGetStrTest() {
        CompletableFuture<String> future = new CompletableFuture();
        pool.execute(() -> {
            try {
                System.out.println(Thread.currentThread().getName() + " future");
                Thread.sleep(3000);
                future.complete("success");
            } catch (InterruptedException e) {
                System.out.println("系统异常：" + e);
            }
        });
        //主线程调用 get 方法阻塞
        System.out.println("主线程调用 get 方法获取结果为: " + future.get());
        System.out.println("主线程完成,阻塞结束!!!!!!");
    }

    @Test
    @SneakyThrows
    public void futureGetVoidTest() {
        System.out.println("主线程启动：" + Thread.currentThread().getName());
        //同步调用
        CompletableFuture<Void> future1 = CompletableFuture.runAsync(() -> {
            try {
                System.out.println("子线程启动：" + Thread.currentThread().getName());
                Thread.sleep(3000);
                System.out.println("子线程完成：" + Thread.currentThread().getName());
            } catch (Exception e) {
                System.out.println("系统异常：" + e);
            }
        });
        future1.get();
        System.out.println("主线程结束：" + Thread.currentThread().getName());
    }

    /**
     * 依赖线程 thenApply
     * 执行顺醋：主线程->子线程->依赖线程
     * num = 10，子线程结果为：400
     */
    @Test
    @SneakyThrows
    public void futureRelyTest() {
        System.out.println("主线程启动：" + Thread.currentThread().getName());
        //同步调用
        CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
            System.out.println("子线程启动：" + Thread.currentThread().getName());
            try {
                num += 10;
            } catch (Exception e) {
                System.out.println("系统异常：" + e);
            }
            System.out.println("子线程结果：" + num);
            return num;
        }).thenApply(integer -> {
            int res = num * num;
            System.out.println("依赖线程执行结果：" + res);
            return res;
        });

        // 获取结果
        Integer result = future.get();
        System.out.println("主线程结束，子线程结果为：" + result);
    }

    /**
     * 消费处理结果 thenAccept
     */
    @Test
    @SneakyThrows
    public void futureAcceptTest() {
        System.out.println("主线程启动：" + Thread.currentThread().getName());
        //同步调用
        CompletableFuture.supplyAsync(() -> {
            System.out.println("子线程启动：" + Thread.currentThread().getName());
            try {
                num += 10;
            } catch (Exception e) {
                System.out.println("系统异常：" + e);
            }
            System.out.println("子线程结果：" + num);
            return num;
        }).thenApply(integer -> {
            int res = num * num;
            System.out.println("依赖线程执行结果：" + res);
            return res;
        }).thenAccept(integer -> {
            integer = integer - 10;
            System.out.println("子线程全部处理完，最后调用thenAccept结果为：" + integer);
        });
        System.out.println("主线程结束：" + Thread.currentThread().getName());
    }

    /**
     * 异常处理 exceptionally
     */
    @Test
    @SneakyThrows
    public void futureGetExceptionTest() {
        System.out.println("主线程启动：" + Thread.currentThread().getName());

        //异步调用
        CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName() + " future2");
            //模拟异常
            int i = 10 / 0;
            return 1024;
        }).exceptionally(ex -> {
            System.out.println(ex.getMessage());
            return -1;
        });

        // 获取结果
        Integer result = future.get();
        System.out.println("主线程结束，子线程结果为：" + result);
    }

    /**
     * 异常处理 Handle
     */
    @Test
    @SneakyThrows
    public void futureGetHandleTest() {
        System.out.println("主线程启动：" + Thread.currentThread().getName());

        //异步调用
        CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName() + " future2");
            //模拟异常
            int i = 10 / 0;
            return 1024;
        }).handle((i, ex) -> {
            System.out.println("进入handle方法");
            if (ex != null) {
                System.out.println("方法异常：" + ex.getMessage());
                return -1;
            } else {
                System.out.println("方法正常：" + i);
                return i;
            }
        });

        // 获取结果
        Integer result = future.get();
        System.out.println("主线程结束，子线程结果为：" + result);
    }

    /**
     * 结果合并
     */
    @Test
    @SneakyThrows
    public void futureComposeTest() {
        System.out.println("主线程启动：" + Thread.currentThread().getName());

        //异步调用
        CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName() + " future");
            return 1000;
        });
        //合并
        CompletableFuture<Integer> future1 = future.thenCompose(integer -> CompletableFuture.supplyAsync(() -> integer + 1000));

        //获取结果
        System.out.println("主线程结束，future结果为：" + future.get());
        System.out.println("主线程结束，future1合并结果为：" + future1.get());
    }
}