【连载 05】自定义线程池（上）

【连载 05】自定义线程池（上）

1.4 自定义线程池

现在你已经对创建和使用线程池有了初步了解，包括线程池创建参数的认识，现在我们将目光放在对象参数上，看它们在实际使用中，能达到什么效果，这样可以加深我们对这些参数的理解，帮助我们在后面的使用当中更加得心应手。

1.4.1 等待队列

线程池等待队列的参数类型是BlockingQueue<Runnable>，这是一个Java接口，它的实现类比较多，在java.Executors应用中，用到了两个实现类，分别是java.LinkedBlockingQueue和java.SynchronousQueue。

1.LinkedBlockingQueue

LinkedBlockingQueue 是 Java 中的一个阻塞队列实现，它基于链表数据结构实现。它的特点是：

• FIFO（先进先出）。队列中的元素按照它们入的顺序进行处理，即先进入队列的元素将会被优先处理。
• 线程安全。LinkedBlockingQueue 是线程安全的，可以在并发访问和修改场景中，保障线程安全。
• 阻塞操作。LinkedBlockingQueue 支持阻塞操作，当队列为空时，消费线程可以阻塞在获取方法，知道队列中有新的元素可用；当队列已满，生产线程会阻塞在提交元素，直到队列有空闲接收新元素。
• 可选容量。LinkedBlockingQueue 可以选择是否设置容量限制，如果不设置容量限制，则队列容量默认java.lang.Integer#MAX_VALUE。

2.SynchronousQueue

SynchronousQueue是Java SDK中一种特殊的阻塞队列，它的最大的特点就是容量为零。SynchronousQueue容量是零，不保存任何元素。每一个提交元素的操作都要等待一个消费线程移除操作，反之也成立。

在创建java.Executors#newCachedThreadPool(java.ThreadFactory)线程池时就是用到SynchronousQueue。根据我们之前对线程池创建新线程的分析，当向等待队列提交任务时，调用了java.SynchronousQueue#offer(E)方法时返回false，所以会直接进入创建新的线程逻辑，也就是java.ThreadPoolExecutor#addWorker方法。

.LinkedBlockingDeque

LinkedBlockingDeque 是Java SDK提供的一个双端队列实现，它与LinkedBlockingQueue一样基于链表数据结构实现，不同的是LinkedBlockingDeque因为是双端链表，所以不仅添加和删除操作不受限于固定的位置，可以java.LinkedBlockingDeque#offerFirst(E)，也可以java.LinkedBlockingDeque#offerLast(E)，同样删除操作也是。另外LinkedBlockingDeque也实现了java.BlockingQueue接口的所有方法，默认的操作是跟LinkedBlockingQueue一样的，在队列末尾添加，从队列开头移除。源码如下：

代码语言：javascript代码运行次数：0运行复制

    public boolean offer(E e) {

        return offerLast(e);

    }

    public E poll() {

        return pollFirst();

    }

这个队列有什么用呢？当我们使用线程池处理大量异步任务的场景中，假如我们期望其中一部分异步任务优先执行，如果要实现这样的功能，就需要给线程池配置一个双端链表LinkedBlockingDeque。演示代码如下：

代码语言：javascript代码运行次数：0运行复制

package org.funtester.performance.section4;

import java.LinkedBlockingDeque;

import java.ThreadPoolExecutor;

import java.TimeUnit;

/**

 * 双端列表在线程池应用功能示例

 */

public class DueueDemo {

    public static void main(String[] args) {

        ThreadPoolExecutor executor = new ThreadPoolExecutor(1, 1, 60, TimeUnit.SECODS, new LinkedBlockingDeque<>(10));// 创建线程池，使用双端列表

        for (int i = 0; i < 4; i++) {// 提交4个任务

            int index = i;// 任务索引,用于标识任务,由于lambda表达式中的变量必须是final或者等效的,所以这里使用局部变量

            Thread thread = new Thread(() -> {// 提交任务

                try {

                    Thread.sleep(1000);// 模拟任务执行,睡眠1秒,避免任务过快执行完毕

                } catch (InterruptedException e) {

                    throw new RuntimeException(e);

                }

                println(().getame() + "  " + () + "  " + index + "  执行任务");// 打印任务执行信息

            });

            if (i == ) {// 第4个任务插入到队列头部

                LinkedBlockingDeque<Runnable> queue = (LinkedBlockingDeque<Runnable>) executor.getQueue();// 获取线程池队列

                (thread);// 将任务插入到队列头部

            } else {

                (thread);// 提交任务

            }

            println(().getame() + "  " + () + "  " + index + "  提交任务");// 打印任务提交信息

        }

        executor.shutdown();// 关闭线程池,不再接受新任务,但会执行完队列中的任务,并不会立即关闭

    }

}

在这个演示代码中，将最后一个提交的任务插入了等待队列的头部，理论上会在第一个任务执行完成之后执行最后一个任务。至于效果如何，我们来执行代码验证，控制台打印内容如下：

代码语言：javascript代码运行次数：0运行复制

main  171000268720  0  提交任务

main  171000268720  1  提交任务

main  171000268720  2  提交任务

main  171000268720    提交任务

pool-1-thread-1  171000269721  0  执行任务

pool-1-thread-1  171000270722    执行任务

pool-1-thread-1  17100027172  1  执行任务

pool-1-thread-1  171000272724  2  执行任务

跟我们预想的一模一样，完美地实现了我们将优先级高的任务优先执行的设想。但是这种优先级队列在处理优先级时颗粒度比较粗，如果业务上有很多优先级级别的话，这个方案就显得难以为继。不过没关系，说到优先级，Java SDK还提供了一个java.util.PriorityQueue供我们使用，可很不幸它并不是java.BlockingQueue接口的实现类，但是还有一个java.PriorityBlockingQueue供我们使用。

4.PriorityBlockingQueue

PriorityBlockingQueue是Java SDK提供的一个线程安全的阻塞优先级队列。相比较LinkedBlockingQueue，它新增了两点特性：

• 优先级支持。PriorityBlockingQueue可以根据元素的优先级进行排序，保障优先的元素排在队列的头部。
• 容量。PriorityBlockingQueue容量不受初始容量限制，可以动态扩容。

要实现多优先级线程池，无法直接使用PriorityBlockingQueue，因为PriorityBlockingQueue要求元素必须实现java.lang.Comparable，或者在创建时指定一个java.util.Comparator实现类。而线程池等待队列中的对象类型都是java.lang.Runnable，要想兼顾两个方面，必须要做点改动。我选择创建一个新的抽象类，实现java.util.Comparator和java.lang.Runnable，当然只会实现java.util.Comparator的方法，这样依然可以使用线程池的提交方法java.ThreadPoolExecutor#execute和Lambda语法。抽象类代码如下：

代码语言：javascript代码运行次数：0运行复制

package org.funtester.performance.section4;

/**

 * 优先级任务抽象类

 */

public abstract class PriorityRunnable implements Comparable<PriorityRunnable>, Runnable {

    int priorityLevel;// 优先等级，值越小优先级越高，用于优先级队列排序

    public PriorityRunnable(int priorityLevel) {≈

        this.priorityLevel = priorityLevel;

    }

    /**

     * 用与比较两个对象的优先级

     *

     * @param o

     * @return

     */

    @Override

    public int compareTo(PriorityRunnable o) {

        return this.priorityLevel - o.priorityLevel;

    }

}

下面演示多优先级线程池的使用：

代码语言：javascript代码运行次数：0运行复制

package org.funtester.performance.section4;

import java.PriorityBlockingQueue;

import java.ThreadPoolExecutor;

import java.TimeUnit;

/**

 * 多优先级线程池使用示例

 */

public class PriorityTaskDemo {

    public static void main(String[] args) {

        ThreadPoolExecutor executor = new ThreadPoolExecutor(0, 2, 60L, TimeUnit.SECODS, new PriorityBlockingQueue<Runnable>());// 创建线程池,核心线程数0,最大线程数2,线程空闲时间60秒,任务队列为优先级阻塞队列

        for (int i = 0; i < 5; i++) {// 提交5个任务

            int priorityLevel = 5 - i;// 优先级递增

            (new PriorityRunnable(priorityLevel) {// 提交任务,优先级递增

                @Override

                public void run() {

                    try {

                        Thread.sleep(1000);// 休眠1秒,模拟任务执行时间

                    } catch (InterruptedException e) {

                        throw new RuntimeException(e);

                    }

                    println(().getame() + "  " + () + "  " + priorityLevel + "  执行任务");// 打印任务执行信息

                }

            });

            println(().getame() + "  " + () + "  " + priorityLevel + "  提交任务");// 打印任务提交信息

        }

        executor.shutdown();// 关闭线程池

    }

}

代码语言：javascript代码运行次数：0运行复制

控制台打印信息如下：

main  17100406180  5  提交任务

main  17100406180  4  提交任务

main  17100406180    提交任务

main  17100406180  2  提交任务

main  17100406180  1  提交任务

pool-1-thread-1  17100407181  5  执行任务

pool-1-thread-1  17100408182  1  执行任务

pool-1-thread-1  1710040918  2  执行任务

pool-1-thread-1  17100410184    执行任务

pool-1-thread-1  17100411185  4  执行任务

可以看出，除了第一个提交的优先级为5的任务以外（因为这个任务提交之后直接执行，并未参加排序），其他任务均按照优先级从高到低运行的。

至此，我们已经将Java线程池所用到还有将来各位可能用到的队列分享完了，笔者建议初学者掌握LinkedBlockingQueue和SynchronousQueue即可。

本文参与 腾讯云自媒体同步曝光计划，分享自。原始发表：2024-12-18，如有侵权请联系 cloudcommunity@tencent 删除线程池队列链表线程线程安全