1. 编程号首页
  2. 编程基础

自定义线程池详解_如何创建线程池[通俗易懂]

编程小号 • • 编程基础

自定义线程池详解_如何创建线程池[通俗易懂]

一、java.util.concurrent包下的ThreadPoolExecutor

线程池基本概念

概念：线程池主要是控制运行线程的数量，将待处理任务放到等待队列，然后创建线程执行这些任务。如果超过了最大线程数，则等待。
优点：

  1. 线程复用：不用一直new新线程，重复利用已经创建的线程来降低线程的创建和销毁开销，节省系统资源。
  2. 提高响应速度：当任务达到时，不用创建新的线程，直接利用线程池的线程。
  3. 管理线程：可以控制最大并发数，控制线程的创建等。
    体系：Executor→ExecutorService→AbstractExecutorService→ThreadPoolExecutor。ThreadPoolExecutor是线程池创建的核心类。类似Arrays、Collections工具类，Executor也有自己的工具类Executors。

线程池三种常用创建方式

newFixedThreadPool：使用LinkedBlockingQueue实现，定长线程池。

public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                                  0L, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue<Runnable>());
}

newSingleThreadExecutor：使用LinkedBlockingQueue实现，一池只有一个线程。

public static ExecutorService newSingleThreadExecutor() {
    return new FinalizableDelegatedExecutorService(new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
}

newCachedThreadPool：使用SynchronousQueue实现，变长线程池。

public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                 60L, TimeUnit.SECONDS,
                                 new SynchronousQueue<Runnable>());
}

线程池创建的七个参数

参数 意义
corePoolSize 线程池常驻核心线程数
maximumPoolSize 能够容纳的最大线程数
keepAliveTime 空闲线程存活时间
unit 存活时间单位
workQueue 存放提交但未执行任务的队列
threadFactory 创建线程的工厂类
handler 等待队列满后的拒绝策略

理解：线程池的创建参数，就像一个银行。

corePoolSize就像银行的“当值窗口“，比如今天有2位柜员在受理客户请求（任务）。如果超过2个客户，那么新的客户就会在等候区（等待队列workQueue）等待。当等候区也满了，这个时候就要开启“加班窗口”，让其它3位柜员来加班，此时达到最大窗口maximumPoolSize，为5个。如果开启了所有窗口，等候区依然满员，此时就应该启动”拒绝策略“handler，告诉不断涌入的客户，叫他们不要进入，已经爆满了。由于不再涌入新客户，办完事的客户增多，窗口开始空闲，这个时候就通过keepAlivetTime将多余的3个”加班窗口“取消，恢复到2个”当值窗口“。

线程池底层原理

原理图：上面银行的例子，实际上就是线程池的工作原理。

自定义线程池详解_如何创建线程池[通俗易懂]

流程图：

自定义线程池详解_如何创建线程池[通俗易懂]

新任务到达→

如果正在运行的线程数小于corePoolSize，创建核心线程；大于等于corePoolSize，放入等待队列。

如果等待队列已满，但正在运行的线程数小于maximumPoolSize，创建非核心线程；大于等于maximumPoolSize，启动拒绝策略。

当一个线程无事可做一段时间keepAliveTime后，如果正在运行的线程数大于corePoolSize，则关闭非核心线程。

线程池的拒绝策略

当等待队列满时，且达到最大线程数，再有新任务到来，就需要启动拒绝策略。JDK提供了四种拒绝策略，分别是。

  1. AbortPolicy：默认的策略，直接抛出RejectedExecutionException异常，阻止系统正常运行。
  2. CallerRunsPolicy：既不会抛出异常，也不会终止任务，而是将任务返回给调用者。
  3. DiscardOldestPolicy：抛弃队列中等待最久的任务，然后把当前任务加入队列中尝试再次提交任务。
  4. DiscardPolicy：直接丢弃任务，不做任何处理。

实际生产使用哪一个线程池？

单一、可变、定长都不用！原因就是FixedThreadPool和SingleThreadExecutor底层都是用LinkedBlockingQueue实现的，这个队列最大长度为Integer.MAX_VALUE，显然会导致OOM。所以实际生产一般自己通过ThreadPoolExecutor的7个参数，自定义线程池。

ExecutorService threadPool=new ThreadPoolExecutor(2,5,
                        1L,TimeUnit.SECONDS,
                        new LinkedBlockingQueue<>(3),
                        Executors.defaultThreadFactory(),
                        new ThreadPoolExecutor.AbortPolicy());

自定义线程池参数选择

对于CPU密集型任务，最大线程数是CPU线程数+1。对于IO密集型任务，尽量多配点，可以是CPU线程数*2，或者CPU线程数/(1-阻塞系数)。

二、SpringBoot 的默认线程池-ThreadPoolTaskExecutor

1、了解与使用

ThreadPoolTaskExecutor是springboot提供的默认线程池 。也就是说如果没有自定义线程池，那么会自动装配这个默认的。

In the absence of an Executor bean in the context, Spring Boot auto-configures a ThreadPoolTaskExecutor with sensible defaults that can be automatically associated to asynchronous task execution (@EnableAsync) and Spring MVC asynchronous request processing.

在springboot当中，根据 官方文档的说明，如果没有配置线程池的话，springboot会自动配置一个ThreadPoolTaskExecutor 线程池到bean当中，我们只需要按照他的方式调用就可以了！！

PS:

ThreadPoolExecutor：这个是JAVA自己实现的线程池执行类，基本上创建线程池都是通过这个类进行的创建！
ThreadPoolTaskExecutor ：这个是springboot基于ThreadPoolExecutor实现的一个线程池执行类。

这是默认提供的几个配置参数看一下

自定义线程池详解_如何创建线程池[通俗易懂]

拒绝策略默认是超出后抛出RejectedExecutionException异常并直接丢弃。

以下是springboot默认的线程池配置，可以在application.properties文件当中进行相关的设置！！！

# 核心线程数
spring.task.execution.pool.core-size=8  
# 最大线程数
spring.task.execution.pool.max-size=16
# 空闲线程存活时间
spring.task.execution.pool.keep-alive=60s
# 是否允许核心线程超时
spring.task.execution.pool.allow-core-thread-timeout=true
# 线程队列数量
spring.task.execution.pool.queue-capacity=100
# 线程关闭等待
spring.task.execution.shutdown.await-termination=false
spring.task.execution.shutdown.await-termination-period=
# 线程名称前缀
spring.task.execution.thread-name-prefix=task-

package com.xiaotong.cleandata.config;

import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

/**
 * 自定义线程池配置
 * @date 2021年2月21日14:28:26
 * @author Lvshiyang
 */
@Configuration
@ConfigurationProperties(prefix = "spring.task.execution")
public class ThreadPoolConfiguration {

    @Value("${pool.core-size}")
    private int corePoolSize;

    @Value("${pool.max-size}")
    private int maxPoolSize;

    @Value("${pool.queue-capacity}")
    private int queueCapacity;

    @Value("${pool.keep-alive}")
    private int keepAliveSeconds;

    @Value("${thread-name-prefix}")
    private String threadNamePrefix;

    @Bean(name = "threadPoolTaskExecutor")
    public ThreadPoolTaskExecutor threadPoolTaskExecutor() {
        ThreadPoolTaskExecutor pool = new ThreadPoolTaskExecutor();
        pool.setCorePoolSize(corePoolSize);//核心线程池数
        pool.setMaxPoolSize(maxPoolSize); // 最大线程数
        pool.setQueueCapacity(queueCapacity);//队列容量,当核心线程数达到最大时，新任务会放在队列中排队等待执行
        pool.setKeepAliveSeconds(keepAliveSeconds);//线程空闲时间
        pool.setAllowCoreThreadTimeOut(false);//核心线程会一直存活，即使没有任务需要执行。（默认false）时，核心线程会超时关闭
        pool.setThreadNamePrefix(threadNamePrefix);//线程前缀名称
        //默认:AbortPolicy 丢弃任务，抛运行时异常
        //CallerRunsPolicy由调用线程处理该任务
        pool.setRejectedExecutionHandler(new java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy()); 

        return pool; 
   } 
}

使用方式：

@Resource(name="threadPoolTaskExecutor")
ThreadPoolTaskExecutor threadPoolTaskExecutor;

threadPoolTaskExecutor.execute(() -> {
    //业务代码
});

如果想知道什么时候执行完可搭配CountDownLatch，主线程countDownLatch.await()等待即可。也可以使用待返回参数的Fulture.get()处理。

2、深入springboot默认的线程池

根据官方文档的说明，Spring Boot auto-configures a ThreadPoolTaskExecutor 。最终找到springboot的线程池自动装配类：TaskExecutionAutoConfiguration

    @Bean
    @ConditionalOnMissingBean
    public TaskExecutorBuilder taskExecutorBuilder(TaskExecutionProperties properties, ObjectProvider<TaskExecutorCustomizer> taskExecutorCustomizers, ObjectProvider<TaskDecorator> taskDecorator) {
        Pool pool = properties.getPool();
        TaskExecutorBuilder builder = new TaskExecutorBuilder();
        builder = builder.queueCapacity(pool.getQueueCapacity());
        builder = builder.corePoolSize(pool.getCoreSize());
        builder = builder.maxPoolSize(pool.getMaxSize());
        builder = builder.allowCoreThreadTimeOut(pool.isAllowCoreThreadTimeout());
        builder = builder.keepAlive(pool.getKeepAlive());
        Shutdown shutdown = properties.getShutdown();
        builder = builder.awaitTermination(shutdown.isAwaitTermination());
        builder = builder.awaitTerminationPeriod(shutdown.getAwaitTerminationPeriod());
        builder = builder.threadNamePrefix(properties.getThreadNamePrefix());
        Stream var10001 = taskExecutorCustomizers.orderedStream();
        var10001.getClass();
        builder = builder.customizers(var10001::iterator);
        builder = builder.taskDecorator((TaskDecorator)taskDecorator.getIfUnique());
        return builder;
    }

同时在ThreadPoolTaskExecutor源码当中可以看到线程池的初始化方式是直接调用的ThreadPoolExecutor进行的初始化。

protected ExecutorService initializeExecutor(ThreadFactory threadFactory, RejectedExecutionHandler rejectedExecutionHandler) {
       BlockingQueue<Runnable> queue = this.createQueue(this.queueCapacity);
       ThreadPoolExecutor executor;
       if (this.taskDecorator != null) {
           executor = new ThreadPoolExecutor(this.corePoolSize, this.maxPoolSize, (long)this.keepAliveSeconds, TimeUnit.SECONDS, queue, threadFactory, rejectedExecutionHandler) {
               public void execute(Runnable command) {
                   Runnable decorated = ThreadPoolTaskExecutor.this.taskDecorator.decorate(command);
                   if (decorated != command) {
                       ThreadPoolTaskExecutor.this.decoratedTaskMap.put(decorated, command);
                   }

                   super.execute(decorated);
               }
           };
       } else {
           executor = new ThreadPoolExecutor(this.corePoolSize, this.maxPoolSize, (long)this.keepAliveSeconds, TimeUnit.SECONDS, queue, threadFactory, rejectedExecutionHandler);
       }

       if (this.allowCoreThreadTimeOut) {
           executor.allowCoreThreadTimeOut(true);
       }

       this.threadPoolExecutor = executor;
       return executor;
   }

同时会发现默认的线程池拒绝策略是： AbortPolicy 直接抛出异常！！！

private RejectedExecutionHandler rejectedExecutionHandler = new AbortPolicy();

3、使用自定义的线程池

在默认配置信息里面是没有线程池的拒绝策略设置的方法的，如果需要更换拒绝策略就需要自定义线程池，并且如果项目当中需要多个自定义的线程池，又要如何进行管理呢？

自定义Configuration
第一步：创建一个ThreadPoolConfig 先只配置一个线程池，并设置拒绝策略为CallerRunsPolicy

@Configuration
public class ThreadPoolConfig {

    @Bean("taskExecutor")
    public Executor taskExecutor() {
        ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
        //设置线程池参数信息
        taskExecutor.setCorePoolSize(10);
        taskExecutor.setMaxPoolSize(50);
        taskExecutor.setQueueCapacity(200);
        taskExecutor.setKeepAliveSeconds(60);
        taskExecutor.setThreadNamePrefix("myExecutor--");
        taskExecutor.setWaitForTasksToCompleteOnShutdown(true);
        //线程池中任务等待时间，超过等待时间直接销毁
        taskExecutor.setAwaitTerminationSeconds(60);
        //修改拒绝策略为使用当前线程执行
        taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        //初始化线程池
        taskExecutor.initialize();
        return taskExecutor;
    }
}

4、配置多个自定义线程池

/**
 * @author shf
 * @date 2022-03-14 11:46
 */
@Configuration
public class ThreadPoolConfig {
    /**
     * 核心线程数 默认的核心线程数为1
     */
    private static final int CORE_POOL_SIZE = 2;
    /**
     * 最大线程数 默认的最大线程数是Integer.MAX_VALUE
     */
    private static final int MAX_POOL_SIZE = 20;
    /**
     * 缓冲队列数 默认的缓冲队列数是Integer.MAX_VALUE
     */
    private static final int QUEUE_CAPACITY = 50;
    /**
     * 允许线程空闲时间 默认的线程空闲时间为60秒
     */
    private static final int KEEP_ALIVE_SECONDS = 30;

    @Bean("destroyResource")
    public AsyncTaskExecutor destroyImGroupTaskExecutor() {
        return getAsyncTaskExecutor("del-resource-td-", MAX_POOL_SIZE, QUEUE_CAPACITY);
    }

    @Bean("statisticsData")
    public AsyncTaskExecutor statisticsDataExecutor() {
        return getAsyncTaskExecutor("save-data-td-", MAX_POOL_SIZE, QUEUE_CAPACITY);
    }

    @Bean("commonTaskExecutor")
    public AsyncTaskExecutor get() {
        return getAsyncTaskExecutor("common-ex-td-", MAX_POOL_SIZE, QUEUE_CAPACITY);
    }

    private AsyncTaskExecutor getAsyncTaskExecutor(String threadNamePrefix
        , int MAX_POOL_SIZE, int QUEUE_CAPACITY) {
        ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
        taskExecutor.setCorePoolSize(CORE_POOL_SIZE);
        taskExecutor.setMaxPoolSize(MAX_POOL_SIZE);
        taskExecutor.setQueueCapacity(QUEUE_CAPACITY);
        taskExecutor.setKeepAliveSeconds(KEEP_ALIVE_SECONDS);
        taskExecutor.setThreadNamePrefix(threadNamePrefix);
        taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        return taskExecutor;
    }
}

使用：

@Autowired
ThreadPoolTaskExecutor commonTaskExecutor; //会去匹配 @Bean("commonTaskExecutor") 这个线程池
如果是使用的@Async注解，只需要在注解里面指定bean的名称就可以切换到对应的线程池去了。如下所示：

@Async("commonTaskExecutor")
public void hello(String name){
    logger.info("异步线程启动 started."+name);
}

5、使用异步注解@Async注意事项
总结@Async可能失效的原因
1.@SpringBootApplication启动类当中没有添加@EnableAsync注解。
2.异步方法使用注解@Async的返回值只能为void或者Future。
3.没有走Spring的代理类。因为@Transactional和@Async注解的实现都是基于Spring的AOP，而AOP的实现是基于动态代理模式实现的。那么注解失效的原因就很明显了，有可能因为调用方法的是对象本身而不是代理对象，因为没有经过Spring容器管理。
解决方法:
这里具体说一下第三种情况的解决方法。
1、注解的方法必须是public方法。
2、注解的方法不要定义为static
3、方法一定要从另一个类中调用，也就是从类的外部调用，类的内部调用是无效的。
4、如果需要从类的内部调用，需要先获取其代理类。

三、自定义拒绝策略

ThreadPoolExecutor内部有实现4个拒绝策略，默认为AbortPolicy策略

CallerRunsPolicy：由调用execute方法提交任务的线程来执行这个任务
AbortPolicy：抛出异常RejectedExecutionException拒绝提交任务
DiscardPolicy：直接抛弃任务，不做任何处理
DiscardOldestPolicy：去除任务队列中的第一个任务，重新提交
线程池中，有三个重要的参数，决定影响了拒绝策略：corePoolSize – 核心线程数，也即最小的线程数。workQueue – 阻塞队列 。 maximumPoolSize – 最大线程数

当提交任务数大于 corePoolSize 的时候，会优先将任务放到 workQueue 阻塞队列中。当阻塞队列饱和后，会扩充线程池中线程数，直到达到 maximumPoolSize 最大线程数配置。此时，再多余的任务，则会触发线程池的拒绝策略了。

总结起来，也就是一句话，当提交的任务数大于（workQueue.size() + maximumPoolSize ），就会触发线程池的拒绝策略。
拒绝策略的源码
CallerRunsPolicy

    /**
     * A handler for rejected tasks that runs the rejected task
     * directly in the calling thread of the {@code execute} method,
     * unless the executor has been shut down, in which case the task
     * is discarded.
     * 用于拒绝任务的处理程序，
     * 可以直接在{@code execute}方法的调用线程中运行被拒绝的任务
     * 除非执行器已被关闭，否则将丢弃该任务。
     */
    public static class CallerRunsPolicy implements RejectedExecutionHandler {
        /**
         * Creates a {@code CallerRunsPolicy}.
         * 创建一个{@code CallerRunsPolicy}。
         */
        public CallerRunsPolicy() { }

        /**
         * Executes task r in the caller's thread, unless the executor
         * has been shut down, in which case the task is discarded.
         * 除非执行器已关闭，否则在调用者线程中执行任务，
         * r 在这种情况下，该任务将被丢弃。
         *
         * @param r the runnable task requested to be executed
         *          r 请求执行的可运行任务
         * @param e the executor attempting to execute this task
         *          e 尝试执行此任务的执行者
         */
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            if (!e.isShutdown()) {
                r.run();
            }
        }
    }

分析：
CallerRunsPolicy：线程调用运行该任务的 execute 本身。此策略提供简单的反馈控制机制，能够减缓新任务的提交速度。

这个策略显然不想放弃执行任务。但是由于池中已经没有任何资源了，那么就直接使用调用该execute的线程本身来执行。（开始我总不想丢弃任务的执行，但是对某些应用场景来讲，很有可能造成当前线程也被阻塞。如果所有线程都是不能执行的，很可能导致程序没法继续跑了。需要视业务情景而定吧。）

这样生产者虽然没有被阻塞，但提交任务也会被暂停。

但这种策略也有隐患，当生产者较少时，生产者消费任务的时间里，消费者可能已经把任务都消费完了，队列处于空状态，当生产者执行完任务后才能再继续生产任务，这个过程中可能导致消费者线程的饥饿。

AbortPolicy

    /**
     * A handler for rejected tasks that throws a
     * {@code RejectedExecutionException}.
     * 抛出{@code RejectedExecutionException}的拒绝任务处理程序。
     */
    public static class AbortPolicy implements RejectedExecutionHandler {
        /**
         * Creates an {@code AbortPolicy}.
         */
        public AbortPolicy() { }

        /**
         * Always throws RejectedExecutionException.
         * 总是抛出RejectedExecutionException
         * @param r the runnable task requested to be executed
         * @param e the executor attempting to execute this task
         * @throws RejectedExecutionException always
         */
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            throw new RejectedExecutionException("Task " + r.toString() +
                                                 " rejected from " +
                                                 e.toString());
        }
    }

分析：
该策略是默认饱和策略。

使用该策略时在饱和时会抛出RejectedExecutionException（继承自RuntimeException），调用者可捕获该异常自行处理。
DiscardPolicy

    /**
     * A handler for rejected tasks that silently discards the
     * rejected task.
     * 拒绝任务的处理程序，默认丢弃拒绝任务。
     */
    public static class DiscardPolicy implements RejectedExecutionHandler {
        /**
         * Creates a {@code DiscardPolicy}.
         */
        public DiscardPolicy() { }

        /**
         * Does nothing, which has the effect of discarding task r.
         * 不执行任何操作，这具有丢弃任务 r 的作用。
         * @param r the runnable task requested to be executed
         * @param e the executor attempting to execute this task
         */
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        }
    }

分析：
如代码所示，不做任何处理直接抛弃任务

DiscardOldestPolicy

    /**
     * A handler for rejected tasks that discards the oldest unhandled
     * request and then retries {@code execute}, unless the executor
     * is shut down, in which case the task is discarded.
     * 处理被拒绝任务的处理程序，它丢弃最旧的未处理请求，
     * 然后重试{@code execute}，
     * 除非执行器*被关闭，在这种情况下，该任务将被丢弃。
     */
    public static class DiscardOldestPolicy implements RejectedExecutionHandler {
        /**
         * Creates a {@code DiscardOldestPolicy} for the given executor.
         */
        public DiscardOldestPolicy() { }

        /**
         * Obtains and ignores the next task that the executor
         * would otherwise execute, if one is immediately available,
         * and then retries execution of task r, unless the executor
         * is shut down, in which case task r is instead discarded.
         * 获取并忽略执行者*会立即执行的下一个任务（如果一个任务立即可用），
         * 然后重试任务r的执行，除非执行者*被关闭，在这种情况下，任务r会被丢弃。
         * @param r the runnable task requested to be executed
         * @param e the executor attempting to execute this task
         */
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            if (!e.isShutdown()) {
                e.getQueue().poll();
                e.execute(r);
            }
        }
    }

分析：
如代码，先将阻塞队列中的头元素出队抛弃，再尝试提交任务。如果此时阻塞队列使用PriorityBlockingQueue优先级队列，将会导致优先级最高的任务被抛弃，因此不建议将该种策略配合优先级队列使用。

自定义策略
看完发现默认的几个拒绝策略并不是特别的友好，那么可不可以咱们自己搞个呢？

可以发现，所有的拒绝策略都是实现了 RejectedExecutionHandler 接口

public interface RejectedExecutionHandler {

    /**
     * Method that may be invoked by a {@link ThreadPoolExecutor} when
     * {@link ThreadPoolExecutor#execute execute} cannot accept a
     * task.  This may occur when no more threads or queue slots are
     * available because their bounds would be exceeded, or upon
     * shutdown of the Executor.
     *
     * <p>In the absence of other alternatives, the method may throw
     * an unchecked {@link RejectedExecutionException}, which will be
     * propagated to the caller of {@code execute}.
     *
     * @param r the runnable task requested to be executed
     * @param executor the executor attempting to execute this task
     * @throws RejectedExecutionException if there is no remedy
     */
    void rejectedExecution(Runnable r, ThreadPoolExecutor executor);
}

这个接口只有一个 rejectedExecution 方法。

r 为待执行任务；executor 为线程池；方法可能会抛出拒绝异常。

那么咱们就可以通过实现 RejectedExecutionHandler 接口扩展

两个栗子：一
netty自己实现的线程池里面私有的一个拒绝策略。单独启动一个新的临时线程来执行任务。

    private static final class NewThreadRunsPolicy implements RejectedExecutionHandler {
        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);
            }
        }
    }

两个栗子：二
dubbo的一个例子，它直接继承的 AbortPolicy ，加强了日志输出，并且输出dump文件

public class AbortPolicyWithReport extends ThreadPoolExecutor.AbortPolicy {
    @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);
    }
}

自定义：
背景描述：为什么会有队列拒绝，就是放到线程池的任务大于队列的最大值了，就会进入一个队列拒绝任务的一些策略，包括，报错，直接丢弃等。有自带的一些策略。
但是，有很多时候，我就是不希望它把我的任务丢弃掉，我宁愿让它在那儿等着排队，排队进入线程池的队列，也不能让它把我的任务直接丢弃，这个时候，就用到了自定义拒绝策略了。
参考类似的思路，最简单的做法，我们可以直接定义一个RejectedExecutionHandler，当队列满时改为调用BlockingQueue.put来实现生产者的阻塞：

    new RejectedExecutionHandler() {
            @Override
            public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                if (!executor.isShutdown()) {
                    try {
                        executor.getQueue().put(r);
                    } catch (InterruptedException e) {
                        // should not be interrupted
                    }
                }
            }
        };

这样，我们就无需再关心Queue和Consumer的逻辑，只要把精力集中在生产者和消费者线程的实现逻辑上，只管往线程池提交任务就行了。

相比最初的设计，这种方式的代码量能减少不少，而且能避免并发环境的很多问题。当然，你也可以采用另外的手段，例如在提交时采用信号量做入口限制等，但是如果仅仅是要让生产者阻塞，那就显得复杂了。

参考：https://blog.csdn.net/Wan_Yuan/article/details/109580896

今天的文章自定义线程池详解_如何创建线程池[通俗易懂]分享到此就结束了，感谢您的阅读，如果确实帮到您，您可以动动手指转发给其他人。

版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 举报，一经查实，本站将立刻删除。
如需转载请保留出处：http://bianchenghao.cn/59342.html

赞 (0)
编程小号编程小号
0
上一篇 2023-09-18
下一篇 2023-09-18

相关推荐

发表回复

您的电子邮箱地址不会被公开。 必填项已用*标注