ThreadPoolExecutor-线程池知多少

本文主要是介绍ThreadPoolExecutor-线程池知多少，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

ThreadPoolExecutor

创建线程池的目的？

线程池为了避免频繁的创建和销毁所带来的的性能消耗，而建立的一种池化技术，它是把已经创建的线程放入到池中。当有任务来临的时候可以重用已有的线程，无需等待创建的过程，这用有效的提高程序的响应速度。

为什么用ThreadPoolExecutor创建而不是Executors

线程池不允许使用Executors去创建，而是通过ThreadPoolExecutor的方式，这样的处理方式让写的读者更加明确线程池的运行规则，规避资源耗尽的风险。

Executors返回的线程池对象的弊端：

newFixedThreadPool
newSingleThreadExecutor
newCachedThreadPool

上面三个方法的底层原理都是使用ThreadPoolExecutor

newSingleThreadScheduledExecutor

允许请求的队列长度为Integer.MAX_VALUE可能会堆积大量的请求，从而导致OOM

ThreadPoolExecutor构造

/*** Creates a new {@code ThreadPoolExecutor} with the given initial* parameters.** @param corePoolSize the number of threads to keep in the pool, even*        if they are idle, unless {@code allowCoreThreadTimeOut} is set//表示线程池的常驻核心数，如果设置为0，则表示在没有任何任务，销毁线程池；如果大于0，即使没有任务，也会保证线程池的线程数量等于此值。如果设置过小，频繁的创建和销毁，设置过大，浪费系统资源。* @param maximumPoolSize the maximum number of threads to allow in the*        pool必须大于0，大于corePoolSize。此值只有在任务比较多，且不能存放在任务队列时，才能用到。* @param 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.表示线程的存活时间，当线程池空闲时间并且超过了此时间，多余的线程就会销毁到线程池的核心线程数的数量为止。* @param unit the time unit for the {@code keepAliveTime} argumentkeepAliveTime的时间单位* @param workQueue the queue to use for holding tasks before they are*        executed.  This queue will hold only the {@code Runnable}*        tasks submitted by the {@code execute} method.线程池的任务队列，线程池的所有线程都在处理任务的时候，新任务缓存到这里排队等到执行* @param threadFactory the factory to use when the executor*        creates a new thread通常在创建线程池时不指定此参数，会使用默认的线程创建工厂的模式，创建线程。* @param handler the handler to use when execution is blocked*        because the thread bounds and queue capacities are reached线程池的拒绝策略，如果workQueue存储满之后，不能创建新的线程来执行任务，就会使用拒绝策略，属于限流保护机制* @throws IllegalArgumentException if one of the following holds:<br>*         {@code corePoolSize < 0}<br>*         {@code keepAliveTime < 0}<br>*         {@code maximumPoolSize <= 0}<br>*         {@code maximumPoolSize < corePoolSize}* @throws NullPointerException if {@code workQueue}*         or {@code threadFactory} or {@code handler} is null*/
public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory,RejectedExecutionHandler handler) {if (corePoolSize < 0 ||maximumPoolSize <= 0 ||maximumPoolSize < corePoolSize ||keepAliveTime < 0)throw new IllegalArgumentException();if (workQueue == null || threadFactory == null || handler == null)throw new NullPointerException();this.acc = System.getSecurityManager() == null ?null :AccessController.getContext();this.corePoolSize = corePoolSize;this.maximumPoolSize = maximumPoolSize;this.workQueue = workQueue;this.keepAliveTime = unit.toNanos(keepAliveTime);this.threadFactory = threadFactory;this.handler = handler;}

第六个参数threadFactory默认的线程创建工厂

public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue) {// Executors.defaultThreadFactory() 为默认的线程创建工厂this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,Executors.defaultThreadFactory(), defaultHandler);
}
public static ThreadFactory defaultThreadFactory() {return new DefaultThreadFactory();
}
// 默认的线程创建工厂，需要实现 ThreadFactory 接口
static class DefaultThreadFactory implements ThreadFactory {private static final AtomicInteger poolNumber = new AtomicInteger(1);private final ThreadGroup group;private final AtomicInteger threadNumber = new AtomicInteger(1);private final String namePrefix;DefaultThreadFactory() {SecurityManager s = System.getSecurityManager();group = (s != null) ? s.getThreadGroup() :Thread.currentThread().getThreadGroup();namePrefix = "pool-" +poolNumber.getAndIncrement() +"-thread-";}// 创建线程public Thread newThread(Runnable r) {Thread t = new Thread(group, r,namePrefix + threadNumber.getAndIncrement(),0);if (t.isDaemon()) t.setDaemon(false); // 创建一个非守护线程if (t.getPriority() != Thread.NORM_PRIORITY)t.setPriority(Thread.NORM_PRIORITY); // 线程优先级设置为默认值return t;}
}

第七个参数RejectedExecutionHandler handler

public void execute(Runnable command) {if (command == null)throw new NullPointerException();int c = ctl.get();// 当前工作的线程数小于核心线程数if (workerCountOf(c) < corePoolSize) {// 创建新的线程执行此任务if (addWorker(command, true))return;c = ctl.get();}// 检查线程池是否处于运行状态，如果是则把任务添加到队列if (isRunning(c) && workQueue.offer(command)) {int recheck = ctl.get();// 再出检查线程池是否处于运行状态，防止在第一次校验通过后线程池关闭// 如果是非运行状态，则将刚加入队列的任务移除if (! isRunning(recheck) && remove(command))reject(command);// 如果线程池的线程数为 0 时（当 corePoolSize 设置为 0 时会发生）else if (workerCountOf(recheck) == 0)addWorker(null, false); // 新建线程执行任务}// 核心线程都在忙且队列都已爆满，尝试新启动一个线程执行失败else if (!addWorker(command, false)) // 执行拒绝策略reject(command);
}

addWorker(Runnable firstTask, boolean core)

第一个参数表示，线程应首先运行的任务，如果没有则设置为null

第二个参数，true表示使用corePoolSize作为阀值，false表示，maximumPoolSize作为阀值。

任务使用线程图：

在这里插入图片描述

execute与submit方法的区别

ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 10, 10L,TimeUnit.SECONDS, new LinkedBlockingQueue(20));
// execute 使用
executor.execute(new Runnable() {@Overridepublic void run() {System.out.println("Hello, execute.");}
});
// submit 使用
Future<String> future = executor.submit(new Callable<String>() {@Overridepublic String call() throws Exception {System.out.println("Hello, submit.");return "Success";}
});
System.out.println(future.get());

两者都是执行线程池任务的，submit可以接受返回值。submit配合future使用。

submit是继承executorService中的方法，execute是继承Executor中的方法。

线程池的拒绝策略

AbortPolicy,终止策略，线程池会抛出异常并终止执行，他是默认的拒绝策略

CallerRunsPolicy，把任务交给当前线程来执行

DiscardPolicy，忽略此任务（最新的任务）

DiscardOldestPolicy，忽略最早的任务（最先加入队列的任务）

 ThreadPoolExecutor executor=new ThreadPoolExecutor(1,3,10,TimeUnit.SECONDS,new LinkedBlockingQueue<>(2),new ThreadPoolExecutor.AbortPolicy());for (int i = 0; i < 6; i++) {executor.execute(()->{System.out.println(Thread.currentThread().getName());});}

运行结果

pool-1-thread-2
pool-1-thread-1
pool-1-thread-3
pool-1-thread-1
pool-1-thread-3
Exception in thread "main" java.util.concurrent.RejectedExecutionException: Task com.example.tangtang.boot.launch.LeetCodeSolution.ThreadExample3$$Lambda$1/1496724653@48533e64 rejected from java.util.concurrent.ThreadPoolExecutor@64a294a6[Running, pool size = 3, active threads = 0, queued tasks = 0, completed tasks = 5]at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2063)at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:830)at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1379)at com.example.tangtang.boot.launch.LeetCodeSolution.ThreadExample3.main(ThreadExample3.java:9)

分析：采用了AbortPolicy拒绝策略，抛出了异常，队列中可以存储两个任务，最大可以创建三个线程来完成任务（2+3=5），所以第六个来了就会抛出异常。

自定义拒绝策略

ThreadPoolExecutor executor=new ThreadPoolExecutor(1, 3, 10, TimeUnit.SECONDS, new LinkedBlockingQueue<>(2), new RejectedExecutionHandler() {@Overridepublic void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {System.out.println("这是自定义的拒绝策略");}});for (int i = 0; i <15; i++) {executor.execute(()->{System.out.println(Thread.currentThread().getName());});}

结果如下：

pool-1-thread-1
pool-1-thread-2
pool-1-thread-2
pool-1-thread-2
这是自定义的拒绝策略
这是自定义的拒绝策略
这是自定义的拒绝策略
这是自定义的拒绝策略
这是自定义的拒绝策略
这是自定义的拒绝策略
这是自定义的拒绝策略
这是自定义的拒绝策略
pool-1-thread-2
pool-1-thread-2
pool-1-thread-3

ThreadPoolExecutor扩展方法

beforeExecute
afterExecute

public class ThreadPoolExtend {public static void main(String[] args) {MyThreadPoolExecutor executor=new MyThreadPoolExecutor(2,4,10, TimeUnit.SECONDS,new LinkedBlockingQueue());for (int i = 0; i < 3; i++) {executor.execute(()->{Thread.currentThread().getName();});}}static class MyThreadPoolExecutor extends ThreadPoolExecutor{private final ThreadLocal<Long> localTime = new ThreadLocal<>();public MyThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);}@Overrideprotected void beforeExecute(Thread t, Runnable r) {Long sTime = System.nanoTime();// 开始时间 (单位：纳秒)localTime.set(sTime);System.out.println(String.format("%s | before | time=%s",t.getName(),sTime));super.beforeExecute(t,r);}@Overrideprotected void afterExecute(Runnable r,Throwable t){Long eTime = System.nanoTime();// 开始时间 (单位：纳秒)Long totalTime = eTime - localTime.get(); // 执行总时间System.out.println(String.format("%s | after | time=%s | 耗时：%s 毫秒",Thread.currentThread().getName(),eTime,(totalTime/1000000.0)));super.afterExecute(r,t);}}