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

祗疼妳一个 · 2023-9-23 05:33:55

前言

多线程是每个程序员的噩梦,用得好可以提升效率很爽，用得不好就是埋汰的火葬场。
这里不深入介绍，主要是讲解一些标准用法，熟读唐诗三百首，不会作诗也会吟。
这里就介绍一下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 业务逻辑");
}
}

复制代码

出处：https://www.cnblogs.com/LoveBB/本文版权归作者和博客园共有，欢迎转载，但未经作者同意必须在文章页面给出原文链接，否则保留追究法律责任的权利。
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！

		自动登录	找回密码
密码			立即注册

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

0 个回复

快速回复

楼主热帖

标签云