JUC的线程池架构

🙈作者简介：练习时长两年半的Java up主
🙉个人主页：程序员老茶
🙊 ps:点赞👍是免费的，却可以让写博客的作者开心好久好久😎
📚系列专栏：Java全栈，计算机系列（火速更新中）
💭 格言：种一棵树最好的时间是十年前，其次是现在
🏡动动小手，点个关注不迷路，感谢宝子们一键三连

课程名：Java

内容/作用：知识点/设计/实验/作业/练习

学习：JUC的线程池架构

JUC的线程池架构

本文主要介绍Java中如何使用java.util.concurrent包中的线程池（ExecutorService和ThreadPoolExecutor）来实现高并发、高可用的系统。我们将从线程池的基本概念、使用方法、核心组件以及自定义线程池四个方面进行阐述。

一、线程池的基本概念

线程池是一种管理线程的机制，它可以有效地控制线程的数量，避免大量线程之间的切换导致性能下降。线程池中的线程可以被复用，当一个任务完成后，线程不会被销毁，而是被重新分配给新的任务。

二、线程池的使用方法

1. 创建线程池

在Java中，可以通过Executors工具类来创建不同类型的线程池。例如，创建一个固定大小的线程池：

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ThreadPoolDemo {
    public static void main(String[] args) {
        ExecutorService executorService = Executors.newFixedThreadPool(5);
    }
}
1
2
3
4
5
6
7
8

2. 提交任务到线程池

使用submit()方法将任务提交到线程池中执行：

executorService.submit(new Runnable() {
    @Override
    public void run() {
        System.out.println("任务执行中...");
    }
});
1
2
3
4
5
6

3. 关闭线程池

当所有任务都执行完毕后，需要关闭线程池以释放资源：

executorService.shutdown();
1

三、线程池的核心组件

1. ThreadPoolExecutor

ThreadPoolExecutor是ExecutorService接口的实现类，它提供了更多的功能，如定时执行、定期执行、异常处理等。我们通常需要自己实现一个ThreadPoolExecutor来满足业务需求。

2. BlockingQueue

BlockingQueue是一个阻塞队列，用于存放待处理的任务。它是一个FIFO（先进先出）的队列，当队列满时，新来的任务会等待；当队列为空时，正在执行的任务会等待。常用的阻塞队列有ArrayBlockingQueue、LinkedBlockingQueue和SynchronousQueue。

3. 拒绝策略

当线程池无法处理新提交的任务时，需要采取一定的策略来处理这些任务。Java中的RejectedExecutionHandler接口提供了四种默认的拒绝策略：

1. CallerRunsPolicy：直接在调用者线程中运行任务。
2. AbortPolicy：抛出一个未检查异常中断任务。
3. DiscardPolicy：丢弃任务，不做任何处理。
4. DiscardOldestPolicy：丢弃队列中最旧的任务，尝试重新提交新任务。

我们可以根据实际需求自定义拒绝策略。例如，下面是一个自定义的拒绝策略示例：

import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

public class CustomRejectedExecutionHandler implements RejectedExecutionHandler {
    private static final int MAX_CAPACITY = 100; // 最大容量
    private AtomicInteger rejectedCount = new AtomicInteger(); // 被拒绝的任务计数器
    private ArrayBlockingQueue<Runnable> taskQueue; // 任务队列

    public CustomRejectedExecutionHandler() {
        taskQueue = new ArrayBlockingQueue<>(MAX_CAPACITY); // 初始化任务队列
    }

    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) { // 拒绝策略实现方法
        if (taskQueue.remainingCapacity() == 0) { // 如果队列已满，添加拒绝策略
            executor.setRejectedExecutionHandler(new RejectedExecutionHandler() { // 设置新的拒绝策略
                @Override
                public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) { // 拒绝策略实现方法（AbortPolicy）
                    System.err.println("Task queue is full, rejecting task: " + r); // 输出错误信息并抛出异常（AbortPolicy）
                    throw new RejectedExecutionException("Task queue is full", r); // 抛出RejectedExecutionException异常（AbortPolicy）
                }
            });
        } else { // 如果队列未满，将任务添加到队列中，并增加计数器（CallerRunsPolicy）或直接抛出异常（DiscardPolicy/DiscardOldestPolicy）
            taskQueue.put(r); // 将任务添加到队列中（非阻塞）
            rejectedCount.incrementAndGet(); // 被拒绝的任务计数器加1（CallerRunsPolicy/DiscardPolicy/DiscardOldestPolicy）或直接抛出异常（DiscardPolicy/DiscardOldestPolicy）
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

四、自定义线程池示例

下面我们通过一个示例来说明如何自定义线程池：假设我们需要一个支持定时执行和周期性执行的任务线程池，我们可以这样实现：
要实现一个支持定时执行和周期性执行的任务线程池，我们可以创建一个自定义的线程池类，继承自ThreadPoolExecutor，并重写其中的方法。以下是一个简单的示例：

import java.util.concurrent.*;

public class CustomThreadPoolExecutor extends ThreadPoolExecutor {
    private final long keepAliveTime;
    private final TimeUnit unit;

    public CustomThreadPoolExecutor(int corePoolSize, long keepAliveTime, TimeUnit unit) {
        super(corePoolSize);
        this.keepAliveTime = keepAliveTime;
        this.unit = unit;
    }

    @Override
    protected void beforeExecute(Thread t, Runnable r) {
        super.beforeExecute(t, r);
        // 在任务执行前执行的操作，例如记录日志、初始化资源等
    }

    @Override
    protected void afterExecute(Runnable r, Throwable t) {
        super.afterExecute(r, t);
        // 在任务执行后执行的操作，例如清理资源、记录日志等
    }

    public void scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit) {
        super.scheduleAtFixedRate(command, initialDelay, period, unit);
    }

    public void scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit) {
        super.scheduleWithFixedDelay(command, initialDelay, delay, unit);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

在这个示例中，我们创建了一个名为CustomThreadPoolExecutor的自定义线程池类，它继承自ThreadPoolExecutor。我们为这个类添加了两个方法：scheduleAtFixedRate和scheduleWithFixedDelay，分别用于定时执行和周期性执行任务。这两个方法内部调用了父类的相应方法来实现任务调度。

使用这个自定义线程池的示例代码如下：

import java.util.concurrent.*;

public class CustomThreadPoolExample {
    public static void main(String[] args) throws InterruptedException {
        CustomThreadPoolExecutor threadPool = new CustomThreadPoolExecutor(2, 10, TimeUnit.SECONDS);

        for (int i = 0; i < 5; i++) {
            final int taskId = i;
            threadPool.execute(() -> {
                System.out.println("Task " + taskId + " is running");
                try {
                    TimeUnit.SECONDS.sleep(1);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("Task " + taskId + " is completed");
            });
        }

        // 关闭线程池（所有任务执行完成后，线程池不再接受新任务）
        threadPool.shutdown();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

在这个示例中，我们创建了一个定时执行任务的线程池，核心线程数为2，任务间隔时间为10秒。然后，我们提交了5个任务到线程池，每个任务打印一条开始和结束信息，并在运行过程中暂停1秒。由于线程池具有定时执行功能，因此这些任务会按照设定的时间间隔依次执行。最后，我们在所有任务执行完成后关闭线程池。

在Java中，我们可以通过java.util.concurrent.Executors类创建线程池。然而，有时候我们需要创建一个具有特定功能的线程池，例如定时执行任务、周期性执行任务等。这时，我们可以创建一个自定义的线程池来实现这些功能。本文将介绍如何创建一个自定义线程池，并通过一个示例来演示其使用方法。

1. 创建自定义线程池

要创建一个自定义线程池，我们需要实现java.util.concurrent.ThreadPoolExecutor接口，并重写其中的方法。以下是一个简单的自定义线程池实现：

import java.util.concurrent.*;

public class CustomThreadPoolExecutor extends ThreadPoolExecutor {
    private final int corePoolSize;
    private final BlockingQueue<Runnable> workQueue;
    private final long keepAliveTime;
    private final TimeUnit unit;

    public CustomThreadPoolExecutor(int corePoolSize, BlockingQueue<Runnable> workQueue, long keepAliveTime, TimeUnit unit) {
        this.corePoolSize = corePoolSize;
        this.workQueue = workQueue;
        this.keepAliveTime = keepAliveTime;
        this.unit = unit;
    }

    @Override
    protected void beforeExecute(Thread t, Runnable r) {
        super.beforeExecute(t, r);
        // 在任务执行前执行的操作，例如记录日志、初始化资源等
    }

    @Override
    protected void afterExecute(Runnable r, Throwable t) {
        super.afterExecute(r, t);
        // 在任务执行后执行的操作，例如清理资源、记录日志等
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

2. 使用自定义线程池

创建好自定义线程池后，我们可以像使用普通的ThreadPoolExecutor一样使用它。以下是一个使用自定义线程池的示例：

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

public class CustomThreadPoolExample {
    public static void main(String[] args) {
        // 创建一个定时执行任务的线程池
        CustomThreadPoolExecutor threadPool = new CustomThreadPoolExecutor(2, new LinkedBlockingQueue<>(2), 30, TimeUnit.SECONDS);

        // 提交任务到线程池
        for (int i = 0; i < 5; i++) {
            final int taskId = i;
            threadPool.execute(() -> {
                System.out.println("Task " + taskId + " is running");
                try {
                    TimeUnit.SECONDS.sleep(1);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("Task " + taskId + " is completed");
            });
        }

        // 关闭线程池（所有任务执行完成后，线程池不再接受新任务）
        threadPool.shutdown();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

在这个示例中，我们创建了一个定时执行任务的线程池，最大容量为2，任务队列大小为2，空闲线程的存活时间为30秒。然后，我们提交了5个任务到线程池，每个任务打印一条开始和结束信息，并在运行过程中暂停1秒。由于线程池具有定时执行功能，因此这些任务会按照设定的时间间隔依次执行。最后，我们在所有任务执行完成后关闭线程池。