java线程池ThreadPoolExecutor的八种拒绝策略示例详解

目录
  • 池化设计思想
  • 线程池触发拒绝策略的时机
  • JDK内置4种线程池拒绝策略
    • 拒绝策略接口定义
  • AbortPolicy(中止策略)
  • DiscardPolicy(丢弃策略)
  • DiscardOldestPolicy(弃老策略)
  • 第三方实现的拒绝策略
    • Dubbo 中的线程拒绝策略
    • Netty 中的线程池拒绝策略
    • ActiveMQ 中的线程池拒绝策略
    • PinPoint 中的线程池拒绝策略

谈到 Java 的线程池最熟悉的莫过于 ExecutorService 接口了,jdk1.5 新增的 java.util.concurrent 包下的这个 api,大大的简化了多线程代码的开发。而不论你用 FixedThreadPool 还是 CachedThreadPool 其背后实现都是ThreadPoolExecutor。

池化设计思想

池话设计应该不是一个新名词。我们常见的如 Java 线程池、JDBC 连接池、Redis 连接池等就是这类设计的代表实现。这种设计会初始预设资源,解决的问题就是抵消每次获取资源的消耗,如创建线程的开销,获取远程连接的开销等。

就好比你去食堂打饭,打饭的大妈会先把饭盛好几份放那里,你来了就直接拿着饭盒加菜即可,不用再临时又盛饭又打菜,效率就高了。

除了初始化资源,池化设计还包括如下这些特征:池子的初始值、池子的活跃值、池子的最大值等,这些特征可以直接映射到 Java 线程池和数据库连接池的成员属性中。

线程池触发拒绝策略的时机

和数据源连接池不一样,线程池除了初始大小和池子最大值,还多了一个阻塞队列来缓冲。数据源连接池一般请求的连接数超过连接池的最大值的时候就会触发拒绝策略,策略一般是阻塞等待设置的时间或者直接抛异常。

想要了解线程池什么时候触发拒绝粗略,需要明确上面三个参数的具体含义,是这三个参数总体协调的结果,而不是简单的超过最大线程数就会触发线程拒绝粗略,当提交的任务数大于 corePoolSize 时,会优先放到队列缓冲区,只有填满了缓冲区后,才会判断当前运行的任务是否大于 maxPoolSize,小于时会新建线程处理。大于时就触发了拒绝策略,总结就是:当前提交任务数大于(maxPoolSize + queueCapacity)时就会触发线程池的拒绝策略了。

JDK内置4种线程池拒绝策略

拒绝策略接口定义

在分析 JDK 自带的线程池拒绝策略前,先看下 JDK 定义的 拒绝策略接口,如下:

public interface RejectedExecutionHandler {
    void rejectedExecution(Runnable r, ThreadPoolExecutor executor);
}

接口定义很明确,当触发拒绝策略时,线程池会调用你设置的具体的策略,将当前提交的任务以及线程池实例本身传递给你处理,具体作何处理,不同场景会有不同的考虑,下面看 JDK 为我们内置了哪些实现:

CallerRunsPolicy(调用者运行策略)

   public static class CallerRunsPolicy implements RejectedExecutionHandler {
         public CallerRunsPolicy() { }
          public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
             if (!e.isShutdown()) {
               r.run();
             }
         }
   }

功能:当触发拒绝策略时,只要线程池没有关闭,就由提交任务的当前线程处理。

使用场景:一般在不允许失败的、对性能要求不高、并发量较小的场景下使用,因为线程池一般情况下不会关闭,也就是提交的任务一定会被运行,但是由于是调用者线程自己执行的,当多次提交任务时,就会阻塞后续任务执行,性能和效率自然就慢了。

AbortPolicy(中止策略)

     public static class AbortPolicy implements RejectedExecutionHandler {
         public AbortPolicy() { }
         public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
             throw new RejectedExecutionException("Task " + r.toString() +
                                                  " rejected from " +
                                                  e.toString());
         }
    }

功能:当触发拒绝策略时,直接抛出拒绝执行的异常,中止策略的意思也就是打断当前执行流程

使用场景:这个就没有特殊的场景了,但是一点要正确处理抛出的异常。ThreadPoolExecutor 中默认的策略就是AbortPolicy,ExecutorService 接口的系列 ThreadPoolExecutor 因为都没有显示的设置拒绝策略,所以默认的都是这个。但是请注意,ExecutorService 中的线程池实例队列都是无界的,也就是说把内存撑爆了都不会触发拒绝策略。当自己自定义线程池实例时,使用这个策略一定要处理好触发策略时抛的异常,因为他会打断当前的执行流程。

DiscardPolicy(丢弃策略)

    public static class DiscardPolicy implements RejectedExecutionHandler {
        public DiscardPolicy() { }
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        }
    }

功能:直接静悄悄的丢弃这个任务,不触发任何动作

使用场景:如果你提交的任务无关紧要,你就可以使用它 。因为它就是个空实现,会悄无声息的吞噬你的的任务。所以这个策略基本上不用了

DiscardOldestPolicy(弃老策略)

     public static class DiscardOldestPolicy implements RejectedExecutionHandler {
         public DiscardOldestPolicy() { }
         public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
             if (!e.isShutdown()) {
                 e.getQueue().poll();
                 e.execute(r);
             }
        }
    }

功能:如果线程池未关闭,就弹出队列头部的元素,然后尝试执行

使用场景:这个策略还是会丢弃任务,丢弃时也是毫无声息,但是特点是丢弃的是老的未执行的任务,而且是待执行优先级较高的任务。基于这个特性,我能想到的场景就是,发布消息,和修改消息,当消息发布出去后,还未执行,此时更新的消息又来了,这个时候未执行的消息的版本比现在提交的消息版本要低就可以被丢弃了。因为队列中还有可能存在消息版本更低的消息会排队执行,所以在真正处理消息的时候一定要做好消息的版本比较。

第三方实现的拒绝策略

Dubbo 中的线程拒绝策略

 public class AbortPolicyWithReport extends ThreadPoolExecutor.AbortPolicy {
      protected static final Logger logger = LoggerFactory.getLogger(AbortPolicyWithReport.class);
      private final String threadName;
      private final URL url;
      private static volatile long lastPrintTime = 0;
    private static Semaphore guard = new Semaphore(1);
    public AbortPolicyWithReport(String threadName, URL url) {
        this.threadName = threadName;
        this.url = url;
    }
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        String msg = String.format("Thread pool is EXHAUSTED!" +
                        " Thread Name: %s, Pool Size: %d (active: %d, core: %d, max: %d, largest: %d), Task: %d (completed: %d)," +
                        " Executor status:(isShutdown:%s, isTerminated:%s, isTerminating:%s), in %s://%s:%d!",
                threadName, e.getPoolSize(), e.getActiveCount(), e.getCorePoolSize(), e.getMaximumPoolSize(), e.getLargestPoolSize(),
                e.getTaskCount(), e.getCompletedTaskCount(), e.isShutdown(), e.isTerminated(), e.isTerminating(),
                url.getProtocol(), url.getIp(), url.getPort());
        logger.warn(msg);
        dumpJStack();
        throw new RejectedExecutionException(msg);
    }
    private void dumpJStack() {
       //省略实现
    }
}

可以看到,当dubbo的工作线程触发了线程拒绝后,主要做了三个事情,原则就是尽量让使用者清楚触发线程拒绝策略的真实原因

  • 输出了一条警告级别的日志,日志内容为线程池的详细设置参数,以及线程池当前的状态,还有当前拒绝任务的一些详细信息。可以说,这条日志,使用dubbo的有过生产运维经验的或多或少是见过的,这个日志简直就是日志打印的典范,其他的日志打印的典范还有spring。得益于这么详细的日志,可以很容易定位到问题所在
  • 输出当前线程堆栈详情,这个太有用了,当你通过上面的日志信息还不能定位问题时,案发现场的dump线程上下文信息就是你发现问题的救命稻草,这个可以参考《dubbo线程池耗尽事件-"CyclicBarrier惹的祸"》
  • 继续抛出拒绝执行异常,使本次任务失败,这个继承了JDK默认拒绝策略的特性

Netty 中的线程池拒绝策略

     private static final class NewThreadRunsPolicy implements RejectedExecutionHandler {
         NewThreadRunsPolicy() {
             super();
         }
          public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
             try {
                 final Thread t = new Thread(r, "Temporary task executor");
                 t.start();
            } catch (Throwable e) {
                throw new RejectedExecutionException(
                        "Failed to start a new thread", e);
            }
        }
    }

Netty 中的实现很像 JDK 中的 CallerRunsPolicy,舍不得丢弃任务。不同的是,CallerRunsPolicy 是直接在调用者线程执行的任务。而 Netty是新建了一个线程来处理的。所以,Netty的实现相较于调用者执行策略的使用面就可以扩展到支持高效率高性能的场景了。但是也要注意一点,Netty的实现里,在创建线程时未做任何的判断约束,也就是说只要系统还有资源就会创建新的线程来处理,直到new不出新的线程了,才会抛创建线程失败的异常

ActiveMQ 中的线程池拒绝策略

  new RejectedExecutionHandler() {
                 @Override
                 public void rejectedExecution(final Runnable r, final ThreadPoolExecutor executor) {
                     try {
                         executor.getQueue().offer(r, 60, TimeUnit.SECONDS);
                     } catch (InterruptedException e) {
                         throw new RejectedExecutionException("Interrupted waiting for BrokerService.worker");
                     }
                     throw new RejectedExecutionException("Timed Out while attempting to enqueue Task.");
                }
            });

activeMq中的策略属于最大努力执行任务型,当触发拒绝策略时,在尝试一分钟的时间重新将任务塞进任务队列,当一分钟超时还没成功时,就抛出异常

PinPoint 中的线程池拒绝策略

 public class RejectedExecutionHandlerChain implements RejectedExecutionHandler {
     private final RejectedExecutionHandler[] handlerChain;
     public static RejectedExecutionHandler build(List<RejectedExecutionHandler> chain) {
         Objects.requireNonNull(chain, "handlerChain must not be null");
         RejectedExecutionHandler[] handlerChain = chain.toArray(new RejectedExecutionHandler[0]);
         return new RejectedExecutionHandlerChain(handlerChain);
     }
     private RejectedExecutionHandlerChain(RejectedExecutionHandler[] handlerChain) {
        this.handlerChain = Objects.requireNonNull(handlerChain, "handlerChain must not be null");
    }
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        for (RejectedExecutionHandler rejectedExecutionHandler : handlerChain) {
            rejectedExecutionHandler.rejectedExecution(r, executor);
        }
    }
}

pinpoint的拒绝策略实现很有特点,和其他的实现都不同。他定义了一个拒绝策略链,包装了一个拒绝策略列表,当触发拒绝策略时,会将策略链中的rejectedExecution依次执行一遍。

以上就是java线程池ThreadPoolExecutor的八种拒绝策略示例详解的详细内容,更多关于ThreadPoolExecutor的八种拒绝策略的资料请关注我们其它相关文章!

时间: 2021-11-24

Java ThreadPoolExecutor的参数深入理解

Java ThreadPoolExecutor的参数深入理解 一.使用Executors创建线程池    之前创建线程的时候都是用的Executors的newFixedThreadPool(),newSingleThreadExecutor(),newCachedThreadPool()这三个方法.当然Executors也是用不同的参数去new ThreadPoolExecutor     1. newFixedThreadPool() 创建线程数固定大小的线程池. 由于使用了LinkedBlo

详解什么是Java线程池的拒绝策略?

拒绝策略 (JDK提供了4种,另外也可以自定义拒绝策略,因此总共有5种.) 线程池中的线程已经用完了,无法继续为新任务服务,同时,等待队列也已经排满了,再也塞不下新任务了.这时候我们就需要拒绝策略机制合理的处理这个问题. JDK 内置的拒绝策略如下: 1.AbortPolicy : 直接抛出异常,阻止系统正常运行. 2.CallerRunsPolicy : 只要线程池未关闭,该策略直接在调用者线程中,运行当前被丢弃的任务.显然这样做不会真的丢弃任务,但是,任务提交线程的性能极有可能会急剧下降.

Java线程池ThreadPoolExecutor原理及使用实例

引导 要求:线程资源必须通过线程池提供,不允许在应用自行显式创建线程: 说明:使用线程池的好处是减少在创建和销毁线程上所花的时间以及系统资源的开销,解决资源不足的问题.如果不使用线程池,有可能造成系统创建大量同类线程而导致消耗内存或者"过度切换"的问题. 线程池介绍线程池概述   线程池,顾名思义是一个放着线程的池子,这个池子的线程主要是用来执行任务的.当用户提交任务时,线程池会创建线程去执行任务,若任务超过了核心线程数的时候,会在一个任务队列里进行排队等待,这个详细流程,我们会后面细

Java线程池的拒绝策略实现详解

一.简介 jdk1.5 版本新增了JUC并发编程包,大大的简化了传统的多线程开发. Java线程池,是典型的池化思想的产物,类似的还有数据库的连接池.redis的连接池等.池化思想,就是在初始的时候去申请资源,创建一批可使用的连接,这样在使用的时候,就不必再进行创建连接信息的开销了.举个生活中鲜明的例子,在去著名洋快餐某基或者某劳的时候,配餐人员是字节从一个中间的保温箱里面直接取,然后打包就好了.不用再临时的来了一个单子,又要去拿原材料,又要去进行加工.效率明显的就是提高了很多. 既然是池子,那

java线程池工作队列饱和策略代码示例

线程池(Thread Pool) 是并行执行任务收集的实用工具.随着 CPU 引入适合于应用程序并行化的多核体系结构,线程池的作用正日益显现.通过 ThreadPoolExecutor类及其他辅助类,Java 5 引入了这一框架,作为新的并发支持部分. ThreadPoolExecutor框架灵活且功能强大,它支持特定于用户的配置并提供了相关的挂钩(hook)和饱和策略来处理满队列 Java线程池会将提交的任务先置于工作队列中,在从工作队列中获取(SynchronousQueue直接由生产者提交

ThreadPoolExecutor线程池原理及其execute方法(详解)

jdk1.7.0_79 对于线程池大部分人可能会用,也知道为什么用.无非就是任务需要异步执行,再者就是线程需要统一管理起来.对于从线程池中获取线程,大部分人可能只知道,我现在需要一个线程来执行一个任务,那我就把任务丢到线程池里,线程池里有空闲的线程就执行,没有空闲的线程就等待.实际上对于线程池的执行原理远远不止这么简单. 在Java并发包中提供了线程池类--ThreadPoolExecutor,实际上更多的我们可能用到的是Executors工厂类为我们提供的线程池:newFixedThreadP

python多进程使用及线程池的使用方法代码详解

多进程:主要运行multiprocessing模块 import os,time import sys from multiprocessing import Process class MyProcess(Process): """docstring for MyProcess""" def __init__(self, arg, callback): super(MyProcess, self).__init__() self.arg = a

线程池的原理与实现详解

一. 线程池的简介通常我们使用多线程的方式是,需要时创建一个新的线程,在这个线程里执行特定的任务,然后在任务完成后退出.这在一般的应用里已经能够满足我们应用的需求,毕竟我们并不是什么时候都需要创建大量的线程,并在它们执行一个简单的任务后销毁. 但是在一些web.email.database等应用里,比如彩铃,我们的应用在任何时候都要准备应对数目巨大的连接请求,同时,这些请求所要完成的任务却又可能非常的简单,即只占用很少的处理时间.这时,我们的应用有可能处于不停的创建线程并销毁线程的状态.虽说比起

Java线程安全和锁Synchronized知识点详解

一.进程与线程的概念 (1)在传统的操作系统中,程序并不能独立运行,作为资源分配和独立运行的基本单位都是进程. 在未配置 OS 的系统中,程序的执行方式是顺序执行,即必须在一个程序执行完后,才允许另一个程序执行:在多道程序环境下,则允许多个程序并发执行.程序的这两种执行方式间有着显著的不同.也正是程序并发执行时的这种特征,才导致了在操作系统中引入进程的概念. 自从在 20 世纪 60 年代人们提出了进程的概念后,在 OS 中一直都是以进程作为能拥有资源和独立运行的基本单位的.直到 20 世纪 8

Java分代垃圾回收策略原理详解

一.为什么要分代 分代的垃圾回收策略,是基于这样一个事实:不同的对象的生命周期是不一样的.因此,不同生命周期的对象可以采取不同的收集方式,以便提高回收效率. 在Java程序运行的过程中,会产生大量的对象,其中有些对象是与业务信息相关,比如Http请求中的Session对象.线程.Socket连接,这类对象跟业务直接挂钩,因此生命周期比较长.但是还有一些对象,主要是程序运行过程中生成的临时变量,这些对象生命周期会比较短,比如:String对象,由于其不变类的特性,系统会产生大量的这些对象,有些对象

Java线程池FutureTask实现原理详解

前言 线程池可以并发执行多个任务,有些时候,我们可能想要跟踪任务的执行结果,甚至在一定时间内,如果任务没有执行完成,我们可能还想要取消任务的执行,为了支持这一特性,ThreadPoolExecutor提供了 FutureTask 用于追踪任务的执行和取消.本篇介绍FutureTask的实现原理. 类视图 为了更好的理解FutureTask的实现原理,这里先提供几个重要接口和类的结构,如下图所示: RunnableAdapter ThreadPoolExecutor提供了submit接口用于提交任

Java线程池的几种实现方法和区别介绍实例详解

下面通过实例代码为大家介绍Java线程池的几种实现方法和区别: import java.text.DateFormat; import java.text.SimpleDateFormat; import java.util.ArrayList; import java.util.Date; import java.util.List; import java.util.Random; import java.util.concurrent.Callable; import java.util.

Java线程池使用与原理详解

线程池是什么? 我们可以利用java很容易创建一个新线程,同时操作系统创建一个线程也是一笔不小的开销.所以基于线程的复用,就提出了线程池的概念,我们使用线程池创建出若干个线程,执行完一个任务后,该线程会存在一段时间(用户可以设定空闲线程的存活时间,后面会介绍),等到新任务来的时候就直接复用这个空闲线程,这样就省去了创建.销毁线程损耗.当然空闲线程也会是一种资源的浪费(所有才有空闲线程存活时间的限制),但总比频繁的创建销毁线程好太多. 下面是我的测试代码 /* * @TODO 线程池测试 */ @