前言
- 多线程是每个程序员的噩梦,用得好可以提升效率很爽,用得不好就是埋汰的火葬场。
- 这里不深入介绍,主要是讲解一些标准用法,熟读唐诗三百首,不会作诗也会吟。
- 这里就介绍一下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;
@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 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());
}
}
ThreadExceptionLogHandler 编写
点击查看代码
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
@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;
@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;
@Bean
public Executor asyncServiceExecutor() {
VisiableThreadPoolTaskExecutor executor = new VisiableThreadPoolTaskExecutor();
executor.setCorePoolSize(coreSize);
executor.setMaxPoolSize(maxSize);
executor.setQueueCapacity(queueSize);
executor.setThreadNamePrefix(threadNamePrefix);
executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
executor.setKeepAliveSeconds(keepAlive);
executor.initialize();
return executor;
}
@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;
@RestController
@RequestMapping("/executor")
public class ExecutorController {
public void asyncDemo1() {
List nameList = new ArrayList<>(Arrays.asList("张三", "李四", "王五"));
int j = 0, size = nameList.size(), batchSize = 10;
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.stream().parallel().forEach(this::asynchronousAuthorization1);
}
@Async("asyncServiceExecutor")
public void asynchronousAuthorization1(List paramList) {
paramList.forEach(System.out::println);
System.out.println("异步执行 paramList 业务逻辑");
}
@Autowired
private ExecutorService workerPool;
public void asyncDemo2() {
List nameList = new ArrayList<>(Arrays.asList("张三", "李四", "王五"));
int j = 0, size = nameList.size(), batchSize = 10;
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.stream().parallel().forEach(paramList->{
workerPool.execute(()->asynchronousAuthorization2(paramList));
});
}
public void asynchronousAuthorization2(List paramList) {
paramList.forEach(System.out::println);
System.out.println("异步执行 paramList 业务逻辑");
}
public void asyncDemo3() {
List nameList = new ArrayList<>(Arrays.asList("张三", "李四", "王五"));
int j = 0, size = nameList.size(), batchSize = 10;
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> futures = new ArrayList<>();
list.forEach(paramList->{
futures.add(CompletableFuture.supplyAsync(() -> {
asynchronousAuthorization3(paramList);
return "默认值";
}, workerPool));
try {
Thread.sleep(500);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
});
CompletableFuture[] futuresArray = futures.toArray(new CompletableFuture[0]);
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 业务逻辑");
}
}