Naylor

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 的时候,将自定义策略配置进去。

相关文章: