Springboot中使用线程池的三种方式

import lombok.extern.slf4j.Slf4j;import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;import org.springframework.util.concurrent.ListenableFuture;import java.util.concurrent.Callable;import java.util.concurrent.Future;import java.util.concurrent.ThreadPoolExecutor;/** * @author love ice * @create 2023-09-19 0:17 */@Slf4jpublic class VisiableThreadPoolTaskExecutor extends ThreadPoolTaskExecutor {    @Override    public void execute(Runnable task){        showThreadPoolInfo("execute一个参数的方法执行");    }    @Override    public void execute(Runnable task, long startTimeout){        showThreadPoolInfo("execute两个参数的方法执行");    }    @Override    public Future submit(Runnable task){        showThreadPoolInfo("submit Runnable task 入参方法执行");        return super.submit(task);    }    @Override    public  Future submit(Callable task){        showThreadPoolInfo("submit Callable task 入参方法执行");        return super.submit(task);    }    @Override    public ListenableFuture submitListenable(Runnable task){        showThreadPoolInfo("submitListenable(Runnable task) 方法执行");        return super.submitListenable(task);    }    @Override    public ListenableFuture submitListenable(Callable task){        showThreadPoolInfo("submitListenable(Callable task) 方法执行");        return super.submitListenable(task);    }    private void showThreadPoolInfo(String prefix){        ThreadPoolExecutor threadPoolExecutor = getThreadPoolExecutor();        log.info("{}, {}, taskCount[{}], completedTaskCount[{}], activeCount[{}], queueSize[{}]",                this.getThreadNamePrefix(), prefix, threadPoolExecutor.getTaskCount(),                threadPoolExecutor.getCompletedTaskCount(), threadPoolExecutor.getActiveCount(),                threadPoolExecutor.getQueue().size());    }}

import lombok.extern.slf4j.Slf4j;import org.springframework.stereotype.Component;/** * @author love ice * @create 2023-09-19 0:13 */@Slf4j@Componentpublic class ThreadExceptionLogHandler implements Thread.UncaughtExceptionHandler {    @Override    public void uncaughtException(Thread t, Throwable e) {        log.error("[{}]线程池异常,异常信息为:{}",t.getName(),e.getMessage(),e);    }}

import com.test.redis.Infrastructure.handler.ThreadExceptionLogHandler;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.beans.factory.annotation.Value;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;import org.springframework.scheduling.annotation.EnableAsync;import java.util.concurrent.*;import java.util.concurrent.atomic.AtomicInteger;/** * 线程池配置 * * @author love ice * @create 2023-09-19 0:09 */@Configuration@EnableAsyncpublic class ExecutorConfig {    @Value("${thread.pool.coreSize:50}")    private int coreSize;    @Value("${thread.pool.maxSize:50}")    private int maxSize;    @Value("${thread.pool.queueSize:9999}")    private int queueSize;    @Value("${thread.pool.threadNamePrefix:thread-name}")    private String threadNamePrefix;    @Value("${thread.pool.keepAlive:60}")    private int keepAlive;    @Autowired    private ThreadExceptionLogHandler threadExceptionLogHandler;    /**     * 方式一: new VisiableThreadPoolTaskExecutor() 方式创建线程池,返回值是 Executor     * 适用于 @Async("asyncServiceExecutor") 注解     * 也可以     *  @Autowired     *  private Executor asyncServiceExecutor;     *     * @return Executor     */    @Bean    public Executor asyncServiceExecutor() {        VisiableThreadPoolTaskExecutor executor = new VisiableThreadPoolTaskExecutor();        // 配置核心线程数 50        executor.setCorePoolSize(coreSize);        // 配置最大线程数 50        executor.setMaxPoolSize(maxSize);        // 配置队列大小 9999        executor.setQueueCapacity(queueSize);        // 配置线程池中的线程名称前缀 模块-功能-作用        executor.setThreadNamePrefix(threadNamePrefix);        // rejection-policy：当pool已经达到max size的时候，如何处理新任务        // CALLER_RUNS：不在新线程中执行任务，而是有调用者所在的线程来执行        // 线程池无法接受新的任务并且队列已满时，如果有新的任务提交给线程池，而线程池已经达到了最大容量限制，那么这个任务不会被丢弃，而是由调用该任务的线程来执行。        // 这样可以避免任务被直接丢弃，并让调用者自己执行任务以减轻任务提交频率。        // 这个拒绝策略可能会导致任务提交者的线程执行任务，这可能会对调用者的性能产生一些影响，因为调用者线程需要等待任务执行完成才能继续进行其他操作。        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());        // 线程空闲后的最大存活时间 60        executor.setKeepAliveSeconds(keepAlive);        // 执行初始化        executor.initialize();        return executor;    }    /**     * 方式二: new ThreadPoolExecutor() 方式创建线程池     * 适用于:     * @Autowired     * private ExecutorService fbWorkerPool;     * @return ExecutorService     */    @Bean    public ExecutorService workerPool() {        return new ThreadPoolExecutor(coreSize, maxSize, keepAlive, TimeUnit.MILLISECONDS,                new LinkedBlockingDeque(20000),                new ThreadFactory() {                    private final AtomicInteger threadNumber = new AtomicInteger(1);                    @Override                    public Thread newThread(Runnable runnable) {                        Thread thread = new Thread(runnable, threadNamePrefix + threadNumber.getAndIncrement());                        thread.setUncaughtExceptionHandler(threadExceptionLogHandler);                        return thread;                    }                });    }}

import org.springframework.beans.factory.annotation.Autowired;import org.springframework.scheduling.annotation.Async;import org.springframework.web.bind.annotation.RequestMapping;import org.springframework.web.bind.annotation.RestController;import java.util.ArrayList;import java.util.Arrays;import java.util.List;import java.util.concurrent.CompletableFuture;import java.util.concurrent.ExecutorService;import java.util.stream.Collectors;/** * 这里是demo演示、把业务写在 controller 了,一般开发都是在 service 层实现的。 * * @author love ice * @create 2023-09-19 0:59 */@RestController@RequestMapping("/executor")public class ExecutorController {    /**     * demo1: 使用异步注解 @Async("asyncServiceExecutor") 执行方法,适用于没有返回值的情况下     */    public void asyncDemo1() {        // 假设这是从数据库查询出来的数据        List nameList = new ArrayList(Arrays.asList("张三", "李四", "王五"));        // 把 nameList 进行切分        int j = 0, size = nameList.size(), batchSize = 10;        List<List> list = new ArrayList();        while (j < size) {            List batchList = nameList.stream().skip(j).limit(Math.min(j + batchSize, size) - j).collect(Collectors.toList());            list.add(batchList);            j += batchSize;        }        // 先把 list 切分成小份数据,在使用 @Async(),异步处理数据        list.stream().parallel().forEach(this::asynchronousAuthorization1);    }    /**     * 异步注解处理业务逻辑,实际业务开发,需要提取到 Service 层,否则会报错。     *     * @param paramList 入参     */    @Async("asyncServiceExecutor")    public void asynchronousAuthorization1(List paramList) {        paramList.forEach(System.out::println);        System.out.println("异步执行 paramList 业务逻辑");    }//================================分隔符======================    @Autowired    private ExecutorService workerPool;    /**     * demo2: workerPool.execute() 实现异步逻辑。适用于没有返回值的情况下     */    public void asyncDemo2() {        // 假设这是从数据库查询出来的数据        List nameList = new ArrayList(Arrays.asList("张三", "李四", "王五"));        // 把 nameList 进行切分        int j = 0, size = nameList.size(), batchSize = 10;        List<List> list = new ArrayList();        while (j < size) {            List batchList = nameList.stream().skip(j).limit(Math.min(j + batchSize, size) - j).collect(Collectors.toList());            list.add(batchList);            j += batchSize;        }        // 将 list 切分成小份数据,workerPool.execute(),异步处理数据        list.stream().parallel().forEach(paramList->{            workerPool.execute(()->asynchronousAuthorization2(paramList));        });    }    public void asynchronousAuthorization2(List paramList) {        paramList.forEach(System.out::println);        System.out.println("异步执行 paramList 业务逻辑");    }//================================分隔符======================    /**     * demo3: futures.add() 实现异步逻辑。适用于有返回值的情况下     */    public void asyncDemo3() {        // 假设这是从数据库查询出来的数据        List nameList = new ArrayList(Arrays.asList("张三", "李四", "王五"));        // 把 nameList 进行切分        int j = 0, size = nameList.size(), batchSize = 10;        List<List> list = new ArrayList();        while (j < size) {            List batchList = nameList.stream().skip(j).limit(Math.min(j + batchSize, size) - j).collect(Collectors.toList());            list.add(batchList);            j += batchSize;        }        List<CompletableFuture> futures = new ArrayList();        // 将 list 切分成小份数据,futures.add(),异步处理数据,有返回值的情况下        list.forEach(paramList->{            // CompletableFuture.supplyAsync() 该任务会在一个新的线程中执行，并返回一个结果            // 通过futures.add(...)将这个异步任务添加到futures列表中。这样可以方便后续对多个异步任务进行管理和处理            futures.add(CompletableFuture.supplyAsync(() -> {                asynchronousAuthorization3(paramList);               return "默认值";            }, workerPool));            // 防止太快,让它休眠一下            try {                Thread.sleep(500);            } catch (InterruptedException e) {                throw new RuntimeException(e);            }        });        //new CompletableFuture[0] 创建了一个初始长度为 0 的 CompletableFuture 数组，作为目标数组。然后，futures.toArray(new CompletableFuture[0]) 将 futures 列表中的元素复制到目标数组中，并返回结果数组。        CompletableFuture[] futuresArray = futures.toArray(new CompletableFuture[0]);        // 通过将多个异步任务添加到futures列表中，我们可以使用CompletableFuture提供的方法来对这些异步任务进行组合、等待和处理。        // 例如使用CompletableFuture.allOf(...)等待所有任务完成，或者使用CompletableFuture.join()获取单个任务的结果等。        CompletableFuture.allOf(futures.toArray(futuresArray)).join();        // 获取每个任务的结果或处理异常        List results = new ArrayList();        for (CompletableFuture future :futuresArray) {            // 处理任务的异常            future.exceptionally(ex -> {                System.out.println("Task encountered an exception: " + ex.getMessage());                return "0"; // 返回默认值或者做其他补偿操作            });            // 获取任务结果            String result = future.join();            results.add(result);        }        // 所有任务已完成，可以进行下一步操作    }    public void asynchronousAuthorization3(List paramList) {        paramList.forEach(System.out::println);        System.out.println("异步执行 paramList 业务逻辑");    }}