线程池ThreadPoolExecutor与阻塞队列BlockingQueue应用

1.线程池介绍

JDK5.0以上： java.util.concurrent.ThreadPoolExecutor 

构造函数：

public ThreadPoolExecutor(
    int corePoolSize,
    int maximumPoolSize,
    long keepAliveTime,
    TimeUnit unit,
    BlockingQueue<Runnable> workQueue,
    RejectedExecutionHandler handler
);

参数介绍:

corePoolSize 核心线程数，指保留的线程池大小（不超过maximumPoolSize值时，线程池中最多有corePoolSize 个线程工作）。 

maximumPoolSize 指的是线程池的最大大小（线程池中最大有corePoolSize 个线程可运行）。 

keepAliveTime 指的是空闲线程结束的超时时间（当一个线程不工作时，过keepAliveTime 长时间将停止该线程）。 

unit 是一个枚举，表示 keepAliveTime 的单位（有NANOSECONDS, MICROSECONDS, MILLISECONDS, SECONDS, MINUTES, HOURS, DAYS，7个可选值）。 

workQueue 表示存放任务的队列（存放需要被线程池执行的线程队列）。 

handler 拒绝策略（添加任务失败后如何处理该任务）。

 

2.运行策略

 

1、线程池刚创建时，里面没有一个线程。任务队列是作为参数传进来的。不过，就算队列里面有任务，线程池也不会马上执行它们。

2、当调用 execute() 方法添加一个任务时，线程池会做如下判断：

    a. 如果正在运行的线程数量小于 corePoolSize，那么马上创建线程运行这个任务；

    b. 如果正在运行的线程数量大于或等于 corePoolSize，那么将这个任务放入队列。

    c. 如果这时候队列满了，而且正在运行的线程数量小于 maximumPoolSize，那么还是要创建线程运行这个任务；

    d. 如果队列满了，而且正在运行的线程数量大于或等于 maximumPoolSize，那么线程池会抛出异常，告诉调用者“我不能再接受任务了”。

3、当一个线程完成任务时，它会从队列中取下一个任务来执行。

4、当一个线程无事可做，超过一定的时间（keepAliveTime）时，线程池会判断，如果当前运行 的线程数大于 corePoolSize，那么这个线程就被停掉。所以线程池的所有任务完成后，它最终会收缩到 corePoolSize 的大小。

这个过程说明，并不是先加入任务就一定会先执行。假设队列大小为 4，corePoolSize为2，maximumPoolSize为6，那么当加入15个任务时，执行的顺序类似这样：首先执行任务 1、2，然后任务3~6被放入队列。这时候队列满了，任务7、8、9、10 会被马上执行，而任务 11~15 则会抛出异常。最终顺序是：1、2、7、8、9、10、3、4、5、6。当然这个过程是针对指定大小的ArrayBlockingQueue<Runnable>来说，如果是LinkedBlockingQueue<Runnable>，因为该队列无大小限制，所以不存在上述问题。

 

3.测试示例

(1)LinkedBlockingQueue<Runnable>队列使用1：

package threadQueueTest;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
/**
* 测试用例
*/
public class ThreadPoolTest implements Runnable {
    public void run() {
    　　synchronized(this) {
   　　 　　try{
  　　  　　　　System.out.println(Thread.currentThread().getName());
  　　 　　　　 Thread.sleep(3000);
  　　 　　 }catch (InterruptedException e){
   　　　　　　 e.printStackTrace();
   　　　　 }
  　　  }
    }
 
    public static void main(String[] args) {
   　　 BlockingQueue<Runnable> queue = new LinkedBlockingQueue<Runnable>();
    　　ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 6, 1, TimeUnit.DAYS, queue);
    　　for (int i = 0; i < 10; i++) {
   　　　　 executor.execute(new Thread(new ThreadPoolTest(),"TestThread".concat(""+i)));
    　　　　int threadSize = queue.size();
    　　　　System.out.println("线程队列大小为-->"+threadSize);
    　　}
   　　 executor.shutdown();
    }
}

 测试结果：

线程队列大小为-->0

线程队列大小为-->0

线程队列大小为-->1

线程队列大小为-->2

线程队列大小为-->3

线程队列大小为-->4

线程队列大小为-->5

线程队列大小为-->6

线程队列大小为-->7

线程队列大小为-->8

pool-1-thread-2

pool-1-thread-1

pool-1-thread-2

pool-1-thread-1

pool-1-thread-2

pool-1-thread-1

pool-1-thread-2

pool-1-thread-1

pool-1-thread-2

pool-1-thread-1

说明：可见，线程队列最大为8，共执行了10个线线程。因为是从线程池里运行的线程，所以虽然将线程的名称设为"TestThread".concat(""+i)，但输出后还是变成了pool-1-thread-x。

(2)LinkedBlockingQueue<Runnable>队列使用2：

package threadQueueTest;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
/**
* 测试用例
*/
public class ThreadPoolTest2 implements Runnable {
　　public void run() {
　　　　synchronized(this) {
　　　　　　try{
　　　　　　　　System.out.println("线程名称："+Thread.currentThread().getName());
　　　　　　　　Thread.sleep(3000); //休眠是为了让该线程不至于执行完毕后从线程池里释放
　　　　　　}catch (InterruptedException e){
　　　　　　　　e.printStackTrace();
　　　　　　}
　　　　}
　　}
 
public static void main(String[] args) throws InterruptedException {
　　BlockingQueue<Runnable> queue = new ArrayBlockingQueue<Runnable>(4); //固定为4的线程队列
　　ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 6, 1, TimeUnit.DAYS, queue);
　　for (int i = 0; i < 10; i++) {
　　　　executor.execute(new Thread(new ThreadPoolTest2(), "TestThread".concat(""+i)));
　　　　int threadSize = queue.size();
　　　　System.out.println("线程队列大小为-->"+threadSize);
　　}
　　executor.shutdown();
　　}
}

测试结果:

线程队列大小为-->0

线程队列大小为-->0

线程队列大小为-->1

线程队列大小为-->2

线程队列大小为-->3

线程队列大小为-->4

线程队列大小为-->4

线程队列大小为-->4

线程队列大小为-->4

线程队列大小为-->4

线程名称：pool-1-thread-2

线程名称：pool-1-thread-4

线程名称：pool-1-thread-6

线程名称：pool-1-thread-1

线程名称：pool-1-thread-3

线程名称：pool-1-thread-5

线程名称：pool-1-thread-4

线程名称：pool-1-thread-2

线程名称：pool-1-thread-6

线程名称：pool-1-thread-1

说明: 可见，总共10个线程，因为核心线程数为2，2个线程被立即运行，线程队列大小为4，所以4个线程被加入队列，最大线程数为6，还能运行6-2=4个，其10个线程的其余4个线程又立即运行了。

(3)LinkedBlockingQueue<Runnable>队列使用3(测试异常)：

如果将我们要运行的线程数10改为11，则由于最大线程数6+线程队列大小4=10<11，则根据线程池工作原则，最后一个线程将被拒绝策略拒绝,将示例二的main方法改为如下：

public static void main(String[] args) throws InterruptedException {
　　BlockingQueue<Runnable> queue = new ArrayBlockingQueue<Runnable>(4); //固定为4的线程队列
　　ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 6, 1, TimeUnit.DAYS, queue);
　　for (int i = 0; i < 11; i++) {
　　　　executor.execute(new Thread(new ThreadPoolTest2(), "TestThread".concat(""+i)));
　　　　int threadSize = queue.size();
　　　　System.out.println("线程队列大小为-->"+threadSize);
　　}
　　executor.shutdown();
}

测试结果:

线程队列大小为-->0

线程队列大小为-->0

线程队列大小为-->1

线程队列大小为-->2

线程队列大小为-->3

线程队列大小为-->4

线程队列大小为-->4

线程队列大小为-->4

线程队列大小为-->4

线程队列大小为-->4

线程名称：pool-1-thread-2

线程名称：pool-1-thread-4

线程名称：pool-1-thread-6

Exception in thread "main" java.util.concurrent.RejectedExecutionException: Task Thread[TestThread10,5,main] rejected from java.util.concurrent.ThreadPoolExecutor@21588809[Running, pool size = 6, active threads = 6, queued tasks = 4, completed tasks = 0]

at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2047)

at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:823)

at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1369)

at threadQueueTest.ThreadPoolTest2.main(ThreadPoolTest2.java:27)

线程名称：pool-1-thread-1

线程名称：pool-1-thread-3

线程名称：pool-1-thread-5

线程名称：pool-1-thread-2

线程名称：pool-1-thread-4

线程名称：pool-1-thread-6

线程名称：pool-1-thread-1

 

说明:很明显，抛RejectedExecutionException异常了，被拒绝策略拒绝了，这就说明线程超出了线程池的总容量（线程队列大小+最大线程数）。

(4)LinkedBlockingQueue<Runnable>队列使用3(测试add(E e)异常)：

将示例二的main方法改为如下:

public static void main(String[] args) throws InterruptedException {
　　BlockingQueue<Runnable> queue = new ArrayBlockingQueue<Runnable>(4); //固定为4的线程队列
　　ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 6, 1, TimeUnit.DAYS, queue);
　　for (int i = 0; i < 10; i++) {
　　　　executor.execute(new Thread(new ThreadPoolTest4(), "TestThread".concat(""+i)));
　　　　int threadSize = queue.size();
　　　　System.out.println("线程队列大小为-->"+threadSize);
　　　　if (threadSize==4){
　　　　　　queue.add(new Runnable() { //队列已满，抛异常
　　　　　　  @Override
　　　　　 　 public void run(){
　　　　　 　　　System.out.println("我是新线程，看看能不能搭个车加进去！");
 
　　　　　　  }
　　　　   });
　　　　}
　　}
　　executor.shutdown();
}

测试结果:

线程队列大小为-->0

线程队列大小为-->0

线程队列大小为-->1

线程队列大小为-->2

线程队列大小为-->3

线程队列大小为-->4

Exception in thread "main" java.lang.IllegalStateException: Queue full

at java.util.AbstractQueue.add(AbstractQueue.java:98)

at java.util.concurrent.ArrayBlockingQueue.add(ArrayBlockingQueue.java:312)

at threadQueueTest.ThreadPoolTest4.main(ThreadPoolTest4.java:33)

at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)

at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)

at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)

at java.lang.reflect.Method.invoke(Method.java:498)

at com.intellij.rt.execution.application.AppMain.main(AppMain.java:147)

线程名称：pool-1-thread-2

线程名称：pool-1-thread-1

线程名称：pool-1-thread-2

线程名称：pool-1-thread-1

线程名称：pool-1-thread-2

线程名称：pool-1-thread-1

 

(5)LinkedBlockingQueue<Runnable>队列使用3(测试 offer(E e)异常)：

将示例二的main方法改为如下:

public static void main(String[] args) throws InterruptedException {
　　BlockingQueue<Runnable> queue = new ArrayBlockingQueue<Runnable>(4); //固定为4的线程队列
　　ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 6, 1, TimeUnit.DAYS, queue);
　　for (int i = 0; i < 10; i++) {
　　　　executor.execute(new Thread(new ThreadPoolTest5(), "TestThread".concat(""+i)));
　　　　int threadSize = queue.size();
　　　　System.out.println("线程队列大小为-->"+threadSize);
　　　　if (threadSize==4){
　　　　　　final boolean flag = queue.offer(new Runnable() {
　　　　　　　　@Override
　　　　　　　　public void run(){
　　　　　　　　　　System.out.println("我是新线程，看看能不能搭个车加进去！");
　　　　　　　　}
　　　　　　});
　　　　　　System.out.println("添加新线程标志为-->"+flag);
　　　　}
　　}
　　executor.shutdown();
}

测试结果:

线程队列大小为-->0

线程队列大小为-->0

线程队列大小为-->1

线程队列大小为-->2

线程队列大小为-->3

线程队列大小为-->4

添加新线程标志为-->false

线程队列大小为-->4

添加新线程标志为-->false

线程队列大小为-->4

添加新线程标志为-->false

线程队列大小为-->4

添加新线程标志为-->false

线程队列大小为-->4

添加新线程标志为-->false

线程名称：pool-1-thread-2

线程名称：pool-1-thread-4

线程名称：pool-1-thread-6

线程名称：pool-1-thread-1

线程名称：pool-1-thread-3

线程名称：pool-1-thread-5

线程名称：pool-1-thread-2

线程名称：pool-1-thread-4

线程名称：pool-1-thread-6

线程名称：pool-1-thread-1

 

(6)LinkedBlockingQueue<Runnable>队列使用3(测试put(E e)异常)：

将示例二的main方法改为如下:

public static void main(String[] args) throws InterruptedException {
　　BlockingQueue<Runnable> queue = new ArrayBlockingQueue<Runnable>(4); //固定为4的线程队列
　　ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 6, 1, TimeUnit.DAYS, queue);
　　for (int i = 0; i < 10; i++) {
　　　　executor.execute(new Thread(new ThreadPoolTest6(), "TestThread".concat(""+i)));
　　　　int threadSize = queue.size();
　　　　System.out.println("线程队列大小为-->"+threadSize);
　　　　if (threadSize==4){
　　　　　　queue.put(new Runnable() {
　　　　　　　　@Override
　　　　　　　　public void run(){
　　　　　　　　　　System.out.println("我是新线程，看看能不能搭个车加进去！");
　　　　　　　　}
　　　　　　});
　　　　}
　　}
　　executor.shutdown();
}

测试结果:

线程队列大小为-->0

线程队列大小为-->0

线程队列大小为-->1

线程队列大小为-->2

线程队列大小为-->3

线程队列大小为-->4

线程名称：pool-1-thread-1

线程名称：pool-1-thread-2

线程名称：pool-1-thread-1

线程队列大小为-->4

线程名称：pool-1-thread-3

线程名称：pool-1-thread-2

线程队列大小为-->4

线程名称：pool-1-thread-4

线程名称：pool-1-thread-1

线程名称：pool-1-thread-3

线程队列大小为-->3

线程队列大小为-->4

线程名称：pool-1-thread-5

我是新线程，看看能不能搭个车加进去！

我是新线程，看看能不能搭个车加进去！

我是新线程，看看能不能搭个车加进去！

线程名称：pool-1-thread-2

我是新线程，看看能不能搭个车加进去！

说明:很明显，尝试了四次才加进去，前面三次尝试添加，但由于线程sleep(3000)，所以没有执行完，线程队列一直处于满的状态，直到某个线程执行完，队列有空位，新线程才加进去，没空位之前一直阻塞（即等待），我能加进去为止。

 

4.总结:

那么线程池的排除策略是什么样呢，一般按如下规律执行：

A.  如果运行的线程少于 corePoolSize，则 Executor 始终首选添加新的线程，而不进行排队。

B.  如果运行的线程等于或多于 corePoolSize，则 Executor 始终首选将请求加入队列，而不添加新的线程。

C.  如果无法将请求加入队列，则创建新的线程，除非创建此线程超出 maximumPoolSize，在这种情况下，任务将被拒绝。

 

总结：

1. 线程池可立即运行的最大线程数 即maximumPoolSize 参数。

2. 线程池能包含的最大线程数 = 可立即运行的最大线程数 + 线程队列大小 (一部分立即运行，一部分装队列里等待)

3. 核心线程数可理解为建议值，即建议使用的线程数，或者依据CPU核数

4. add，offer，put三种添加线程到队列的方法只在队列满的时候有区别，add为抛异常，offer返回boolean值，put直到添加成功为止。

5.同理remove，poll， take三种移除队列中线程的方法只在队列为空的时候有区别， remove为抛异常，poll为返回boolean值， take等待直到有线程可以被移除。

看看下面这张图就清楚了：

 至此，讲解完毕~

接下来是正确使用线程池：

配置类：

ExecutorConfig.java

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;


/**
 * @version v1.0
 * @description: 自定义线程池
 * @author: 47 on 2020/3/24 10:23
 */
@Configuration
@ConditionalOnProperty(prefix="threadPool", name="enable", havingValue="true")
public class ExecutorConfig {

    private final static Logger logger = LoggerFactory.getLogger(ExecutorConfig.class);

    @Value("${threadPool.coreSize}")
    private int coreSize;
    @Value("${threadPool.maxSize}")
    private int maxSize;
    @Value("${threadPool.queueSize}")
    private int queueSize;
    @Value("${threadPool.keepAliveTime}")
    private int keepAliveTime;
    @Value("${threadPool.allowCoreThreadTimeout}")
    private boolean allowCoreThreadTimeout;
    @Value("${threadPool.rejectStrategy}")
    private String rejectStrategy;

    @Bean
    public ExecutorService threadPoolExecutor(){
        BlockingQueue<Runnable> blockingQueue = new ArrayBlockingQueue<>(queueSize);
        ThreadFactory myThreadFactory = new MyThreadFactory();
        String rejectStrategyClassName = "java.util.concurrent.ThreadPoolExecutor$" + rejectStrategy;
        ThreadPoolExecutor threadPoolExecutor = null;
        try {
            threadPoolExecutor = new ThreadPoolExecutor(coreSize, maxSize, keepAliveTime, TimeUnit.SECONDS, blockingQueue, myThreadFactory, (RejectedExecutionHandler)Class.forName(rejectStrategyClassName).newInstance());
            threadPoolExecutor.allowCoreThreadTimeOut(allowCoreThreadTimeout);
        } catch (InstantiationException | IllegalAccessException | ClassNotFoundException e) {
            logger.error(e.getMessage());
        }
        return threadPoolExecutor;
    }

    private class MyThreadFactory implements ThreadFactory{

        private AtomicInteger sequence = new AtomicInteger();

        @Override
        public Thread newThread(Runnable runnable) {
            sequence.incrementAndGet();
            Thread thread = new Thread(runnable);
            thread.setName("sckj-pool-" + sequence);
            thread.setDaemon(false);
            return thread;
        }
    }
}

yml文件：

#自定义线程池
threadPool:
  enable: true #是否开启线程池
  coreSize: 10 #核心线程数
  maxSize: 20 #最大线程数
  queueSize: 50 #最大任务队列数
  keepAliveTime: 30 #线程空闲时间（秒）
  allowCoreThreadTimeout: true  #允许核心线程释放
  #拒绝策略，可选：AbortPolicy(抛出异常)；
  #                DiscardPolicy(丢弃任务，不抛异常)；
  #                DiscardOldestPolicy(丢弃队列最前面的任务)；
  #                CallerRunsPolicy(调用当前线程池所在的主线程去执行任务)
  rejectStrategy: AbortPolicy

 你以为到这里就结束了吗？？？接下来我们还可以用AOP对自己定义的线程池做增强通知，比如打印日志或者报错提示：

首先定义一个枚举：

package com.game.enums;

/**
 * @version v1.0
 * @description:
 * @author: 47 on 2020/4/30 15:01
 */
public enum ThreadPoolType {
    /**
     * 无返回值的线程池
     */
    RUNNABLE,
    /**
     * 带返回值的线程池，返回值类型为Future
     */
    CALLABLE;

}

 接下来定义线程池注解：

package com.game.annotation;

import com.game.enums.ThreadPoolType;

import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;

/**
 * @version v1.0
 * @description: 线程池注解
 * @author: 47on 2020/3/24 10:40
 */
@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.METHOD)
public @interface CustomThreadPool {

    ThreadPoolType value() default ThreadPoolType.RUNNABLE;

}

 接着定义线程池切面：

package com.game.aop;

import com.game.annotation.CustomThreadPool;
import com.game.enums.ThreadPoolType;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.annotation.Pointcut;
import org.aspectj.lang.reflect.MethodSignature;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.concurrent.ExecutorService;


/**
 * @version v1.0
 * @description: 线程池切面
 * @author: 47on 2020/3/24 15:40
 */
@Aspect
@Component
@ConditionalOnProperty(prefix="threadPool", name="enable", havingValue="true")
public class ExecutorAop {

    private final static Logger logger = LoggerFactory.getLogger(ExecutorAop.class);

    @Resource
    private ExecutorService threadPoolExecutor;

    @Pointcut("@annotation(com.game.annotation.CustomThreadPool)")
    public void threadPool(){}

    @Around(value = "threadPool()")
    public Object threadPool(ProceedingJoinPoint joinPoint) throws NoSuchMethodException {
        Object result = null;
        MethodSignature methodSignature = (MethodSignature) joinPoint.getSignature();
        //获取方法上CustomThreadPool注解的value值
        ThreadPoolType threadPoolType = joinPoint.getTarget().getClass()
                .getDeclaredMethod(methodSignature.getName(), methodSignature.getParameterTypes())
                .getAnnotation(CustomThreadPool.class).value();

        switch (threadPoolType){
            //无返回值的线程池
            case RUNNABLE:
                threadPoolExecutor.execute(() -> {
                    try {
                        joinPoint.proceed();
                    } catch (Throwable throwable) {
                        logger.error(throwable.getMessage());
                    }
                });
                break;
            //有返回值的线程池
            case CALLABLE:
                result = threadPoolExecutor.submit(() -> {
                    Object proceed = null;
                    try {
                        proceed = joinPoint.proceed();
                    } catch (Throwable throwable) {
                        logger.error("ExecutorAop : {}", throwable);
                    }
                    return proceed;
                });
                break;
            default:
        }
        return result;
    }

}

 最后是怎么使用,随便写个方法加上注解即可：

@CustomThreadPool
public void test(String testStr) {
    你的业务、、、、
}