✨作者：猫十二懿

❤️‍🔥账号：CSDN 、掘金 、个人博客 、Github

🎉公众号：猫十二懿

学习地址

写在最前

JUC并发编程（上）

JUC（Java Util Concurrent）学习内容框架：

6、callable 接口

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

6.1 创建线程的第三种方案—Callable 接口

Callable 接口特点：

• 为了实现 Runnable，需要实现不返回任何内容的 run())方法，而对于 Callable，需要实现在完成时返回结果的 cal()方法。 •
• call() 方法可以引发异常，而 run() 则不能。
• 为实现 Callable 而必须重写 call()方法
• 不能直接替换 runnable，因为 Thread 类的构造方法根本没有 Callable

6.2 Future 接口

当 call() 方法完成时，结果必须存储在主线程已知的对象中，以便主线程可 以知道该线程返回的结果。为此，可以使用 Future 对象。

将 Future 视为保存结果的对象，它可能暂时不保存结果，一旦Callable 返回将会进行保存。Future 基本上是主线程可以跟踪进度以及得到其他线程的结果的一种方式。要实现此接口，必须重写 5 种方法，这里列出了重要的方法，如下:

1. public boolean cancel（boolean mayInterrupt）：用于停止任务。如果尚未启动，它将停止任务。如果已启动，则仅在 mayInterrupt 为 true 时才会中断任务。
2. public Object get() 抛出 InterruptedException，ExecutionException： 用于获取任务的结果。如果任务完成，它将立即返回结果，否则将等待任务完成，然后返回结果。
3. public boolean isDone() ：如果任务完成，则返回 true，否则返回 false

可以看到 Callable 和 Future 做两件事，Callable 与 Runnable 类似，因为它封装了要在另一个线程上运行的任务，而 Future 用于存储从另一个线程获得的结果。实际上，future 也可以与 Runnable 一起使用。

6.3 FutureTask

Java 库具有具体的 FutureTask 类型，该类型实现 Runnable 和 Future，并方便地将两种功能组合在一起。 可以通过为其构造函数提供 Callable 来创建 FutureTask。然后，将 FutureTask 对象提供给 Thread 的构造函数以创建 Thread 对象。因此，间接地使用 Callable 创建线程。

核心原理:(重点)

1. 在主线程中需要执行比较耗时的操作时，但又不想阻塞主线程时，可以把这些任务交给 Future 对象在后台完成，
2. 当主线程将来需要时，就可以通过 Future 对象获得后台任务的计算结果或者执行状态，
3. 一般 FutureTask 多用于耗时的计算，主线程可以在完成自己的任务后，再去获取结果，
4. 仅在计算完成时才能检索结果；如果计算尚未完成，则阻塞 get 方法，一旦计算完成，就不能再重新开始或取消计算 ，
5. get 方法获取结果只有在计算完成时获取，否则会一直阻塞直到任务转入完成状态，然后会返回结果或者抛出异常 ，
6. get 只计算一次，因此 get 方法放到最后。

例子说明：

1. 我在敲代码，突然口渴了，出去买水不太合适（打断思路），我要继续敲代码，只能单独的叫一个人去帮我买水回来，然后放到我的桌子上，我就可以直接喝。（不影响主线程的情况下，单独开启一个单线程完成任务）
2. 我口渴了，叫人去买水，但是他第一次买的水不是我想喝的，后面他又在去一趟买水，买了几趟才买到想喝的水，如果我下次还叫他去帮忙买水，那么他就可以直接买我喜欢喝的就可以了，不用再折腾那么多次。（也就是汇总一次，计算一次）

// Runnable 创建线程
class RunnableThread implements Runnable {
    @Override
    public void run() {
    }
}
// Callable 接口 , 有返回值
class CallableThread implements Callable {
    @Override
    public Integer call() throws Exception {
        System.out.println(Thread.currentThread().getName() + "进入callable");
        return 2222;
    }
}
public class CallableDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // Runnable 创建线程
        new Thread(new RunnableThread(), "Runnable创建线程").start();
        // Callable 接口创建线程, FutureTask未来任务
        FutureTask<Integer> futureTask = new FutureTask<>(new CallableThread());
        // 简化写法，FutureTask是一个函数式接口的类
        FutureTask<Integer> ft = new FutureTask<>(() -> {
            System.out.println(Thread.currentThread().getName() + "进入callable");
            return 1024;
        });
        // 创建线程
        new Thread(ft, "ft").start();
        new Thread(futureTask, "futureTask").start();
        // 没有计算完成就一直等待
        while (!ft.isDone()) {
            System.out.println("还在计算.....");
        }
        // 计算完成则会进行汇总 get到结果
        System.out.println(ft.get()); // 1024
        System.out.println(futureTask.get()); // 2222
        System.out.println(Thread.currentThread().getName() + "结束"); // main线程结束
    }
}

7、JUC三大辅助类

JUC 中提供了三种常用的辅助类，通过这些辅助类可以很好的解决线程数量过 多时 Lock 锁的频繁操作。这三种辅助类为：

1. CountDownLatch: 减少计数
2. CyclicBarrier: 循环栅栏
3. Semaphore: 信号灯

7.1 减少计数 CountDownLatch

CountDownLatch 类可以设置一个计数器，然后通过 countDown 方法来进行减 1 的操作，使用 await 方法等待计数器不大于 0，然后继续执行 await 方法之后的语句。

• CountDownLatch 主要有两个方法，当一个或多个线程调用 await 方法时，这些线程会阻塞
• 其它线程调用 countDown 方法会将计数器减 1(调用 countDown 方法的线程不会阻塞)
• 当计数器的值变为 0 时，因 await 方法阻塞的线程会被唤醒，继续执行

实际例子说明：

• 6 个同学陆续离开教室后值班同学才可以关门。

/**
 * @author Shier 2023/2/21 22:11
 * 模拟教室人走完关灯效果
 */
public class CountDownLatchDemo {
    public static void main(String[] args) {
        // 总共10个人，走完了才可以关灯
        // 使用CountDownLatch 进行初始化数据
        CountDownLatch countDownLatch = new CountDownLatch(10);
        for (int i = 10; i > 0; i--) {
            // 创建线程
            new Thread(() -> {
                System.out.println(Thread.currentThread().getName() + "学号学生离开教室");
                // 计数减一
                countDownLatch.countDown();
            }, String.valueOf(i)).start();
        }
        // 如果还没有为0 则会一直等待
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(Thread.currentThread().getName() + "关门走人了...");
    }
}

7.2 循环栅栏 CyclicBarrier

使用中 CyclicBarrier 的构造方法第一个参数是目标障碍数，每次执行 CyclicBarrier 一次障碍数会加一，如果达到了目标障碍数，才会执行 cyclicBarrier.await()之后 的语句。可以将 CyclicBarrier 理解为加 1 操作

例子：

• 收集七颗龙珠召唤神龙许愿

/**
 * @author Shier 2023/2/21 22:30
 */
public class CyclicBarrierDemo {
    // 创建目标到达的值
    private static final int NUMBER = 7;

    public static void main(String[] args) {
        // 创建CyclicBarrier对象
        CyclicBarrier cyclicBarrier = new CyclicBarrier(NUMBER, (() -> {
            System.out.println("集齐完毕，开始许愿，祝你愿望成真！");
        }));
        // 创建线程
        for (int i = 1; i <= 7; i++) {
            new Thread(() -> {
                try {
                    System.out.println(Thread.currentThread().getName() + "星珠到手！");
                    // 否则等待
                    cyclicBarrier.await();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }, String.valueOf(i)).start();
        }
    }
}

7.3 信号灯 Semaphore

Semaphore 的构造方法中传入的第一个参数是最大信号量（可以看成最大线 程池），每个信号量初始化为一个最多只能分发一个许可证。使用 acquire 方法获得许可证，获得许可之后其他的线程都处于阻塞状态，release 方法释放许可，其他线程方可获得许可证。

场景: 抢车位, 6 部汽车 3 个停车位

package com.shier.jucauxiliary;

import javax.swing.plaf.TableHeaderUI;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

/**
 * @author Shier 2023/2/21 22:45
 * 6 部汽车 3 个停车位
 */
public class SemaphoreDemo {
    public static void main(String[] args) {
        // 创建Semaphore 设置许可的数量
        Semaphore semaphore = new Semaphore(3);

        // 循环六次，创建六个线程
        for (int i = 1; i <= 6; i++) {
            new Thread(() -> {
                // 获得许可
                try {
                    semaphore.acquire();
                    System.out.println("-------" + Thread.currentThread().getName() + "号车寻找车位ing。");
                    // 设置一个随机时间来占用车位
                    Thread.sleep(2000);
                    System.out.println(Thread.currentThread().getName() + "号车获得车位。");
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    // 释放许可
                    System.out.println("*******" + Thread.currentThread().getName() + "号车离开车位。");
                    semaphore.release();
                }
            }, String.valueOf(i)).start();
        }
    }
}

8、读写锁

8.1 读写锁介绍

现实中有这样一种场景：对共享资源有读和写的操作，且写操作没有读操作那么频繁。在没有写操作的时候，多个线程同时读一个资源没有任何问题，所以应该允许多个线程同时读取共享资源；但是如果一个线程想去写这些共享资源， 就不应该允许其他线程对该资源进行读和写的操作了。

针对这种场景，JAVA 的并发包提供了读写锁 ReentrantReadWriteLock， 它表示两个锁，一个是读操作相关的锁，称为共享锁；一个是写相关的锁，称为排他锁

线程进入写锁的前提条件：

1. 没有其他线程的读锁
2. 没有其他线程的写锁

而读写锁有以下三个重要的特性：

（1）公平选择性：支持非公平（默认）和公平的锁获取方式，吞吐量还是非公 平优于公平。

（2）重进入：读锁和写锁都支持线程重进入。

（3）锁降级：遵循获取写锁、获取读锁再释放写锁的次序，写锁能够降级成为 读锁。

8.1.1 读锁发生死锁

8.1.2 写锁发生死锁

当线程1 在对第一个进行写操作，线程2对第二个进行写操作，由于是可以并发操作的，所以当线程1或者线程2对另外的一个线程正在写操作的数据进行修改时需要等待对方写操作完成才可以进行，此时就有可能会出现互相操作对方的写操作进而出现死锁。

锁的比较

读写锁的案例实现：

/**
 * @author Shier 2023/2/22 10:27
 */
class CacheDemo {

    // 创建一个map集合 volatile表示不断变化
    private volatile Map<String, Object> map = new HashMap<>();
    // 创建一个读写锁
    private ReadWriteLock readWriteLock = new ReentrantReadWriteLock();

    // 写入过程
    public void writeMethod(String key, Object value) {
        // 写锁上锁
        readWriteLock.writeLock().lock();
        try {
            System.out.println(Thread.currentThread().getName() + "正在写入内容" + key);
            TimeUnit.MICROSECONDS.sleep(300);
            // 写入内容
            map.put(key, value);
            System.out.println(Thread.currentThread().getName() + "写入完成" + key);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 释放锁
            readWriteLock.writeLock().unlock();
        }
    }

    // 读取过程
    public Object readMethod(String key) {

        // 读锁上锁
        readWriteLock.readLock().lock();
        Object result = null;
        try {
            System.out.println(Thread.currentThread().getName() + "正在读取内容" + key);
            TimeUnit.MICROSECONDS.sleep(300);
            // 读取内容
            result = map.get(key);
            System.out.println(Thread.currentThread().getName() + "读取到了" + key);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 释放锁
            readWriteLock.readLock().unlock();
        }
        return result;
    }
}

public class ReadWriteLockDemo {
    public static void main(String[] args) {
        CacheDemo cacheDemo = new CacheDemo();
        // 写
        for (int i = 1; i <= 10; i++) {
            final int number = i;
            new Thread(() -> {
                cacheDemo.writeMethod(String.valueOf(number), number);
            },String.valueOf(i)).start();
        }
        // 读
        for (int i = 1; i <= 10; i++) {
            final int number = i;
            new Thread(() -> {
                cacheDemo.readMethod(String.valueOf(number));
            },String.valueOf(i)).start();
        }
    }
}

8.1.3 读写锁降级

锁降级是指将当前持有的锁从高级别锁降为低级别锁的过程。在多线程并发执行的场景中，通常情况下我们会在业务处理的过程中加锁，为了避免死锁、性能问题等问题，有时候需要在代码执行的过程中进行锁降级。

锁降级的主要目的是为了在持有高级别锁的情况下，可以继续执行需要持有低级别锁才能进行的操作，而不需要释放高级别锁再重新获取低级别锁的操作。例如，在一段代码中持有了数据库的行级锁，为了执行某些操作需要进行表级锁的操作，可以通过锁降级将行级锁降为表级锁。

锁降级的操作一般有以下几个步骤：

1. 获取高级别锁；
2. 执行需要高级别锁才能执行的操作；
3. 获取低级别锁；
4. 释放高级别锁。
5. 使用低级别锁

在执行锁降级的过程中，需要保证高级别锁和低级别锁的顺序，先获取高级别锁再获取低级别锁，同时在释放锁的过程中也需要按照相反的顺序进行释放。如果顺序不正确，可能会导致死锁等问题。

总的来说，锁降级可以避免在执行代码的过程中频繁地获取和释放锁，提高并发性能和降低死锁等问题的发生。但是，在实际应用中需要根据具体情况进行评估和选择，避免过度使用锁降级导致代码难以维护。

大致的流程：

写锁的权限大于读锁的权限。

public class LowerLockDemo {
    public static void main(String[] args) {
        // 创建可重入锁对象
        ReentrantReadWriteLock readWriteLock = new ReentrantReadWriteLock();
        // 写锁
        ReentrantReadWriteLock.WriteLock writeLock = readWriteLock.writeLock();
        // 读锁
        ReentrantReadWriteLock.ReadLock readLock = readWriteLock.readLock();
        // 一般来说写锁大于读锁
        // 1.获取高级锁-写锁
        writeLock.lock();
        System.out.println("高级锁-写锁");
        // 2.获取低级锁-读锁
        readLock.lock();
        System.out.println("低级锁-读锁");
        // 3. 释放高级锁
        writeLock.unlock();
        // 4. 释放低级锁
        readLock.unlock();
    }
}

• 读锁不能升级为写锁
• 也就是低级锁不能升级为高级所

原因: 当线程获取读锁的时候，可能有其他线程同时也在持有读锁，因此不能把 获取读锁的线程“升级”为写锁；而对于获得写锁的线程，它一定独占了读写 锁，因此可以继续让它获取读锁，当它同时获取了写锁和读锁后，还可以先释 放写锁继续持有读锁，这样一个写锁就“降级”为了读锁。

8.2 悲观锁

悲观锁是一种保守的锁策略，其核心思想是在访问共享资源时，假定其他线程会对该资源进行修改，因此采取加锁的方式防止其他线程的干扰。悲观锁常见的实现方式是使用 synchronized 关键字或者 ReentrantLock 等锁对象。

当一个人进行操作时会上锁，阻塞别人，其他人就不能进行操作，只有当这个人把锁释放掉，其他人才可以进行操作，当其他人操作时也会重复以上的步骤。

悲观锁的优点：

1. 可以保证并发访问的安全性，避免数据的脏读、幻读等问题；
2. 悲观锁适用于写操作多、读操作少的场景，可以减少写冲突的发生，提高并发性能。

悲观锁的缺点：

1. 悲观锁会在访问共享资源时，阻塞其他线程的访问，降低并发性能；
2. 不支持并发操作，只能一个人一个人的进行操作，导致效率性能降低。
3. 悲观锁对性能的影响和加锁的范围有关，加锁范围过大，会导致并发性能下降；
4. 如果悲观锁的加锁粒度太小，会导致线程频繁加锁和释放锁，从而造成性能瓶颈。

总的来说，悲观锁虽然能够保证并发访问的安全性，但是在高并发场景下可能会导致性能问题，需要开发者在应用程序设计时进行权衡，选择适合的锁策略来提高并发性能

8.3 乐观锁

乐观锁是一种乐观的锁策略，其核心思想是假设在并发访问时，共享资源不会发生冲突，因此不会阻塞其他线程的访问，而是在更新共享资源时检查是否有其他线程对该资源进行修改。乐观锁常见的实现方式有版本号机制和CAS（Compare And Swap）算法等。

乐观锁的优点：

1. 在并发读操作多、写操作少的场景下，乐观锁可以提高并发性能；
2. 乐观锁不会阻塞其他线程的访问，避免了加锁的开销；
3. 乐观锁的实现方式比较简单，实现成本低。

乐观锁的缺点：

1. 乐观锁不能保证并发访问的安全性，存在数据的冲突、丢失等问题；
2. 乐观锁的实现方式需要在更新共享资源时检查是否有其他线程的干扰，如果存在干扰需要进行重试，这会增加一定的开销；
3. 乐观锁的适用场景有限，不适用于写操作比较频繁的场景。

总的来说，乐观锁在一些读操作多、写操作少的场景下可以提高并发性能，但是在一些高并发写操作的场景下可能会导致数据冲突，需要开发者在实际应用中进行评估和选择。

8.4 表锁、行锁

表锁是指对整个表进行加锁，当一个事务访问某个表时，会对整个表进行加锁，其他事务无法对该表进行更新、删除等操作。

优点：实现简单，对于一些只读的操作可以提高并发性能，

缺点：并发性能较低，会对整个表进行加锁，影响其他事务的操作。

行锁是指对表中的一行或多行进行加锁，当一个事务访问某个表时，只对需要访问的行进行加锁，其他行不受影响。

优点：并发性能较高，只对需要访问的行进行加锁，不会影响其他事务的操作

缺点：实现复杂，需要保证行级锁的粒度不会过细，否则会降低并发性能，可能会出现死锁状态。

在实际应用中，如果只有读操作，可以采用表锁提高并发性能；如果有较多的写操作，应该采用行锁避免数据冲突。在实现行锁时，需要注意锁粒度的问题，避免锁定过细或过粗。同时，为了提高并发性能，可以采用一些辅助手段，如缓存、索引等。

9、JUC阻塞队列

9.1 BlockingQueue 简介

Concurrent 包中，BlockingQueue 很好的解决了多线程中，如何高效安全 “传输”数据的问题。通过这些高效并且线程安全的队列类，为我们快速搭建 高质量的多线程程序带来极大的便利。

阻塞队列，顾名思义，首先它是一个队列（先进先出）， 通过一个共享的队列，可以使得数据由队列的一端输入，从另外一端输出；

• 当队列是空的，从队列中获取元素的操作将会被阻塞
• 当队列是满的，从队列中添加元素的操作将会被阻塞
• 试图从空的队列中获取元素的线程将会被阻塞，直到其他线程往空的队列插入新的元素
• 试图向已满的队列中添加新元素的线程将会被阻塞，直到其他线程从队列中移除一个或多个元素或者完全清空，使队列变得空闲起来并后续新增

常用的队列主要有以下两种：

• 先进先出（FIFO）：先插入的队列的元素也最先出队列，类似于排队的功能。 从某种程度上来说这种队列也体现了一种公平性
• 后进先出（LIFO）：后插入队列的元素最先出队列，这种队列优先处理最近发 生的事件(栈)

9.1.1 多线程的阻塞

在多线程领域：所谓阻塞，在某些情况下会挂起线程（即阻塞），一旦条件满足，被挂起 的线程又会自动被唤起。

使用 BlockingQueue 好处是我们不需要关心什么时候需要阻塞线程，什么时候需要唤醒线程，因为这一切 BlockingQueue 都给你一手包办了

在 concurrent 包发布以前，在多线程环境下，我们每个程序员都必须去自己控制这些细节，尤其还要兼顾效率和线程安全，而这会给我们的程序带来不小的复杂度。

多线程环境中，通过队列可以很容易实现数据共享，比如经典的“生产者”和 “消费者”模型中，通过队列可以很便利地实现两者之间的数据共享。假设我们有若干生产者线程，另外又有若干个消费者线程。如果生产者线程需要把准 备好的数据共享给消费者线程，利用队列的方式来传递数据，就可以很方便地 解决他们之间的数据共享问题。但如果生产者和消费者在某个时间段内，万一 发生数据处理速度不匹配的情况呢？理想情况下，如果生产者产出数据的速度 大于消费者消费的速度，并且当生产出来的数据累积到一定程度的时候，那么 生产者必须暂停等待一下（阻塞生产者线程），以便等待消费者线程把累积的 数据处理完毕，反之亦然。

• 当队列中没有数据的情况下，消费者端的所有线程都会被自动阻塞（挂起）， 直到有数据放入队列
• 当队列中填满数据的情况下，生产者端的所有线程都会被自动阻塞（挂起）， 直到队列中有空的位置，线程被自动唤醒

9.2 BlockingQueue 核心方法

9.2.1 放入数据

• offer(anObject)：表示如果可能的话，将 anObject 加到 BlockingQueue 里，即 如果 BlockingQueue 可以容纳,则返回 true,否则返回 false.（本方法不阻塞当 前执行方法的线程）
• offer(E o, long timeout, TimeUnit unit)：可以设定等待的时间，如果在指定 的时间内，还不能往队列中加入 BlockingQueue，则返回失败
• put(anObject)：把 anObject 加到 BlockingQueue 里，如果 BlockQueue 没有空间,则调用此方法的线程被阻断直到 BlockingQueue 里面有空间再继续

9.2.2 获取数据

• poll(time)：取走 BlockingQueue 里排在首位的对象,若不能立即取出,则可以等 time 参数规定的时间,取不到时返回 null
• poll(long timeout, TimeUnit unit)：从 BlockingQueue 取出一个队首的对象， 如果在指定时间内，队列一旦有数据可取，则立即返回队列中的数据。否则知 道时间超时还没有数据可取，返回失败。
• take()：取走 BlockingQueue 里排在首位的对象,若 BlockingQueue 为空,阻断 进入等待状态直到 BlockingQueue 有新的数据被加入;
• drainTo()：一次性从 BlockingQueue 获取所有可用的数据对象（还可以指定 获取数据的个数），通过该方法，可以提升获取数据效率；不需要多次分批加 锁或释放锁。

// 创建一个BlockingQueue对象
ArrayBlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(2);

// 第一组
System.out.println(blockingQueue.add("a"));
System.out.println(blockingQueue.add("cc"));
System.out.println(blockingQueue.element());  // 查看第一个元素添加是否成功
// System.out.println(blockingQueue.add("b")); // 超出容量，报错
System.out.println(blockingQueue.remove());
System.out.println(blockingQueue.remove());
System.out.println(blockingQueue.remove());// 报错
// 第二组
System.out.println(blockingQueue.offer("c"));
System.out.println(blockingQueue.offer("2"));
System.out.println(blockingQueue.offer("3")); // 报错
System.out.println(blockingQueue.poll());
System.out.println(blockingQueue.poll());
System.out.println(blockingQueue.poll());// 报错

// 第三组
blockingQueue.put("dd1");
blockingQueue.put("dd2");
blockingQueue.put("dd3");// 报错
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("cc"));
blockingQueue.offer("asdc", 3L, TimeUnit.SECONDS);// 报错

9.3 BlockingQueue 实现类

9.3.1 ArrayBlockingQueue(常用)

• 由数组结构组成的有界阻塞队列

9.3.2 LinkedBlockingQueue(常用)

• 由链表结构组成的有界（但大小默认值为 integer.MAX_VALUE）阻塞队列

ArrayBlockingQueue 和 LinkedBlockingQueue 是两个最普通也是最常用 的阻塞队列，一般情况下，在处理多线程间的生产者消费者问题，使用这两个 类足以。

9.3.3 DelayQueue

• 使用优先级队列实现的延迟无界阻塞队列

9.3.4 PriorityBlockingQueue

基于优先级的阻塞队列（优先级的判断通过构造函数传入的 Compator 对象来 决定），但需要注意的是 PriorityBlockingQueue 并不会阻塞数据生产者，而只会在没有可消费的数据时，阻塞数据的消费者。 因此使用的时候要特别注意，生产者生产数据的速度绝对不能快于消费者消费 数据的速度，否则时间一长，会最终耗尽所有的可用堆内存空间。

在实现 PriorityBlockingQueue 时，内部控制线程同步的锁采用的是公平锁

• 支持优先级排序的无界阻塞队列

9.3.5 SynchronousQueue

• 不存储元素的阻塞队列，也即单个元素的队列

9.3.6 LinkedTransferQueue

• 由链表组成的无界阻塞队列

9.3.7 LinkedBlockingDeque

• 由链表组成的双向阻塞队列

9.4 总结

在多线程领域：所谓阻塞，在某些情况下会挂起线程（即阻塞），一旦条件满足，被挂起的线程又会自动被唤起

10、线程池

10.1 线程池简介

线程池（英语：thread pool）：一种线程使用模式。

线程池是一个维护线程的池子，可以在需要时动态地创建和回收线程，从而避免了线程创建和销毁的开销，提高了线程的复用性和并发性能。

线程池通常包括以下几个组件：

1. 任务队列：用于存放等待执行的任务。
2. 线程池管理器：用于管理线程池，包括线程的创建、销毁、监控等。
3. 线程工厂：用于创建线程。
4. 线程池执行器：用于提交任务到线程池中执行。

线程池的优点包括：

1. 降低线程创建和销毁的开销，提高了线程的复用性和并发性能。
2. 可以控制并发线程的数量，避免因过多线程导致系统负载过高或资源耗尽的问题。
3. 可以提供线程的可管理性，包括线程的监控、统计和异常处理等。

线程池的缺点主要在于对于任务处理时间较短的场景中，线程池的开销可能会比线程创建和销毁的开销更大。此外，线程池的并发度需要根据实际情况进行调整，如果并发度过高会导致系统负载过高，反之则会影响系统的性能。

在使用线程池时，需要根据实际情况进行配置，包括线程池的大小、任务队列的大小、拒绝策略等。同时，还需要注意线程池的并发度和资源使用情况，避免因过度使用线程池导致系统崩溃或性能下降。

10.2 线程池的使用方式

线程池可以通过调整线程池的大小、任务队列的大小、拒绝策略等参数来优化线程池的性能。同时，在使用线程池的过程中需要注意线程池的并发度和资源使用情况，避免因过度使用线程池导致系统崩溃或性能下降。

10.2.1 一池N线程 newFixedThreadPool(常用)

1. 创建线程池对象：通过线程池工厂类的静态方法创建一个线程池对象，例如通过Executors类的静态方法创建。

   ExecutorService executor = Executors.newFixedThreadPool(10);
   

2. 提交任务到线程池：通过线程池的execute()方法或submit()方法提交任务到线程池中执行。

   executor.execute(new Runnable() {
       @Override
       public void run() {
           // 执行具体的任务
       }
   });
   
   Future<String> future = executor.submit(new Callable<String>() {
       @Override
       public String call() throws Exception {
           // 执行具体的任务，返回结果
           return "result";
       }
   });
   

3. 关闭线程池：在任务执行完毕后，通过线程池的shutdown()方法关闭线程池，防止新的任务提交到线程池中。

   executor.shutdown();
   

特点

1. 线程池中的线程处于一定的量，可以很好的控制线程的并发量
2. 线程可以重复被使用，在显示关闭之前，都将一直存在
3. 超出一定量的线程被提交时候需在队列中等待

场景: 适用于可以预测线程数量的业务中，或者服务器负载较重，对线程数有严 格限制的场景

10.2.2 newCachedThreadPool(常用)

• 创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空 闲线程，若无可回收，则新建线程
• 可扩容线程池大小。

特点

• 线程池中数量没有固定，可达到最大值（Interger. MAX_VALUE）
• 线程池中的线程可进行缓存重复利用和回收（回收默认时间为 1 分钟）
• 当线程池中，没有可用线程，会重新创建一个线程

场景: 适用于创建一个可无限扩大的线程池，服务器负载压力较轻，执行时间较短，任务多的场景。

10.2.3 newSingleThreadExecutor(常用)

• 创建一个使用单个 worker 线程的 Executor，以无界队列方式来运行该线程。
• 一池一线程

特点

• 线程池中最多执行 1 个线程，之后提交的线程活动将会排在队列中以此执行

场景: 适用于需要保证顺序执行各个任务，并且在任意时间点，不会同时有多个 线程的场景

三个线程的使用方式：

/**
 * @author Shier 2023/2/22 13:11
 */
public class ThreadPollDemo1 {
    public static void main(String[] args) {
        // 一池N线程
        // 创建线程池
        ExecutorService executorService1 = Executors.newFixedThreadPool(3);
        // 一池一线程
        ExecutorService executorService2 = Executors.newSingleThreadExecutor();
        // 一池可扩容线程
        ExecutorService executorService3 = Executors.newCachedThreadPool();
        
        try {
            for (int i = 1; i <= 5; i++) {
                // 执行线程
                executorService1.execute(() -> {
                    System.out.println(Thread.currentThread().getName() + "号线程正在执行...");
                });
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 关闭线程池
            executorService1.shutdown();
        }

    }
}

10.3 线程池的七个参数

1. corePoolSize：线程池的核心线程数
2. maximumPoolSize：能容纳的最大线程数
3. keepAliveTime：空闲线程存活时间
4. unit：存活的时间单位
5. workQueue：存放提交但未执行任务的队列
6. threadFactory：创建线程的工厂类
7. handler：等待队列满后的拒绝策略

线程池中，有三个重要的参数，决定影响了拒绝策略：

1. corePoolSize - 核心线程数，也即最小的线程数
2. workQueue - 阻塞队列
3. maximumPoolSize - 最大线程数 当提交任务数大于 corePoolSize 的时候，会优先将任务放到 workQueue 阻塞队列中。

当阻塞队列饱和后，会扩充线程池中线程数，直到达到maximumPoolSize 最大线程数配置。此时，再多余的任务，则会触发线程池的拒绝策略了。 总结起来，也就是一句话，当提交的任务数大于（workQueue.size() + maximumPoolSize ），就会触发线程池的拒绝策略。

10.3.1 拒绝策略(重点)

1. CallerRunsPolicy：当触发拒绝策略，只要线程池没有关闭的话，则使用调用 线程直接运行任务。一般并发比较小，性能要求不高，不允许失败。但是，由 于调用者自己运行任务，如果任务提交速度过快，可能导致程序阻塞，性能效 率上必然的损失较大 。
2. AbortPolicy：丢弃任务，并抛出拒绝执行 RejectedExecutionException 异常 信息。线程池默认的拒绝策略。必须处理好抛出的异常，否则会打断当前的执 行流程，影响后续的任务执行。
3. DiscardPolicy： 直接丢弃，其他啥都没有
4. DiscardOldestPolicy: 当触发拒绝策略，只要线程池没有关闭的话，丢弃阻塞队列 workQueue 中最老的一个任务，并将新任务加入

1. 在创建了线程池后，线程池中的线程数为零
2. 当调用 execute()方法添加一个请求任务时，线程池会做出如下判断：
   1. 如 果正在运行的线程数量小于 corePoolSize，那么马上创建线程运行这个任务；
   2. 如果正在运行的线程数量大于或等于 corePoolSize，那么将这个任务放入 队列；
   3. 如果这个时候队列满了且正在运行的线程数量还小于 maximumPoolSize，那么还是要创建非核心线程立刻运行这个任务；
   4. 如果队列满了且正在运行的线程数量大于或等于 maximumPoolSize，那么线程 池会启动饱和拒绝策略来执行。
3. 当一个线程完成任务时，它会从队列中取下一个任务来执行
4. 当一个线程无事可做超过一定的时间（keepAliveTime）时，线程会判断：
   1. 如果当前运行的线程数大于 corePoolSize，那么这个线程就被停掉。
   2. 所以线程池的所有任务完成后，它最终会收缩到 corePoolSize 的大小。

自定义线程示例：

public class ThreadPollDemo2 {
    public static void main(String[] args) {
        // 自定义线程池
        ThreadPoolExecutor executor = new ThreadPoolExecutor(2,
                5,
                2L,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(2),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.AbortPolicy());
        try {
            for (int i = 1; i <= 5; i++) {
                // 执行线程
                executor.execute(() -> {
                    System.out.println(Thread.currentThread().getName() + "号线程正在执行...");
                });
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 关闭线程池
            executor.shutdown();
        }
    }
}

11、Fork/Join 框架

Fork/Join 它可以将一个大的任务拆分成多个子任务进行并行处理，最后将子 任务结果合并成最后的计算结果，并进行输出。Fork/Join 框架要完成两件事情：

• Fork：把一个复杂任务进行分拆，大事化小
• Join：把分拆任务的结果进行合并

11.1 任务处理

1. 任务分割：首先 Fork/Join 框架需要把大的任务分割成足够小的子任务，如果子任务比较大的话还要对子任务进行继续分割
2. 执行任务并合并结果：分割的子任务分别放到双端队列里，然后几个启动线程 分别从双端队列里获取任务执行。
3. 子任务执行完的结果都放在另外一个队列里， 启动一个线程从队列里取数据，然后合并这些数据。

11.2 Fork 方法

原理：

当我们调用 ForkJoinTask 的 fork 方法时，程序会把 任务放在 ForkJoinWorkerThread 的 pushTask 的 workQueue 中，异步地 执行这个任务，然后立即返回结果

11.3 join 方法

Join 方法的主要作用是阻塞当前线程并等待获取结果。

已完成（NORMAL）、被取消（CANCELLED）、信号（SIGNAL）和出 现异常（EXCEPTIONAL）

• 如果任务状态是已完成，则直接返回任务结果
• 如果任务状态是被取消，则直接抛出 CancellationException
• 如果任务状态是抛出异常，则直接抛出对应的异常

class MyTask extends RecursiveTask<Integer> {

    // 拆分的插值不能超过10
    private static final Integer VALUE = 10;
    private int left;
    private int right;
    private int result;

    public MyTask(int left, int right) {
        this.left = left;
        this.right = right;
    }

    @Override
    protected Integer compute() {
        // 判断差值
        if ((right - left) <= 10) {
            for (int i = left; i <= right; i++) {
                result += i;
            }
        } else {
            // 进一步拆分
            int middle = (left + right) / 2;
            // 左拆分
            MyTask leftTask = new MyTask(left, middle);
            // 右拆分
            MyTask rightTask = new MyTask(middle + 1, right);

            leftTask.fork();
            rightTask.fork();
            // 合并
            result = leftTask.join() + rightTask.join();
        }
        return result;
    }
}

public class AddSum {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        MyTask myTask = new MyTask(0, 100);
        ForkJoinPool forkJoinPool = new ForkJoinPool();
        ForkJoinTask<Integer> submit = forkJoinPool.submit(myTask);
        Integer integer = submit.get();
        System.out.println(integer);
        // 关闭
        forkJoinPool.shutdown();
    }
}

12、CompletableFuture异步回调

使用CompletableFuture可以将耗时的操作异步执行，不会阻塞主线程，从而提高程序的性能和响应速度。同时，它还可以通过回调函数来处理计算结果，以及处理可能出现的异常情况。

12.1 有无返回值的异步调用

public class CompletableFutureDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException {

        // 没有返回值的异步调用
        CompletableFuture<Void> runAsync = CompletableFuture.runAsync(() -> {
            System.out.println(Thread.currentThread().getName() + "*****没有返回值的异步调用测试");
        });
        runAsync.get();

        // 有返回值的异步调用
        CompletableFuture<Integer> supplyAsync = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName() + "======有返回值的异步调用测试");
            // int i = 1 / 0;
            return 200;
        });
        supplyAsync.whenComplete((t, u) -> {
            System.out.println("t:" + t); // 返回值
            System.out.println("u:" + u);// 异常值
        }).get();
    }
}

CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
    // 执行耗时的计算，返回结果
    return 1 + 2;
});

future.thenAccept(result -> {
    // 处理计算结果
    System.out.println("计算结果为：" + result);
});

future.exceptionally(ex -> {
    // 处理可能出现的异常情况
    System.out.println("计算出错：" + ex.getMessage());
    return null;
});

在这个例子中，我们使用CompletableFuture.supplyAsync方法来创建一个异步计算任务。该方法接受一个Supplier函数，用于执行耗时的计算并返回结果。然后，我们使用thenAccept方法来注册一个回调函数，该函数会在计算完成后被执行，并且会接收计算结果作为参数。最后，我们使用exceptionally方法来注册一个异常处理函数，该函数会在计算出错时被执行，并且会接收异常信息作为参数。

需要注意的是，thenAccept方法是一个消费者函数，它不会返回任何值，因此不能在回调函数中修改计算结果。如果需要在回调函数中修改计算结果，可以使用thenApply方法，该方法接受一个函数，用于将计算结果转换为另一个值，并返回转换后的结果。

rkJoinPool();
ForkJoinTask submit = forkJoinPool.submit(myTask);
Integer integer = submit.get();
System.out.println(integer);
// 关闭
forkJoinPool.shutdown();
}
}

# 12、CompletableFuture异步回调

使用CompletableFuture可以将耗时的操作异步执行，不会阻塞主线程，从而提高程序的性能和响应速度。同时，它还可以通过回调函数来处理计算结果，以及处理可能出现的异常情况。



## 12.1 有无返回值的异步调用

```java
public class CompletableFutureDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException {

        // 没有返回值的异步调用
        CompletableFuture<Void> runAsync = CompletableFuture.runAsync(() -> {
            System.out.println(Thread.currentThread().getName() + "*****没有返回值的异步调用测试");
        });
        runAsync.get();

        // 有返回值的异步调用
        CompletableFuture<Integer> supplyAsync = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName() + "======有返回值的异步调用测试");
            // int i = 1 / 0;
            return 200;
        });
        supplyAsync.whenComplete((t, u) -> {
            System.out.println("t:" + t); // 返回值
            System.out.println("u:" + u);// 异常值
        }).get();
    }
}

CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
    // 执行耗时的计算，返回结果
    return 1 + 2;
});

future.thenAccept(result -> {
    // 处理计算结果
    System.out.println("计算结果为：" + result);
});

future.exceptionally(ex -> {
    // 处理可能出现的异常情况
    System.out.println("计算出错：" + ex.getMessage());
    return null;
});

在这个例子中，我们使用CompletableFuture.supplyAsync方法来创建一个异步计算任务。该方法接受一个Supplier函数，用于执行耗时的计算并返回结果。然后，我们使用thenAccept方法来注册一个回调函数，该函数会在计算完成后被执行，并且会接收计算结果作为参数。最后，我们使用exceptionally方法来注册一个异常处理函数，该函数会在计算出错时被执行，并且会接收异常信息作为参数。

需要注意的是，thenAccept方法是一个消费者函数，它不会返回任何值，因此不能在回调函数中修改计算结果。如果需要在回调函数中修改计算结果，可以使用thenApply方法，该方法接受一个函数，用于将计算结果转换为另一个值，并返回转换后的结果。