ThreadPoolExecutor-线程池知多少
本文共 10442 字，大约阅读时间需要 34 分钟。
ThreadPoolExecutor
创建线程池的目的？
线程池为了避免频繁的创建和销毁所带来的的性能消耗，而建立的一种池化技术，它是把已经创建的线程放入到池中。当有任务来临的时候可以重用已有的线程，无需等待创建的过程，这用有效的提高程序的响应速度。
为什么用ThreadPoolExecutor创建而不是Executors
线程池不允许使用Executors去创建，而是通过ThreadPoolExecutor的方式，这样的处理方式让写的读者更加明确线程池的运行规则，规避资源耗尽的风险。
Executors返回的线程池对象的弊端：
newFixedThreadPoolnewSingleThreadExecutornewCachedThreadPool
上面三个方法的底层原理都是使用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} argument keepAliveTime的时间单位 * @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: * {@code corePoolSize < 0} * {@code keepAliveTime < 0} * {@code maximumPoolSize <= 0} * {@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 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, BlockingQueueworkQueue) { // 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() { @Override public void run() { System.out.println("Hello, execute."); }});// submit 使用Future future = executor.submit(new Callable () { @Override public 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-2pool-1-thread-1pool-1-thread-3pool-1-thread-1pool-1-thread-3Exception 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() { @Override public 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-1pool-1-thread-2pool-1-thread-2pool-1-thread-2这是自定义的拒绝策略这是自定义的拒绝策略这是自定义的拒绝策略这是自定义的拒绝策略这是自定义的拒绝策略这是自定义的拒绝策略这是自定义的拒绝策略这是自定义的拒绝策略pool-1-thread-2pool-1-thread-2pool-1-thread-3
ThreadPoolExecutor扩展方法
beforeExecuteafterExecute
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 localTime = new ThreadLocal<>(); public MyThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue workQueue) { super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue); } @Override protected 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); } @Override protected 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); } } pool-1-thread-1 | before | time=38938852458000
pool-1-thread-2 | before | time=38938852540300 pool-1-thread-1 | after | time=38938882198300 | 耗时：29.7403 毫秒 pool-1-thread-2 | after | time=38938882205900 | 耗时：29.6656 毫秒 pool-1-thread-1 | before | time=38938883845000 pool-1-thread-1 | after | time=38938884124100 | 耗时：0.2791 毫秒转载地址：http://ahjmz.baihongyu.com/
你可能感兴趣的文章