ThreadPoolExecutor 简介

ThreadPoolExecutor 是 java.util.concurrent 包下的一个类，在jdk1.5版本引入，帮助开发人员管理线程并方便地执行并行任务。

通俗来说，ThreadPoolExecutor 的作用是生产和管理线程池的，可以通过调用其 execute 方法和 submit 方法执行多线程任务。

ThreadPoolExecutor 使用

创建执行器

ExecutorService 对象和 ThreadPoolExecutor 的关系如下图：

ExecutorServiceConfig：

package com.ramble.threadpool.config;import java.util.concurrent.*;



public class ExecutorServiceConfig {

    /**
     * 定义一个并发任务执行器服务
     */
    private static ExecutorService executorService;
    /**
     * 在类加载的时候初始化并发任务执行器
     */
    static {
        init();
    }

    /**
     * 防止类属性被篡改
     */
    private ExecutorServiceConfig() {
    }

    /**
     * 初始化并发任务执行器。核心线程数量：设置为2，初始创建的线程池大小；最大线程数量：设置为3；空闲线程存活时间：设置为3秒，当非核心线程执行完任务之后，若没有新的任务分派，存活多久后自动销毁；任务队列：设置为2，当线程池创建的线程数量达到最大线程数量后，新进来的任务会排队等候；
     * 拒绝策略：设置为直接抛异常
     * <p>
     * 以上配置需要根据：实际的业务场景、项目实际情况、实际硬件情况等各种因素综合考量
     */
    private static void init() {
        executorService = new ThreadPoolExecutor(2, 3, 3, TimeUnit.SECONDS, new LinkedBlockingDeque<Runnable>(2), new ThreadPoolExecutor.AbortPolicy());
    }

    /**
     * 获取默认并发任务执行器
     *
     * @return
     */
    public static ExecutorService getDefaultExecutor() {
        return executorService;
    }

    /**
     * 获取固定大小并发任务执行器
     *
     * @return
     */
    public static ExecutorService getFixedExecutor() {
        return Executors.newFixedThreadPool(10);
    }

    /**
     * 获取其他并发任务执行器
     * @return
     */
    public static ExecutorService getOtherExecutor() {
        //todo
        return null;
    }
}

这个类的核心目的是构造一个 ExecutorService ，供业务代码调用。当业务代码需要创建线程执行任务的时候，不用创建、管理和维护“执行者（线程）”，只需要告诉 ExecutorService 需要做什么 “事情（任务）”。
ExecutorService 对象必须是单例的，因为此对象本身是管理线程池的，如果自己都不是线程安全的，那使用起来将有可能发生灾难。
一个java进程允许创建多个ExecutorService 。根据业务实际情况，如果业务逻辑比较单一，大概率创建一种就满足使用。若业务繁杂，可根据业务特性抽象出多种类型以满足不同需求。譬如代码中就创建了两个 getDefaultExecutor 和 getFixedExecutor
关于线程池配置，如核心线程数量、最大线程数量、拒绝策略等等没有绝对正确的值做参考，需要根据实际情况设定

创建任务

通过实现 Runnable 接口并重写 run 方法创建无返回值的多线程任务
通过实现 Callable 接口并重写 call 方法创建有返回值的多线程任务

Task1:


package com.ramble.threadpool.task;
import lombok.extern.slf4j.Slf4j;
@Slf4j
public class Task1 implements Runnable {
    /**
     * 新线程执行一个任务，没有参数，不需要返回值，此任务和其他任务没有先后顺序
     */
    @Override
    public void run() {
        try {
            Thread.sleep(3000L);
            log.info("TaskOne,thread is ={},thread name is={}", Thread.currentThread().getId(), Thread.currentThread().getName());
            throw new IllegalAccessException("主动抛一个异常");
        } catch (IllegalAccessException e) {
            throw new RuntimeException(e);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }
}

Task2：


package com.ramble.threadpool.task;
import com.alibaba.fastjson.JSON;
import com.ramble.threadpool.dto.TaskTwoParam;
import lombok.extern.slf4j.Slf4j;
@Slf4j
public class Task2 implements Runnable {
    private TaskTwoParam param;
    public Task2(TaskTwoParam param) {
        this.param = param;
    }
    /**
     * 新线程执行一个任务，参数通过构造函数传递，不需要返回值，此任务和其他任务没有先后顺序
     */
    @Override
    public void run() {
        try {
            Thread.sleep(3000L);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        log.info("TaskTwo,thread  id is ={},thread name is={}, param is ={}", Thread.currentThread().getId(), Thread.currentThread().getName(), JSON.toJSONString(param));
    }
}

Task3：


package com.ramble.threadpool.task;
import com.alibaba.fastjson.JSON;
import com.ramble.threadpool.dto.TaskDto;
import com.ramble.threadpool.dto.TaskTwoParam;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.Callable;
@Slf4j
public class Task3 implements Callable {
    private TaskTwoParam param;
    public Task3(TaskTwoParam param) {
        this.param = param;
    }
    /**
     * 新线程执行一个任务，通过构造函数传递参数，有返回值，此任务和其他任务没有先后顺序
     *
     * @return
     * @throws Exception
     */
    @Override
    public Object call() throws Exception {
        log.info("Task3,thread  id is ={},thread name is={}, param is ={}", Thread.currentThread().getId(), Thread.currentThread().getName(), JSON.toJSONString(param));
        return new TaskDto().setName("task3-callable");
    }
}

Task4：


package com.ramble.threadpool.task;
import lombok.extern.slf4j.Slf4j;
@Slf4j
public class Task4 implements Runnable {
    /**
     * 新线程执行一个任务，没有参数，不需要返回值，此任务和其他任务没有先后顺序
     */
    @Override
    public void run() {
        try {
            Thread.sleep(3000L);
            log.info("Task4,thread is ={},thread name is={}", Thread.currentThread().getId(), Thread.currentThread().getName());
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }
}

执行任务

在 controller 中模拟发起多线程任务。


package com.ramble.threadpool.controller;
import com.alibaba.fastjson.JSON;
import com.ramble.threadpool.config.ExecutorServiceConfig;
import com.ramble.threadpool.dto.TaskDto;
import com.ramble.threadpool.dto.TaskTwoParam;
import com.ramble.threadpool.task.Task1;
import com.ramble.threadpool.task.Task2;
import com.ramble.threadpool.task.Task3;
import com.ramble.threadpool.task.Task4;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.task.TaskExecutionProperties;
import org.springframework.scheduling.annotation.Async;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import javax.annotation.Resource;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;



    @GetMapping("/task")
    public String testTask() {
        log.info("testTask,thread is ={},thread name is={}", Thread.currentThread().getId(), Thread.currentThread().getName());
//        for (int i = 0; i < 4; i++) {
//            ExecutorServiceConfig.getExecutor().execute(new Task1());
//        }
//
//        for (int i = 0; i < 5; i++) {
//            ExecutorServiceConfig.getExecutor().execute(new Task2(new TaskTwoParam().setId(100).setName("cnaylor")));
//        }
        ExecutorServiceConfig.getDefaultExecutor().execute(new Task1());
        ExecutorServiceConfig.getDefaultExecutor().execute(new Task2(new TaskTwoParam().setId(100).setName("cnaylor")));
        ExecutorServiceConfig.getFixedExecutor().execute(new Task4());
        Future<TaskDto> taskResult = ExecutorServiceConfig.getDefaultExecutor().submit(new Task3(new TaskTwoParam().setId(100).setName("cnaylor")));
        TaskDto taskDto;
        try {
            taskDto = taskResult.get();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } catch (ExecutionException e) {
            throw new RuntimeException(e);
        }
        return "testTask";
    }

ThreadPoolExecutor 拒绝策略

所谓拒绝策略，一般是指当前线程池处于满负荷状态，所有的线程都有正在处理的任务，阻塞队列也排满了的情况下，对新进来的任务做出何种响应。

默认的拒绝策略有四种，同时也可以自定义拒绝策略。

AbortPolicy
CallerRunsPolicy
DiscardOldestPolicy
DiscardPolicy

AbortPolicy

抛一个异常（RejectedExecutionException），此任务将无法执行。

CallerRunsPolicy

线程池满载了，此任务将由调用发起线程来执行。比如我们在一个http请求线程中调用了线程池处理异步任务，但是现在线程池满了，那么此任务将转由http请求线程处理。缺点是会导致http请求线程阻塞，达不到异步处理的效果。优点是任务会正常执行，避免被任务执行器丢弃。

DiscardOldestPolicy

在阻塞队列最前端抛弃一个任务，然后将此任务添加到阻塞队列中排队。最前端是指最先添加到阻塞队列的任务。

DiscardPolicy

当前任务不会执行，也不会抛异常，好像什么也没有发生一样。

自定义拒绝策略

自己编写一个类，实现 RejectedExecutionHandler 接口，并重写 rejectedExecution 方法即可实现自定义拒绝策略。需要在实例化 ThreadPoolExecutor 的时候，将自定义策略配置进去。

SpringBoot线程池ThreadPoolExecutor极简教程

ThreadPoolExecutor 简介

ThreadPoolExecutor 使用

创建执行器

创建任务

执行任务

ThreadPoolExecutor 拒绝策略

AbortPolicy

CallerRunsPolicy

DiscardOldestPolicy

DiscardPolicy

自定义拒绝策略

微服务

热门文章

最新文章

相关课程

相关电子书

相关实验场景