ThreadPoolTaskExecutor和ThreadPoolExecutor有何区别?

2022-01-20 13:33:08 阅读：107 来源： 互联网

标签：int 有何 corePoolSize 线程 new ThreadPoolTaskExecutor ThreadPoolExecutor

ThreadPoolTaskExecutor是spring core包中的，而ThreadPoolExecutor是JDK中的JUC。ThreadPoolTaskExecutor是对ThreadPoolExecutor进行了封装处理。

自己在之前写多线程代码的时候都是这么玩的executor=Executors.newCachedThreadPool();但是有一次在大量数据的时候由于入库速度远大于出库速度导致内存急剧膨胀最后悲剧了重写代码，原来spring 早就给我们做好封装了。

来看一下ThreadPoolExecutor结构，祖类都是调用Executor接口：

再来看一下ThreadPoolTaskExecutor结构，祖类都是调用Executor接口：

再来看一下源码：

public class ThreadPoolTaskExecutor extends ExecutorConfigurationSupport implements SchedulingTaskExecutor {
    private final Object poolSizeMonitor = new Object();
    private int corePoolSize = 1;
    private int maxPoolSize = 2147483647;
    private int keepAliveSeconds = 60;
    private boolean allowCoreThreadTimeOut = false;
    private int queueCapacity = 2147483647;
    private ThreadPoolExecutor threadPoolExecutor;   //这里就用到了ThreadPoolExecutor

这是ThreadPoolTaskExecutor用来初始化threadPoolExecutor的方法，BlockingQueue是一个阻塞队列，这个我们先不管。由于ThreadPoolTaskExecutor的实现方式完全是使用threadPoolExecutor进行实现，我们需要知道这个threadPoolExecutor的一些参数。

 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.corePoolSize = corePoolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.workQueue = workQueue;
        this.keepAliveTime = unit.toNanos(keepAliveTime);
        this.threadFactory = threadFactory;
        this.handler = handler;
    }

int corePoolSize:线程池维护线程的最小数量.

int maximumPoolSize:线程池维护线程的最大数量.

long keepAliveTime:空闲线程的存活时间.

TimeUnit unit: 时间单位,现有纳秒,微秒,毫秒,秒枚举值.

BlockingQueue<Runnable> workQueue:持有等待执行的任务队列. 　

RejectedExecutionHandler handler: 用来拒绝一个任务的执行，有两种情况会发生这种情况。

一是在execute方法中若addIfUnderMaximumPoolSize(command)为false，即线程池已经饱和；
二是在execute方法中, 发现runState!=RUNNING || poolSize == 0,即已经shutdown,就调用ensureQueuedTaskHandled(Runnable command)，在该方法中有可能调用reject。

ThreadPoolExecutor池子的处理流程如下：　　

1）当池子大小小于corePoolSize就新建线程，并处理请求

2）当池子大小等于corePoolSize，把请求放入workQueue中，池子里的空闲线程就去从workQueue中取任务并处理

3）当workQueue放不下新入的任务时，新建线程入池，并处理请求，如果池子大小撑到了maximumPoolSize就用RejectedExecutionHandler来做拒绝处理

4）另外，当池子的线程数大于corePoolSize的时候，多余的线程会等待keepAliveTime长的时间，如果无请求可处理就自行销毁

其会优先创建 CorePoolSize 线程，当继续增加线程时，先放入Queue中，当 CorePoolSize 和 Queue 都满的时候，就增加创建新线程，当线程达到MaxPoolSize的时候，就会抛出错误 org.springframework.core.task.TaskRejectedException

另外MaxPoolSize的设定如果比系统支持的线程数还要大时，会抛出java.lang.OutOfMemoryError: unable to create new native thread 异常。

 <!-- 异步线程池 -->
    <bean id="threadPool"
        class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
        <!-- 核心线程数，默认为1 -->
        <property name="corePoolSize" value="3" />
        <!-- 最大线程数，默认为Integer.Max_value -->
        <property name="maxPoolSize" value="10" />
        <!-- 队列最大长度 >=mainExecutor.maxSize -->
        <property name="queueCapacity" value="25" />
        <!-- 线程池维护线程所允许的空闲时间 -->
        <property name="keepAliveSeconds" value="300" />
        <!-- 线程池对拒绝任务(无线程可用)的处理策略 ThreadPoolExecutor.CallerRunsPolicy策略 ,调用者的线程会执行该任务,如果执行器已关闭,则丢弃.  -->
        <property name="rejectedExecutionHandler">
　　　　　　　　<!-- AbortPolicy:直接抛出java.util.concurrent.RejectedExecutionException异常 -->
　　　　　　　　<!-- CallerRunsPolicy:若已达到待处理队列长度，将由主线程直接处理请求 -->
　　　　　　　　<!-- DiscardOldestPolicy:抛弃旧的任务；会导致被丢弃的任务无法再次被执行 -->
　　　　　　　　<!-- DiscardPolicy:抛弃当前任务；会导致被丢弃的任务无法再次被执行 -->
            <bean class="java.util.concurrent.ThreadPoolExecutor$CallerRunsPolicy" />
        </property>
    </bean>

Reject策略预定义有四种：
(1)ThreadPoolExecutor.AbortPolicy策略，是默认的策略,处理程序遭到拒绝将抛出运行时RejectedExecutionException。

(2)ThreadPoolExecutor.CallerRunsPolicy策略 ,调用者的线程会执行该任务,如果执行器已关闭,则丢弃.

(3)ThreadPoolExecutor.DiscardPolicy策略，不能执行的任务将被丢弃.

(4)ThreadPoolExecutor.DiscardOldestPolicy策略，如果执行程序尚未关闭，则位于工作队列头部的任务将被删除，然后重试执行程序（如果再次失败，则重复此过程）.

关于callable回调方法（因为为队列阻塞，如果到取值某个执行的值会等待执行完成）

        ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
        threadPoolTaskExecutor.setCorePoolSize(5);
        threadPoolTaskExecutor.setMaxPoolSize(50);
        threadPoolTaskExecutor.initialize();

        List<String> paymentSeqNoList = new ArrayList<>();
        for (int i = 0; i < 100; i++) {
            paymentSeqNoList.add(String.valueOf(i));
        }
        Long startTime = System.currentTimeMillis();
        Map<String, FutureTask<String>> futureMap = new HashMap<String, FutureTask<String>>();
        //线程池提交返回
        for (String paymentSeqNo : paymentSeqNoList) {
            FutureTask<String> futureTask = new FutureTask<String>(new MyTestCallable(paymentSeqNo));
            futureMap.put(paymentSeqNo, futureTask);
            // submit提交执行
            threadPoolTaskExecutor.submit(futureTask);
        }
        Long endTime = System.currentTimeMillis();
        System.out.println("耗时1：" + (endTime - startTime));

关于callable回调值监听是否成功，JDK1.8 也开始支持guava方法了，guava有ListenableFuture 返回优化如下：

  Long startTime2 = System.currentTimeMillis();
        ListenableFuture<String> listenableFuture = null;
        for (String paymentSeqNo : paymentSeqNoList) {
            ListeningExecutorService executorService = MoreExecutors.listeningDecorator(Executors.newCachedThreadPool());
            listenableFuture = executorService.submit(new Callable<String>() {
                @Override
                public String call() throws Exception {
                    return "成功";
                }
            });
        }
        //监听事件
        Futures.addCallback(listenableFuture, new FutureCallback<String>() {
            @Override
            public void onSuccess(String result) {
                System.out.println("get listenable future's result with callback " + result);
            }
            @Override
            public void onFailure(Throwable t) {
                t.printStackTrace();
            }
        });
        Long endTime2 = System.currentTimeMillis();
        System.out.println("耗时2：" + (endTime2 - startTime2));

标签：int,有何,corePoolSize,线程,new,ThreadPoolTaskExecutor,ThreadPoolExecutor
来源： https://www.cnblogs.com/cchilei/p/15826017.html

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。

ICode9

ThreadPoolTaskExecutor和ThreadPoolExecutor有何区别?