线程池创建使用

Posted 2021-02-20 brant

最近项目中引入了阿里的规范插件,扫描发现之前创建线程池的方式有问题.

比如：Executors.newFixedThreadPool(poolSize) 和 Executors.newCachedThreadPool();

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

说明：Executors返回的线程池对象的弊端如下：

newFixedThreadPool和newSingleThreadExecutor:主要问题是堆积的请求处理队列可能会耗费非常大的内存，甚至OOM。

newCachedThreadPool和newScheduledThreadPool:主要问题是线程数最大数是Integer.MAX_VALUE，可能会创建数量非常多的线程，甚至OOM。

 

推荐示例1

 

 //org.apache.commons.lang3.concurrent.BasicThreadFactory

    ScheduledExecutorService executorService = new ScheduledThreadPoolExecutor(1,

        new BasicThreadFactory.Builder().namingPattern("example-schedule-pool-%d").daemon(true).build());

推荐示例2

 

ThreadFactory namedThreadFactory = new ThreadFactoryBuilder()

        .setNameFormat("demo-pool-%d").build();

 

    //Common Thread Pool

    ExecutorService pool = new ThreadPoolExecutor(5, 200,

         0L, TimeUnit.MILLISECONDS,

         new LinkedBlockingQueue<Runnable>(1024), namedThreadFactory, new ThreadPoolExecutor.AbortPolicy());

 

    pool.execute(()-> System.out.println(Thread.currentThread().getName()));

    pool.shutdown();//gracefully shutdown

      

推荐示例3

 

 

<bean id="userThreadPool"

        class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">

        <property name="corePoolSize" value="10" />

        <property name="maxPoolSize" value="100" />

        <property name="queueCapacity" value="2000" />

 

    <property name="threadFactory" value= threadFactory />

        <property name="rejectedExecutionHandler">

            <ref local="rejectedExecutionHandler" />

        </property>

    </bean>

    //in code

    userThreadPool.execute(thread);

 

 

线程池类java.util.concurrent.ThreadPoolExecutor构造方法：

 

ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue,

RejectedExecutionHandler handler)

corePoolSize： 线程池维护线程的最少数量

maximumPoolSize：线程池维护线程的最大数量

keepAliveTime： 线程池维护线程所允许的空闲时间

unit： 线程池维护线程所允许的空闲时间的单位

workQueue： 线程池所使用的缓冲队列

handler： 线程池对拒绝任务的处理策略

 

但有新任务时

如果线程池中的数量小于corePoolSize，即使线程池中的线程都处于空闲状态，也要创建新的线程来处理被添加的任务。

如果线程池中的数量等于 corePoolSize，但是缓冲队列 workQueue未满，那么任务被放入缓冲队列。

如果线程池中的数量大于corePoolSize，缓冲队列workQueue满，并且线程池中的数量小于maximumPoolSize，建新的线程来处理被添加的任务。

如果线程池中的数量大于corePoolSize，缓冲队列workQueue满，并且线程池中的数量等于maximumPoolSize，那么通过 handler所指定的策略来处理此任务。

 

处理任务的优先级为：

核心线程corePoolSize、

任务队列workQueue、

最大线程maximumPoolSize，

如果三者都满了，使用handler处理被拒绝的任务。

 

keepAliveTime: 当线程池中的线程数量大于 corePoolSize时，如果某线程空闲时间超过keepAliveTime，线程将被回收。

unit: 可选的参数为java.util.concurrent.TimeUnit中的几个静态属性：NANOSECONDS、MICROSECONDS、MILLISECONDS、SECONDS。

workQueue可使用：ArrayBlockingQueue 或者LinkedBlockingQueue

handler有四个选择：

ThreadPoolExecutor.AbortPolicy() 抛出java.util.concurrent.RejectedExecutionException异常

ThreadPoolExecutor.CallerRunsPolicy()重试添加当前的任务，他会自动重复调用execute()方法

ThreadPoolExecutor.DiscardOldestPolicy()抛弃旧的任务

ThreadPoolExecutor.DiscardPolicy()抛弃当前的任务

 

代码样例:

public class ThreadPoolExecutorBuilder {

   private int corePoolSize;//核心线程数
   private int maxPoolSize;//最大线程数
   private long keepAliveTime;//超线程最大空闲时间
   private TimeUnit timeUnit;//单位
   private BlockingQueue<Runnable> workQueue;//工作队列
   private RejectedExecutionHandler handler;//线程池对拒绝任务的处理策略
   private ThreadFactory threadFactory;//线程工厂

   //构造函数-初始化参数默认值
   public ThreadPoolExecutorBuilder(){
      int processors=Runtime.getRuntime().availableProcessors();//读取服务器CPU核心数
      this.corePoolSize=processors;
      this.maxPoolSize=processors*2;
      this.timeUnit=TimeUnit.SECONDS;
      this.keepAliveTime=0;
      this.threadFactory=new ThreadFactoryBuilder().setNameFormat("ThreadPoolExecutor-pool-%d").build();
      this.handler =new ThreadPoolExecutor.AbortPolicy();//饱和抛异常
      this.workQueue =new LinkedBlockingQueue<Runnable>(1024);
   }

   //创建线程池
   public ThreadPoolExecutor build(){
      return new ThreadPoolExecutor(corePoolSize, maxPoolSize, keepAliveTime, timeUnit, workQueue, threadFactory, handler);
   }

   //设置线程数
   public ThreadPoolExecutorBuilder setCapacity(int corePoolSize,int maxPoolSize){
      this.corePoolSize=corePoolSize;
      this.maxPoolSize=maxPoolSize;
      return this;
   }

   //设置线程最大空闲时间
   public ThreadPoolExecutorBuilder setKeepAliveTime(long keepAliveTime,TimeUnit timeUnit){
      this.keepAliveTime=keepAliveTime;
      this.timeUnit=timeUnit;
      return this;
   }
   //设置线程队列
   public ThreadPoolExecutorBuilder setWorkqueue(BlockingQueue<Runnable> workQueue){
      this.workQueue=workQueue;
      return this;
   }
   //设置队列处理策略
   public ThreadPoolExecutorBuilder setRejectedExecutionHandler(RejectedExecutionHandler handler){
      this.handler=handler;
      return this;
   }
   //设置线程工厂
   public ThreadPoolExecutorBuilder setThreadFactory(ThreadFactory threadFactory){
      this.threadFactory=threadFactory;
      return this;
   }
}