Java 多线程（6）----线程池（上）

ゝ一纸荒年。 2022-05-24 22:06 150阅读 0赞

本文标题大纲：

*   *  前言
     *  阻塞队列
     *  自定义阻塞队列
     *  Java 提供的阻塞队列
     *  Future 接口

## 前言 ##

在前面的系列文章中，我们介绍了一下 Java 中多线程的一些主要的知识点和多线程并发程序的设计和处理思想。包括线程的介绍、生命周期、线程的运行控制。之后介绍了如何确保 Java 多线程并发程序的正确性，即通过锁（`ReentrantLock` 、`synchronized` ）的思想来实现多线程执行顺序的控制等。如果你对这些还不熟悉，建议看一下前面的文章。接下来我们来看一下 Java 多线程中另一个重要的知识：线程池，在此之前，我们需要了解一下 Java 中的阻塞队列：

## 阻塞队列 ##

何为阻塞队列呢？首先，这个名字中有一个 `队列` ，证明应该是一个提供了队列相关功能（存取物品）的东西，那么何为 `阻塞` 呢？这个词我们在前面的文章中已经见过多次了，明显是针对线程而言的。那么我们大概可以猜出这个阻塞队列大概是干什么的了：可以使得线程陷入阻塞状态的储存队列。  
这么说其实意思是到了，但是具体在哪里会用到这个阻塞队列呢？不知道你还记不记得《操作系统》课程中讲过一个非常经典的问题：生产者和消费者问题。我们来一起看一下：

假设现在我们有一个固定容量的产品队列，里面放的是生产者产生的产品。  
我们将生产者看做是一个线程，这个线程专门向这个产品队列中提供已经成熟的产品；  
我们将消费者也看作是一个线程，这个线程专门从这个产品队列中取出生产者线程提供的产品。两者的操作过程可以用一张图来表示：  
![这里写图片描述][70]  
这个其实就是我们说的生产者、消费者模型，但是有一些问题需要解决：

`1、产品队列的容量是有限的，也就是说产品队列中只能储存有限个产品，那么当生产者线程把产品送入产品队列时，应该检测产品队列是否已经饱和，如果饱和，那么证明生产者线程此时应该被阻塞（即不能继续输入产品到产品队列中），等到消费者线程从产品队列中取出产品之后，生产者线程才应该被唤醒，继续输送产品。`

`2、同样的问题也会在消费者线程中发生：当产品队列中的产品数目为 0 时，即产品队列中没有产品了，那么此时消费者线程不能取产品，也应该被阻塞，即每次消费者取产品的时候应该检测队列是否为空，如果为空的话，消费者线程被阻塞，直到生产者向产品队列中输送产品之后，消费者线程才应该被唤醒，继续从产品队列中取出产品。`

`3、最后，为了保证产品队列中数据的正确性，在生产者线程和消费者线程在进入产品队列输送 / 取出产品之前，线程应该获取产品队列中的锁资源，没有获取产品队列的锁资源的线程不能进入产品队列中执行操作，即同一个时刻生产者线程和消费者线程不能同时进入产品队列中执行操作。`

## 自定义阻塞队列 ##

我们可以把能够解决上面提出的 3 个问题的队列称之为阻塞队列，我们可以利用前面的知识去构造一个简单的自定义阻塞队列来完成这个功能了：

/**
     * 自定义阻塞队列实现生产者、消费者模式：
     */
    public static class CustomBlockingQueueTest {
        // 自定义的阻塞队列
        public static class CustomBlockingQueue<T> {
            private Object[] elements = null;
            // 当前元素个数
            private int elementsCount = 0;
            // 阻塞队列的锁资源
            private ReentrantLock lock = new ReentrantLock();
            // 队列已满的线程阻塞控制器
            Condition fullLock = null;
            // 队列为空的线程阻塞控制
            Condition emptyLock = null;
    
            public CustomBlockingQueue(int elementsCount) {
                elements = new Object[elementsCount];
                fullLock = lock.newCondition();
                emptyLock = lock.newCondition();
            }
    
            /**
             * 向队列末尾插入新元素的方法，如果队列元素已满，那么阻塞当前线程
             */
            public void put(T ele) throws InterruptedException {
                if (ele == null) {
                    throw new IllegalArgumentException("插入元素不能为空！");
                } else {
                    // 构造同步块
                    lock.lock();
                    try {
                        // 如果当前队列已满，那么阻塞当前生产者线程
                        while (elementsCount == elements.length) {
                            System.out.println("队列已满，元素插入失败，线程被阻塞！");
                            fullLock.await();
                        }
                        // 将元素插入队列末尾
                        elements[elementsCount++] = ele;
                        // 队列中已经有元素，唤醒所有阻塞的消费者线程
                        emptyLock.signalAll();
                        System.out.println("插入元素，当前队列元素数量：" + elementsCount);
                    } finally {
                        lock.unlock();
                    }
                }
            }
    
            /**
             * 从阻塞队列中取出元素的方法，如果队列中没有元素，那么阻塞当前线程
             */
            public T take() throws InterruptedException {
                // 构造同步块
                lock.lock();
                try {
                    // 如果当前队列已空，那么阻塞当前消费者线程
                    while (elementsCount == 0) {
                        System.out.println("队列已空，元素取出失败，线程被阻塞！");
                        emptyLock.await();
                    }
                    // 队列中已经有剩余空间，唤醒所有阻塞的生产者线程
                    fullLock.signalAll();
                    T ele = (T) elements[--elementsCount];
                    System.out.println("取出元素，当前队列元素数量：" + elementsCount);
                    // 将队列中元素的引用置为 null，有利于 GC 回收这个对象
                    elements[elementsCount] = null;
                    return ele;
                } finally {
                    lock.unlock();
                }
            }
        }
    
        // 模拟产品的类
        public static class Product {
            private String productName;
    
            public Product(String name) {
                this.productName = name;
            }
        }
        // 记录产品数量
        private static int productCount = 0;
        // 创建一个容量为 5 的自定义阻塞队列
        private static CustomBlockingQueue<Product> queue = 
                new CustomBlockingQueue<Product>(5);
    
        // 创建生产者线程
        private static Thread productThread = new Thread() {
            @Override
            public void run() {
                Product pro = null;
                while (true) {
                    try {
                        pro = new Product("产品" + (++productCount));
                        queue.put(pro);
                        System.out.println(pro.productName + " 存入成功！");
                    } catch (InterruptedException e){
                        e.printStackTrace();
                    }
                }
            }
        };
        // 创建消费者线程
        private static Thread customThread = new Thread() {
            @Override
            public void run() {
                Product pro = null;
                while (true) {
                    try {
                        pro = queue.take();
                        if (pro != null) {
                            System.out.println(pro.productName + " 取出成功！");
                        }
                    } catch (InterruptedException e){
                        e.printStackTrace();
                    }
                }
            };
        };
    
        public static void startTest() {
            productThread.start();
            customThread.start();
        }
    }

我们来看一下运行结果：  
![这里写图片描述][70 1]  
我们利用自己定义的阻塞队列完成了生产者、消费者模式。

## Java 提供的阻塞队列 ##

其实对于阻塞队列，Java 已经给我们提供了一些常用阻塞队列供我们直接使用：

ArrayBlockingQueue：数组构成的有界阻塞队列，即储存元素的数据结构是数组
    
    LinkedBlockingQueue：链表构成的有界阻塞队列，即储存元素的数据结构是链表
    
    PriorityBlockingQueue：支持按某个优先级对元素进行排序的无界阻塞队列
    
    DelayQueue：使用优先队列实现的无界阻塞队列
    
    SynchronousQueue：不储存元素的阻塞队列
    
    LinkedTransferQueue：由链表组成的无界阻塞队列
    
    LinkedBlockingDeque：由链表构成的双向阻塞队列

什么事有界阻塞队列呢？有界其实就代表队列中元素的个数有限度，即不能无限储存元素，同理，无界即为没有对队列的元素个数加以限制，生产者线程不会因为元素个数的原因而被阻塞。但其实使用无界的阻塞队列是非常不安全的：试想一下，假设我们有多个生产者线程，而仅有一个消费者线程，或者说生产者线程的生产速度远大于消费者线程的消费速度，那么如果不对队列最大元素数加以限制，很可能生产者线程会把计算机内存资源耗光。所以我们在使用阻塞队列时候最好对阻塞队列的最大元素个数加以限制，以保证计算机资源的充裕。

**ArrayBlockingQueue**  
使用数组实现的有界阻塞队列，按照先进先出的顺序对队列元素排序。创建时必须指定队列的元素最大值进行指定，默认情况下其不保证线程公平的访问队列，`公平即为先被阻塞的线程在被唤醒后可以先得到锁资源继续上次未完成的操作`。当然，我们可以通过参数来控制这个属性：

// 创建一个容量为 1000 的公平阻塞队列
    ArrayBlockingQueue queue = new ArrayBlockingQueue(1000, true);

**LinkedBlockingQueue**  
基于链表的阻塞队列，同样按照先进先出的顺序对队列元素排序。其有一个特点是对于生产者和消费者，`分别用两个锁资源来控制生产者线程和消费者线程，即允许一个生产者线程和一个消费者线程在同一时刻同时对队列数据进行操作，这样的话在某个方面上提高了处理效率`

**PriorityBlockingQueue**  
支持自定义元素优先级的队列。默认情况下元素采用升序排序，通过自定义 `compareTo` 方法来实现自定义元素排序。或者在创建队列时传入构造参数 `Compare` 来对元素排序。

**DelayQueue**  
支持延时获取元素的无界阻塞队列，储存的元素必须实现 `Delayed` 接口，对于每个元素，只有元素到达一定时间之后才可以被取走。

**SynchronousQueue**  
不储存元素的阻塞队列，每个生产者线程必须等待另一个消费者线程的执行，同理，每个消费者个线程必须等待另一个生产者线程的执行，因此队列中没有元素。

**LinkedBlockingDeque**  
由链表组成的双向阻塞队列，提供相关方法，可以从队列的两端插入和取出队列元素。

在这些阻塞队列中，我们最常用的莫过于 `ArrayBlockingQueue` 和 `LinkedBlockingQueue` 了，因此下面我们来简单看一下 `ArrayBlockingQueue` 的一些源代码，对于其余阻塞队列，小伙伴们可以自己分析：  
![这里写图片描述][70 2]  
和我们自定义的阻塞队列很像，有储存元素的 `Object` 数组 `items`， 取元素的下标`takeIndex` 和储存元素的下标 `putIndex` ，下面 `count` 为当前阻塞队列中元素的个数。  
后面 3 个为重入锁资源对象 `lock` 、控制消费者线程阻塞和唤醒的对象 `notEmpty` 和控制生产者线程的阻塞和唤醒对象 `notFull` 。我们继续往下看：  
![这里写图片描述][70 3]  
这里是插入元素到阻塞队列尾部的方法，同样的，先检测插入的元素是否为空，然后通过 `lock.lockInterruptibly()` 方法来获取锁资源，之后通过 `count == items.length` 来判断队列是否已满，如果已满则阻塞当前线程，否则执行 `enqueue(e)` 方法来插入元素，我们看看这个方法做了什么：  
![这里写图片描述][70 4]  
其实就是将元素插入到上面定义的 `Object` 对象数组 `items` 中，之后更新元素插入下标和队列中元素数量，操作完成后队列中一定是有元素的，所有最后唤醒所有阻塞的消费者线程来取出队列中的元素。

我们再来看看取出队列元素的方法：  
![这里写图片描述][70 5]  
和插入元素的逻辑几乎一样，还是看看 `dequeue` 方法：  
![这里写图片描述][70 6]  
同样的先取出当前队列首部元素，并且更新取出元素的下标 `takeIndex` 和 `count` ，操作完成后，队列中一定会有多余的空位，所以唤醒所有的生产者线程来向队列中插入元素。

好了，`ArrayBlockingQueue` 的基本操作流程我们分析完了，最后，使用 `ArrayBlockingQUeue` 类来实现上面的生产者、消费者模型：

/**
     * ArrayBlockingQueue 类的使用，使用 ArrayBlockingQueue 实现生产者、消费者问题：
     */
    public static class ArrayBlockingQueueTest {
        
        static int productCount = 0;
        // 模拟产品的类
        public static class Product {
        
            private String productName;
    
            public Product(String name) {
                this.productName = name;
            }
        }
        // 创建一个容量为 5 的公平阻塞队列，
        // 公平即为先被阻塞的线程在被唤醒后可以先得到锁资源继续上次未完成的操作
        private static ArrayBlockingQueue<Product> queue = new ArrayBlockingQueue<Product>(5, true);
        // 创建生产者线程
        private static Thread productThread = new Thread() {
            @Override
            public void run() {
                Product pro = null;
                while (true) {
                    try {
                        pro = new Product("产品" + (++productCount));
                        // 插入元素到队尾
                        queue.put(pro);
                        System.out.println(pro.productName + " 存入成功！");
                    } catch (InterruptedException e){
                        e.printStackTrace();
                    }
                }
            }
        };
        // 创建消费者线程
        private static Thread customThread = new Thread() {
            @Override
            public void run() {
                Product pro = null;
                while (true) {
                    try {
                        // 取出队头元素
                        pro = queue.take();
                        if (pro != null) {
                            System.out.println(pro.productName + " 取出成功！");
                        }
                    } catch (InterruptedException e){
                        e.printStackTrace();
                    }
                }
            };
        };
    
        public static void startTest() {
            productThread.start();
            customThread.start();
        }
    }
    
    public static void main(String[] args) {
        CustomBlockingQueueTest.startTest();
    }

可以看到对于 Java 提供的阻塞队列，我们直接使用就行了，代码量减少了不少，并且准确性上也有了更高的保证。来看看结果：  
![这里写图片描述][70 7]  
因为是使用 Java 提供的阻塞队列，因此我们无法直接的观察到队列元素的具体变化情况，但结合上面的分析，理解了阻塞队列的原理，这些也就不难了。

## Future 接口 ##

接下来介绍一个在线程池中会常用到的接口 `Future`，当我们使用线程池对象的 `submit` 方法向线程池提交任务时，该方法会返回一个 `Future` 类型的对象。那么这个 `Future` 类型的对象有什么作用呢？简单点来说，**`Future` 接口提供了一些方法来获取向线程池提交的任务的执行状态和结果的信息**，再具体一点，我们来看看源码中其定义的方法的解释：

public interface Future<V> {
    
        /**
         * 尝试取消执行提交的对应任务，如果对应任务已经执行完成或者已经被取消或者是其他原因不能被取消，
         * 那该操作将会失败，否则的话提交的对应任务将不会被线程池中的线程执行。
         * 如果尝试取消的任务正在执行，那么通过参数来确定是否应该中断正在执行该任务的线程，
         * 如果为 false，那么将不会中断正在执行该任务的线程。
         * 在该方法返回之后，调用 isDone 方法会返回 true，
         * 调用 isCancalled 方法的返回值和该方法的返回值相同。
         */
        boolean cancel(boolean mayInterruptIfRunning);
    
        /**
         * 如果提交的对应任务还没有完全执行完成之前就被取消了，那么方法会返回 true，否则返回 false
         */
        boolean isCancelled();
    
        /**
         * 返回任务是否完成：包括下面几种情况：
         * 1、任务正常执行完成
         * 2、执行过程发生异常
         * 3、任务被取消
         * 上面几种情况发生，方法均会返回 true，
         * 如果任务正在执行，或者还未执行，那么方法返回 false
         */
        boolean isDone();
    
        /**
         * 阻塞调用该方法的线程，直到提交的对应任务执行完成之后，
         * 方法会返回一个泛型结果对象表示任务执行的结果，
         * 如果提交的任务是一个 Callable 类型的对象，那么返回 Callable 对象的 call 方法的返回值，
         * 如果提交的任务是一个 Runnable 类型的对象，那么返回 Runnable 对象的 run 方法的返回值，即为 null。
         * 如果在调用线程阻塞的过程中发生了中断，那么方法抛出 InterruptedException 异常
         */
        V get() throws InterruptedException, ExecutionException;
    
        /**
         * 功能同上面的重载方法，但是添加了一个条件, 即阻塞时间，
         * 该方法使得调用线程的阻塞时间不会超过参数指定的时间，
         * 如果在规定时间内对应任务没有运行完成，方法抛出一个 TimeoutException 异常
         */
        V get(long timeout, TimeUnit unit)
            throws InterruptedException, ExecutionException, TimeoutException;
    }

上面代码中给出了 `Future` 接口中方法的解释，并涉及到了一个新的接口：`Callable`，我们来看看这个接口的定义：

public interface Callable<V> {
        /**
         * Computes a result, or throws an exception if unable to do so.
         *
         * @return computed result
         * @throws Exception if unable to compute a result
         */
        V call() throws Exception;
    }

类似于 `Runnable` 接口，这个接口提供了一个 `call` 方法，并且这个方法提供了一个返回值作为任务的执行结果。  
那么接下来我们还是通过一个小例子来看一下其用法：

/**
     * Future 接口测试
     */
    public static class FutureTaskTest {
        
        public static void startTest() {
            // 创建一个新的线程池
            ExecutorService executor = Executors.newCachedThreadPool();
            // 向线程池中提交一个新的 Callable 类型的任务，提交之后会对任务对象进行包装
            Future<String> task = executor.submit(new Callable<String>() {
                // 复写 call() 方法，线程池中线程最终会调用 call() 方法
                @Override
                public String call() throws Exception {
                    // 执行线程休眠 5 秒钟
                    Thread.sleep(5000);
                    return "hello world";
                }
            });
            long startTime = System.currentTimeMillis();
            try {
                // 调用 task.get() 获取执行结果，
                // 这里即为获取上述代码中 submit 方法提交的 Callable 中 call() 方法的返回值，
                // 该方法会阻塞当前调用线程，直到任务执行完成后返回
                System.out.println(task.get());
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
            long endTime = System.currentTimeMillis();
            System.out.println("方法获取任务结果所用时间：" + (endTime - startTime) + "ms");
        }
    }
    
    public static void main(String[] args) {
        FutureTaskTest.startTest();
    }

我们创建了一个线程池，然后向其中提交了一个 `Callable` 对象，并且获得了一个返回的 `Future` 对象，之后在主线程中获取提交的任务的执行结果，即为提交的任务的 `call` 方法的执行结果，我们来看看结果：  
![这里写图片描述][70 8]  
如果你运行这个程序，你会发现一开始程序没有反应，直到第 5 秒多的的时候会在控制台中输出 `hello world` 。这个结果对照代码中的注释其实很好理解，正是符合 `Future` 对象中 `get` 方法的特性：阻塞调用线程直到提交的对应任务执行完毕。

另外，想补充的是，我们在 Java 多线程系列的第一篇文章中讲述了如果创建一个线程，当时我们采用了两种方法：  
1、通过自定义类继承 `Thread` 类并且重写其 `run` 方法  
2、通过 `new Thread(Runnable runnable)` 方法传入一个新的 `Runnable` 对象。  
其实还可以通过这里说的 `Callable` 对象来实现线程，但是 `Thread` 的构造方法中并没有提供 `Thread(Callable callabe)` 类型的方法，怎么办呢？我们需要借助另一个类 `FutureTask` 类，这个类实现了 `RunnableFuture` 接口，而这个 `RunnableFuture` 接口继承了 `Runnable` 接口和 `Future` 接口，所以这个 `FutureTask` 类的对象是可以作为参数作为 `Thread(Runnable runnable)` 构造方法的参数的。而 `FutureTask` 的构造方法又提供了通过传入 `Callable` 对象作为参数的形式：`FutureTask(Callable callable)` 。而其 `run` 方法会调用传入的 `Callable` 对象的 `call` 方法，其本质上还是通过第二种方法来新建线程。  
因此我们可以通过 `FutureTask` 对象和 `Callable` 接口来新建线程，并且我们创建的 `FutureTask` 对象还可以用来获取 `Callable` 对象中 `call` 方法的返回值作为执行结果。  
  
  
好了， 这篇文章我们对阻塞队列和 `Future` 等接口的介绍到就这里了，因为线程池的使用依赖于这些类和接口，因此在学习线程池之前了解一下这些知识是很有必要的，算是铺垫吧，下一篇文章将是对线程池源码形式的解析。  
如果博客中有什么不正确的地方，还请多多指点。如果这篇文章对您有帮助，请不要吝啬您的赞，欢迎继续关注本专栏。

谢谢观看。。。

[70]: /images/20220525/3218e68c542d49cca6d2034698622def.png
[70 1]: /images/20220525/bd3d22ec31f14eabbf1c007cbc0ced1e.png
[70 2]: /images/20220525/7c71a2e17c4f42ce81cd65eacc5e6ae3.png
[70 3]: /images/20220525/9643d79ff70142648a74299712450f80.png
[70 4]: /images/20220525/77ac24a9a1d84eec9eff6a6b83d2e636.png
[70 5]: /images/20220525/9184cb2ea44949e298d43c9de4e13a8b.png
[70 6]: /images/20220525/95431e40940f4716bbb7bbad78f416d3.png
[70 7]: /images/20220525/6fb3704befa146eeab1bdde36844dfee.png
[70 8]: /images/20220525/f38382289ff74b519e8942bf063d3037.png