手写线程池

Posted 2020-11-27 dengw125792

Executors.newSingleThreadExecutor()：

        只有一个线程的线程池，因此所有提交的任务是顺序执行

Executors.newCachedThreadPool()：

        线程池里有很多线程需要同时执行，老的可用线程将被新的任务触发重新执行，

        如果线程超过60秒内没执行，那么将被终止并从池中删除

Executors.newFixedThreadPool()：

         拥有固定线程数的线程池，如果没有任务执行，那么线程会一直等待

Executors.newScheduledThreadPool()：

         用来调度即将执行的任务的线程池

Executors.newWorkStealingPool()：

          newWorkStealingPool适合使用在很耗时的操作，

         但是newWorkStealingPool不是ThreadPoolExecutor的扩展，

         它是新的线程池类ForkJoinPool的扩展，但是都是在统一的一个Executors类中实现，

          由于能够合理的使用CPU进行对任务操作（并行操作），所以适合使用在很耗时的任务中

 

生产环境不使用以上线程池，主要原因是，以上线程池底层使用的LinkedBlockingQueue链表阻塞

队列，这样最大值为21亿，范围过大，会造成OOM异常。

需要使用ThreadPoolExecutor传递7个参数手工实现线程池。

 

 

 

 

ThreadPoolExecutor

ThreadPoolExecutor作为java.util.concurrent包对外提供基础实现，以内部线程池的形式对外提供管理任务执行，

线程调度，线程池管理等等服务。

corePoolSize	核心线程池大小
maximumPoolSize	最大线程池大小
keepAliveTime	线程池中超过 corePoolSize 数目的空闲线程最大存活时间；可以allowCoreThreadTimeOut(true) 使得核心线程有效时间
TimeUnit	keepAliveTime 时间单位
workQueue	阻塞任务队列
threadFactory	新建线程工厂
RejectedExecutionHandler	当提交任务数超过 maxmumPoolSize+workQueue 之和时，任务会交给RejectedExecutionHandler 来处理

 

 

Java中的BlockingQueue主要有两种实现，分别是ArrayBlockingQueue 和 LinkedBlockingQueue。

ArrayBlockingQueue是一个用数组实现的有界阻塞队列，必须设置容量。

LinkedBlockingQueue是一个用链表实现的有界阻塞队列，容量可以选择进行设置，

不设置的话，将是一个无边界的阻塞队列，最大长度为Integer.MAX_VALUE。

这里的问题就出在：不设置的话，将是一个无边界的阻塞队列，最大长度为Integer.MAX_VALUE。

也就是说，如果我们不设置LinkedBlockingQueue的容量的话，其默认容量将会是Integer.MAX_VALUE。

 

=================测试===============================================

package t1;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class TestExecutors {

public static void main(String[] args) {
ExecutorService executorService = new ThreadPoolExecutor(2, // corePoolSize 当日值班线程
5, // maximumPoolSize  最大可提供线程
1, // keepAliveTime  除开值班线程等待时间，等待这么多时间后，除开值班线程的其他线程销毁
TimeUnit.SECONDS, // unit  等待时间的单位 ，等待这么多时间后，除开值班线程的其他线程销毁
new LinkedBlockingDeque<>(3), // workQueue  阻塞队列的类型以及大小
Executors.defaultThreadFactory(), // threadFactory 配置 默认即可
new ThreadPoolExecutor.AbortPolicy());// handler 线程多于maximumPoolSize+LinkedBlockingDeque后会报错

// new ThreadPoolExecutor.CallerRunsPolicy());//handler 多的交给调用他的线程处理
// new ThreadPoolExecutor.DiscardOldestPolicy());//handler 丢掉等待最久的线程
// new ThreadPoolExecutor.DiscardPolicy());//handler 丢掉处理不过来的线程

for (int i = 0; i < 9; i++) {
executorService.execute(() -> {
System.out.println(Thread.currentThread().getName());

});

}
}

}

 

输出结果根据拒绝策略而异。

 

 *********************************************************

corePoolSize 参数配置规则

CPU 密集型：比如while循环一开，CPU就很高上去了这种

       CPU 密集的意思是该任务需要大量的运算，而没有阻塞，CPU 一直全速运行。

       CPU 密集型任务尽可能的少的线程数量，一般为 CPU 核数 + 1 个线程的线程池。

IO 密集型：比如磁盘IO特别频繁这种

       由于 IO 密集型任务线程并不是一直在执行任务，可以多分配一点线程数，如 CPU * 2 。

        也可以使用公式：CPU 核数 / (1 - 阻塞系数)；其中阻塞系数在 0.8 ～ 0.9 之间。