spring boot:使用多个线程池实现实现任务的线程池隔离（
2021-02-07 本文已影响0人
Zal哥哥
一，为什么要使用多个线程池?
使用多个线程池，
把相同的任务放到同一个线程池中，
可以起到隔离的作用，避免有线程出错时影响到其他线程池，
例如只有一个线程池时，
有两种任务，下单，处理图片，
如果线程池被处理图片的任务占满，影响下单任务的进行
说明：刘宏缔的架构森林是一个专注架构的博客，地址：https://www.cnblogs.com/architectforest
对应的源码可以访问这里获取： https://github.com/liuhongdi/
二，演示项目的相关信息
1,项目地址:
https://github.com/liuhongdi/multithreadpool
2,项目功能说明:
创建了两个线程池，
一个负责发邮件，
另一个负责处理图片
实际演示中都是sleep
3,项目结构：如图:
image
三，java代码说明:
1,ThreadPoolConfig.java
@Configuration
@EnableAsync public class ThreadPoolConfig { //用来生成缩略图的线程池
@Bean(name = "imageThreadPool") public ThreadPoolTaskExecutor imageThreadPool() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); // 设置核心线程数,它是可以同时被执行的线程数量
executor.setCorePoolSize(2); // 设置最大线程数,缓冲队列满了之后会申请超过核心线程数的线程
executor.setMaxPoolSize(10); // 设置缓冲队列容量,在执行任务之前用于保存任务
executor.setQueueCapacity(50); // 设置线程生存时间（秒）,当超过了核心线程出之外的线程在生存时间到达之后会被销毁
executor.setKeepAliveSeconds(60); // 设置线程名称前缀
executor.setThreadNamePrefix("imagePool-"); // 设置拒绝策略
executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); // 等待所有任务结束后再关闭线程池
executor.setWaitForTasksToCompleteOnShutdown(true); //初始化
executor.initialize(); return executor;
} //用来发邮件的线程池
@Bean(name = "emailThreadPool") public ThreadPoolTaskExecutor emailThreadPool() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); // 设置核心线程数,它是可以同时被执行的线程数量
executor.setCorePoolSize(2); // 设置最大线程数,缓冲队列满了之后会申请超过核心线程数的线程
executor.setMaxPoolSize(10); // 设置缓冲队列容量,在执行任务之前用于保存任务
executor.setQueueCapacity(50); // 设置线程生存时间（秒）,当超过了核心线程出之外的线程在生存时间到达之后会被销毁
executor.setKeepAliveSeconds(60); // 设置线程名称前缀
executor.setThreadNamePrefix("emailPool-"); // 设置拒绝策略
executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); // 等待所有任务结束后再关闭线程池
executor.setWaitForTasksToCompleteOnShutdown(true); //初始化
executor.initialize(); return executor;
}
}
说明：配置要使用的线程池，按业务类型区分开,
注意命名：一个命名为:emailThreadPool
一个命名为:imageThreadPool
另外注意线程池使用了不同的前缀，使实际运行时区分
2,HomeController.java
@RequestMapping("/home")
@Controller public class HomeController {
@Resource private MailService mailService;
@Resource private ImageService imageService;
@Resource private ThreadPoolTaskExecutor imageThreadPool; //监控线程池的状态, //我们得到的数字，只是大体接近，并不是严格的准确数字
@GetMapping("/poolstatus")
@ResponseBody public String poolstatus() {
String statusStr = ""; int queueSize = imageThreadPool.getThreadPoolExecutor().getQueue().size();
statusStr +="当前排队线程数：" + queueSize; int activeCount = imageThreadPool.getThreadPoolExecutor().getActiveCount();
statusStr +="当前活动线程数：" + activeCount; long completedTaskCount = imageThreadPool.getThreadPoolExecutor().getCompletedTaskCount();
statusStr +="执行完成线程数：" + completedTaskCount; long taskCount = imageThreadPool.getThreadPoolExecutor().getTaskCount();
statusStr +="总线程数：" + taskCount; return statusStr;
} //异步发送一封注册成功的邮件
@GetMapping("/asyncmail")
@ResponseBody public String regMail() {
mailService.sendHtmlMail(); return "mail sended";
} //异步执行sleep1秒10次
@GetMapping("/asyncimage")
@ResponseBody public Map<String, Object> asyncsleep() throws ExecutionException, InterruptedException { long start = System.currentTimeMillis();
Map<String, Object> map = new HashMap<>();
List<Future<String>> futures = new ArrayList<>(); for (int i = 0; i < 50; i++) {
Future<String> future = imageService.asynctmb(i);
futures.add(future);
}
List<String> response = new ArrayList<>(); for (Future future : futures) {
String string = (String) future.get();
response.add(string);
}
map.put("data", response);
map.put("消耗时间", String.format("任务执行成功,耗时{%s}毫秒", System.currentTimeMillis() - start)); return map;
}
}
3,MailServiceImpl.java
@Service public class MailServiceImpl implements MailService {
private Logger logger= LoggerFactory.getLogger(MailServiceImpl.class);
@Resource private MailUtil mailUtil; //异步发送html格式的邮件，演示时只是sleep1秒
@Async(value="emailThreadPool")
@Override public void sendHtmlMail() {
logger.info("sendHtmlMail begin"); try {
Thread.sleep(2000); //延时1秒
} catch(InterruptedException e) {
e.printStackTrace();
}
}
}
说明：Async注解指定线程池的名字是:emailThreadPool
4,ImageServiceImpl.java
@Service public class ImageServiceImpl implements ImageService { private Logger logger= LoggerFactory.getLogger(MailServiceImpl.class); //演示处理图片，只是sleep1秒
@Async(value="imageThreadPool")
@Override public Future<String> asynctmb(int i) {
logger.info("asynctmb begin");
String start= TimeUtil.getMilliTimeNow(); try {
Thread.sleep(1000); //延时1秒
} catch(InterruptedException e) {
e.printStackTrace();
} //log.info("async function sleep end");
String end=TimeUtil.getMilliTimeNow(); return new AsyncResult<>(String.format("asynctmb方法,第 %s 个线程:开始时间:%s,结束时间:%s",i,start,end));
}
}
说明：Async注解指定线程池的名字是:imageThreadPool
四，测试效果:
1,测试一个线程:访问:
http://127.0.0.1:8080/home/asyncmail
查看控制台:
2020-08-10 14:54:35.671 INFO 2570 --- [ emailPool-1] c.m.demo.service.impl.MailServiceImpl : sendHtmlMail begin
可以看到线程的前缀是emailThreadPool的线程的前缀
2,测试多个线程：访问:
http://127.0.0.1:8080/home/asyncimage
可以看到返回信息：
...
"消耗时间":"任务执行成功,耗时{25052}毫秒"
执行时每次并发的线程数是2，一共创建了50个线程，
每个线程sleep用时1秒
所以共用时25秒，
3,查看线程池状态:访问:
http://127.0.0.1:8080/home/asyncimage
同时访问:
http://127.0.0.1:8080/home/poolstatus
可以看到返回的状态信息:
当前排队线程数：44当前活动线程数：2执行完成线程数：54总线程数：100
说明：ThreadPoolExecutor中的统计信息只是近似值，不是完全准确的数字
进阶的动态注册线程池
@Value("${service.short.names}")
private String serviceShortNames;
@Bean
public void initExecutors() {
ConfigurableApplicationContext applicationContext = (ConfigurableApplicationContext) ApplicationContextHelper.getApplicationContext();
DefaultListableBeanFactory autowireCapableBeanFactory = (DefaultListableBeanFactory) applicationContext.getAutowireCapableBeanFactory();
for (String serviceName : serviceShortNames.split(",")) {
BeanDefinitionBuilder beanDefinitionBuilder = BeanDefinitionBuilder.rootBeanDefinition(VisiableThreadPoolTaskExecutor.class);
beanDefinitionBuilder.addPropertyValue("corePoolSize", CORE_POOL_SIZE);
beanDefinitionBuilder.addPropertyValue("maxPoolSize", MAX_POOL_SIZE);
beanDefinitionBuilder.addPropertyValue("queueCapacity", QUEUE_CAPACITY);
beanDefinitionBuilder.addPropertyValue("keepAliveSeconds", KEEP_ALIVE_SECONDS);
beanDefinitionBuilder.addPropertyValue("threadNamePrefix", THREAD_NAME_PREFIX + serviceName + "-");
beanDefinitionBuilder.addPropertyValue("waitForTasksToCompleteOnShutdown", WAIT_FOR_TASKS_TO_COMPLETE_ON_SHUTDOWN);
beanDefinitionBuilder.setLazyInit(false);
String beanName = serviceName + ASYNC_EXECUTOR_SUFFIX;
beanDefinitionBuilder.setScope(ConfigurableBeanFactory.SCOPE_SINGLETON); // 默认单例
autowireCapableBeanFactory.registerBeanDefinition(beanName, beanDefinitionBuilder.getBeanDefinition());
}
}
