Springboot中使用线程池的三种方式
2023/9/19 5:52:58
本文主要是介绍Springboot中使用线程池的三种方式,对大家解决编程问题具有一定的参考价值,需要的程序猿们随着小编来一起学习吧!
前言
- 多线程是每个程序员的噩梦,用得好可以提升效率很爽,用得不好就是埋汰的火葬场。
- 这里不深入介绍,主要是讲解一些标准用法,熟读唐诗三百首,不会作诗也会吟。
- 这里就介绍一下springboot中的多线程的使用,使用线程连接池去异步执行业务方法。
- 由于代码中包含详细注释,也为了保持文章的整洁性,我就不过多的做文字描述了。
VisiableThreadPoolTaskExecutor 编写
- new VisiableThreadPoolTaskExecutor() 方式创建线程池, 返回值是 Executor
点击查看代码
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 */ @Slf4j public 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 <T> Future<T> submit(Callable<T> task){ showThreadPoolInfo("submit Callable<T> task 入参方法执行"); return super.submit(task); } @Override public ListenableFuture<?> submitListenable(Runnable task){ showThreadPoolInfo("submitListenable(Runnable task) 方法执行"); return super.submitListenable(task); } @Override public <T>ListenableFuture<T> submitListenable(Callable<T> task){ showThreadPoolInfo("submitListenable(Callable<T> 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()); } }
ThreadExceptionLogHandler 编写
- 主要用于线程池出现异常时的捕获
点击查看代码
import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; /** * @author love ice * @create 2023-09-19 0:13 */ @Slf4j @Component public class ThreadExceptionLogHandler implements Thread.UncaughtExceptionHandler { @Override public void uncaughtException(Thread t, Throwable e) { log.error("[{}]线程池异常,异常信息为:{}",t.getName(),e.getMessage(),e); } }
ExecutorConfig 编写
- 核心配置类
点击查看代码
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 @EnableAsync public 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; } }); } }
ExecutorController 编写
- 演示demo,三种不同的用法, 足以涵盖大部分场景
点击查看代码
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<String> nameList = new ArrayList<>(Arrays.asList("张三", "李四", "王五")); // 把 nameList 进行切分 int j = 0, size = nameList.size(), batchSize = 10; List<List<String>> list = new ArrayList<>(); while (j < size) { List<String> 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<String> paramList) { paramList.forEach(System.out::println); System.out.println("异步执行 paramList 业务逻辑"); } //================================分隔符====================== @Autowired private ExecutorService workerPool; /** * demo2: workerPool.execute() 实现异步逻辑。适用于没有返回值的情况下 */ public void asyncDemo2() { // 假设这是从数据库查询出来的数据 List<String> nameList = new ArrayList<>(Arrays.asList("张三", "李四", "王五")); // 把 nameList 进行切分 int j = 0, size = nameList.size(), batchSize = 10; List<List<String>> list = new ArrayList<>(); while (j < size) { List<String> 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<String> paramList) { paramList.forEach(System.out::println); System.out.println("异步执行 paramList 业务逻辑"); } //================================分隔符====================== /** * demo3: futures.add() 实现异步逻辑。适用于有返回值的情况下 */ public void asyncDemo3() { // 假设这是从数据库查询出来的数据 List<String> nameList = new ArrayList<>(Arrays.asList("张三", "李四", "王五")); // 把 nameList 进行切分 int j = 0, size = nameList.size(), batchSize = 10; List<List<String>> list = new ArrayList<>(); while (j < size) { List<String> batchList = nameList.stream().skip(j).limit(Math.min(j + batchSize, size) - j).collect(Collectors.toList()); list.add(batchList); j += batchSize; } List<CompletableFuture<String>> 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<String>[] futuresArray = futures.toArray(new CompletableFuture[0]); // 通过将多个异步任务添加到futures列表中,我们可以使用CompletableFuture提供的方法来对这些异步任务进行组合、等待和处理。 // 例如使用CompletableFuture.allOf(...)等待所有任务完成,或者使用CompletableFuture.join()获取单个任务的结果等。 CompletableFuture.allOf(futures.toArray(futuresArray)).join(); // 获取每个任务的结果或处理异常 List<String> results = new ArrayList<>(); for (CompletableFuture<String> 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<String> paramList) { paramList.forEach(System.out::println); System.out.println("异步执行 paramList 业务逻辑"); } }
这篇关于Springboot中使用线程池的三种方式的文章就介绍到这儿,希望我们推荐的文章对大家有所帮助,也希望大家多多支持为之网!
- 2024-05-15鸿蒙生态设备数量超8亿台
- 2024-05-13TiDB + ES:转转业财系统亿级数据存储优化实践
- 2024-05-09“2024鸿蒙零基础快速实战-仿抖音App开发(ArkTS版)”实战课程已上线
- 2024-05-09聊聊如何通过arthas-tunnel-server来远程管理所有需要arthas监控的应用
- 2024-05-09log4j2这么配就对了
- 2024-05-09nginx修改Content-Type
- 2024-05-09Redis多数据源,看这篇就够了
- 2024-05-09Google Chrome驱动程序 124.0.6367.62(正式版本)去哪下载?
- 2024-05-09有没有大佬知道这种数据应该怎么抓取呀?
- 2024-05-09这种运行结果里的10.100000001,怎么能最快改成10.1?