Java ThreadPoolExecutor

时间:2022-09-29 明天一定. 人气:0

为什么要有线程池?

在实际使用中，服务器在创建和销毁线程上花费的时间和消耗的系统资源都相当大，所以要尽可能减少创建和销毁线程的次数。
由于没有线程创建和销毁时的消耗，可以提高系统响应速度
可以对线程进行合理的管理

线程池状态

1、RUNNING

状态说明：线程池处于RUNNING状态时，能够接收新任务以及对已添加的任务进行处理。

2、SHUTDOWN

状态说明：线程池处于SHUTDOWN状态时，不接收新任务，但能处理已添加的任务

3、STOP

状态说明：线程池处于STOP状态时，不接收新任务，不处理已添加的任务，并且会中断正在处理的任务

4、TIDYING

状态说明：当所有的任务已终止，ctl记录的任务数为0，线程池的状态会变为TIDYING状态；当线程池的状态变为TIDYING状态时，会调用钩子函数terminated(),该方法在ThreadPoolExecutor中是空的，若用户想在线程池变为TIDYING时进行相应的处理，就需要重载terminated()函数实现。
当线程池为STOP时，线程池中执行的任务为空时，就会又STOP->TIDYING

5、TERMINATED

状态说明：线程池彻底终止，就会变成TERMINATED状态

ThreadPoolExecutor核心参数

corePoolSize

corePoolSize – the number of threads to keep in the pool, even if they are idle, unless allowCoreThreadTimeOut is set

池中持有的线程数，即使它们处于空闲状态，除非设置了allowCoreThreadTimeOut

线程池中的核心线程数，当提交一个任务时，线程池创建一个新线程执行任务，直到当前线程数等于corePoolSize, 即使有其他空闲线程能够执行新来的任务, 也会继续创建线程；

如果执行了线程池的prestartAllCoreThreads()方法，线程池会提前创建并启动所有核心线程。

案例：

核心线程数和最大线程数为1，使用一个不存储元素的阻塞队列。

public static void main(String[] args) throws InterruptedException {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(1, 1, 0,
                TimeUnit.SECONDS,
                new SynchronousQueue<>(),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.AbortPolicy());
 
        for (int i = 0; i < 2; i++) {
            executor.execute(()->{
                System.out.println(Thread.currentThread().getName());
            });
            Thread.sleep(1000);
        }
    }

输出：

pool-1-thread-1
pool-1-thread-1

maximumPoolSize

maximumPoolSize – the maximum number of threads to allow in the pool

池中允许存在的最大线程数

案例：

核心线程数是1，最大线程数为3，使用一个不存储元素的阻塞队列。（注意结合workQueue参数食用~）

public static void main(String[] args) throws InterruptedException {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(1, 3, 0,
                TimeUnit.SECONDS,
                new SynchronousQueue<>(),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.AbortPolicy());
 
        for (int i = 0; i < 3; i++) {
            executor.execute(()->{
                System.out.println(Thread.currentThread().getName());
            });
        }
    }

输出：

pool-1-thread-1
pool-1-thread-3
pool-1-thread-2

keepAliveTime

keepAliveTime – when the number of threads is greater than the core, this is the maximum time that excess idle threads will wait for new tasks before terminating.

当线程数大于核心时，这是多余空闲线程在终止前等待新任务的最长时间。

线程空闲时的存活时间，即当线程没有任务执行时，该线程继续存活的时间；默认情况下，该参数只在线程数大于corePoolSize时才有用, 超过这个时间的空闲线程将被终止；

unit

unit – the time unit for the keepAliveTime argument

keepAliveTime参数的单位

workQueue

workQueue – the queue to use for holding tasks before they are executed. This queue will hold only the Runnable tasks submitted by the execute method.

用来放置没有执行的任务，此队列将仅保存execute方法提交的可运行任务。

用来保存等待被执行的任务的阻塞队列。

如果当前线程数为corePoolSize，继续提交的任务被保存到阻塞队列中，等待被执行；

如果当前阻塞队列满了，且继续提交任务，则创建新的线程执行任务，前提是当前线程数小于maximumPoolSize；当阻塞队列是无界队列, 则maximumPoolSize则不起作用, 因为无法提交至核心线程池的线程会一直持续地放入workQueue。

JDK提供以下队列：

ArrayBlockingQueue: 基于数组结构的有界阻塞队列，按FIFO排序任务；(常用)
LinkedBlockingQueue: 基于链表结构的阻塞队列，按FIFO排序任务，吞吐量通常要高于ArrayBlockingQueue；(常用)
SynchronousQueue: 一个不存储元素的阻塞队列，每个插入操作必须等到另一个线程调用移除操作，否则插入操作一直处于阻塞状态，吞吐量通常要高于LinkedBlockingQueue；(常用)
PriorityBlockingQueue: 具有优先级的无界阻塞队列；
DelayQueue：一个使用优先级队列实现的无界阻塞队列。
LinkedTransferQueue：一个由链表结构组成的无界阻塞队列。
LinkedBlockingDeque：一个由链表结构组成的双向阻塞队列。

案例：

核心线程数是1，最大线程数为3，使用一个容量为1的队列。

public static void main(String[] args) throws InterruptedException {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(1, 3, 0,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(1),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.AbortPolicy());
 
        for (int i = 0; i < 3; i++) {
            executor.execute(()->{
                System.out.println(Thread.currentThread().getName());
            });
        }
    }

输出：

pool-1-thread-2
pool-1-thread-1
pool-1-thread-2

threadFactory

threadFactory – the factory to use when the executor creates a new thread

创建执行器创建线程的工厂

通过自定义的线程工厂可以给每个新建的线程设置一个具有识别度的线程名。默认为DefaultThreadFactory。

案例：

给线程起名字。我是用spring里的类。如不想引入过多依赖，可以自己仿照Executors.defaultThreadFactory()的代码写一个类更改namePrefix即可。

public static void main(String[] args){
        CustomizableThreadFactory customizableThreadFactory = new CustomizableThreadFactory("mine-");//import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
        ThreadPoolExecutor executor = new ThreadPoolExecutor(1, 1, 0,
                TimeUnit.SECONDS,
                new SynchronousQueue<>(),
                customizableThreadFactory,
                new ThreadPoolExecutor.CallerRunsPolicy());
        for (int i = 0; i < 1; i++) {
            executor.execute(()->{
                System.out.println(Thread.currentThread().getName());
            });
        }
    }

输出：

mine-1

handler

handler – the handler to use when execution is blocked because the thread bounds and queue capacities are reached

达到线程边界和队列容量而阻止执行时使用的处理程序

线程池的饱和策略，当阻塞队列满了，且没有空闲的工作线程，如果继续提交任务，必须采取一种策略处理该任务，

线程池提供了4种策略:

AbortPolicy: 直接抛出异常，默认策略；

CallerRunsPolicy: 用调用者所在的线程来执行任务；

DiscardOldestPolicy: 丢弃阻塞队列中靠最前的任务，并执行当前任务；

DiscardPolicy: 直接丢弃任务；

案例：

以 CallerRunsPolicy为案例。核心和最大线程数为1。

public static void main(String[] args) throws InterruptedException {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(1, 1, 0,
                TimeUnit.SECONDS,
                new SynchronousQueue<>(),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.CallerRunsPolicy());
 
        for (int i = 0; i < 3; i++) {
            executor.execute(()->{
                System.out.println(Thread.currentThread().getName());
            });
        }
    }

输出：

main
main
pool-1-thread-1

关闭线程池的方式

shutdown：

修改线程池状态为SHUTDOWN
不再接收新提交的任务
中断线程池中空闲的线程
第3步只是中断了空闲的线程，但正在执行的任务以及线程池任务队列中的任务会继续执行完毕

shutdownNow：

修改线程池状态为STOP
不再接收任务提交
尝试中断线程池中所有的线程（包括正在执行的线程）
返回正在等待执行的任务列表 List<Runnable>

为什么不推荐使用Executors去创建线程池

newFixedThreadPool和newSingleThreadExecutor: 阻塞队列为无界队列，主要问题是堆积的请求处理队列可能会耗费非常大的内存，甚至OOM。newCachedThreadPool和newScheduledThreadPool: 线程数最大数是Integer.MAX_VALUE，可能会创建数量非常多的线程，甚至OOM。

加载全部内容