@Async异步任务与线程池
愿天堂没有BUG
共 48000字，需浏览 96分钟
·
2021-06-28 02:39
写在前面：本篇文章是关于使用@Async进行异步任务，并且关于线程池做了一个初步的梳理和总结，包括遇到过的一些坑
在工作中用到的一些线程池
以下代码已做脱敏处理
1.newCachedThreadPool
private void startTask(List<String> usersList){
ExecutorService executor = Executors.newCachedThreadPool();
executor.submit(()->{
//do someting
});
}
复制代码2.newScheduledThreadPool
@Configuration
public class ScheduleConfig implements SchedulingConfigurer {
@Override
public void configureTasks(ScheduledTaskRegistrar taskRegistrar) {
//当然了，这里设置的线程池是corePoolSize也是很关键了，自己根据业务需求设定
taskRegistrar.setScheduler(Executors.newScheduledThreadPool(10));
}
}
复制代码如果在idea中安装了阿里规范插件，就会发现上面两种创建线程池的方式都会报红。原因是：
线程池不允许使用Executors去创建，而是通过ThreadPoolExecutor的方式，这样的处理方式让写的同学更加明确线程池的运行规则，规避资源耗尽的风险。说明：Executors返回的线程池对象的弊端如下：
FixedThreadPool和SingleThreadPool:
允许的请求队列长度为Integer.MAX_VALUE，可能会堆积大量的请求，从而导致OOM。
CachedThreadPool:
允许的创建线程数量为Integer.MAX_VALUE，可能会创建大量的线程，从而导致OOM。
其实这里CachedThreadPool和newScheduledThreadPool是一样的，都是因为最大线程数被设置成了Integer.MAX_VALUE。
public ScheduledThreadPoolExecutor(int corePoolSize) {
super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
new DelayedWorkQueue());
}
复制代码 public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}
复制代码在源码中可以看的出newCachedThreadPool使用的是synchronousqueue队列，也可以看作是一个长度为1的BlockingQueue所以，再加上最大允许线程数为Integer.MAX_VALUE，就导致可能会创建大量线程导致OOM。
同理ScheduledThreadPoolExecutor使用的是DelayedWorkQueue，初始化大小为16。当队列满后就会创建新线程，就导致可能会创建大量线程导致OOM。
我们不妨实际来测试一下,以newCachedThreadPool为例，jvm参数-Xms64m -Xmx192m -Xss1024K -XX:MetaspaceSize=64m -XX:MaxMetaspaceSize=128m。
@PostMapping("/newCachedThreadPoolExample")
@ResponseBody
public void newCachedThreadPoolExample(){
ExecutorService executorService = Executors.newCachedThreadPool();
while (true){
executorService.submit(()->{
log.info("submit:"+LocalDateTime.now());
try {
Thread.sleep(1000);
}catch (InterruptedException e){
e.printStackTrace();
}
});
}
}
复制代码刚启动时的情况：
请求接口后就开始爆炸
然后就开始卡着不动了
比较尴尬的是一直没有出现报错OOM的情况，就直接卡死了。
总结
以上的线程池虽然可以在外部限制的情况下避免OOM等情况，但是还是建议尽量根据自己的业务情况自定义线程池。
使用@Async快速创建一个异步任务
1. application.yml
这里是线程池相关配置，先不详细说，同理可以在代码里面配置config。
线程池缓冲队列的选择
以上发生的问题大多数都和线程池的缓冲队列有关，选择一个符合自己业务特点的缓冲队列也十分重要。
spring:
task:
execution:
pool:
# 最大线程数
max-size: 16
# 核心线程数
core-size: 16
# 存活时间
keep-alive: 10s
# 队列大小
queue-capacity: 100
# 是否允许核心线程超时
allow-core-thread-timeout: true
# 线程名称前缀
thread-name-prefix: async-task-
复制代码2.ThreadpoolApplication
这里需要在 Application上添加 @EnableAsync注解，开启异步任务。如果是选择在代码里面写config，则需要在config文件上添加@EnableAsync注解。
@EnableAsync
@SpringBootApplication
public class ThreadpoolApplication {
public static void main(String[] args) {
SpringApplication.run(ThreadpoolApplication.class, args);
}
}
复制代码3.AsyncTask
编写一个异步任务处理类，在需要开启异步的方法上面添加@Async
@Component
@Slf4j
public class AsyncTask {
@Async
public void asyncRun() throws InterruptedException {
Thread.sleep(10);
log.info(Thread.currentThread().getName()+":处理完成");
}
}
复制代码4.AsyncService
编写一个调用异步方法的service
@Service
@Slf4j
public class AsyncService {
@Autowired
private AsyncTask asyncTask;
public void asyncSimpleExample() {
try {
log.info("service start");
asyncTask.asyncRun();
log.info("service end");
}catch (InterruptedException e){
e.printStackTrace();
}
}
}
复制代码5.AsyncController
编写一个Controller去调用AsyncService
/**
* @author kurtl
*/
@Controller
@RequestMapping("/")
public class AsyncController {
@Autowired
private AsyncService asyncService;
@PostMapping("/asyncSimpleExample")
@ResponseBody
public void asyncSimpleExample(){
asyncService.asyncSimpleExample();
}
}
复制代码最后请求这个接口
可以看到，先输出了asyncSimpleExample里面打印的service start与service end，表示service方法先执行完毕了，而异步方法则在调用后进行了一个sleep，service没有同步等待sleep完成，而是直接返回，表示这个是异步任务。至此我们已经通过@Async成功创建的异步任务。
关于@Async和@EnableAsync的原理
个人觉得源码中很重要的一部分就是源码中的注释，阅读注释也可以帮你快速了解源码的作用等，所有我会把重要的注释稍微翻译一下
1.@Async源码
@Target({ElementType.TYPE, ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface Async {
/**
* A qualifier value for the specified asynchronous operation(s).
* <p>May be used to determine the target executor to be used when executing
* the asynchronous operation(s), matching the qualifier value (or the bean
* name) of a specific {@link java.util.concurrent.Executor Executor} or
* {@link org.springframework.core.task.TaskExecutor TaskExecutor}
* bean definition.
* <p>When specified on a class-level {@code @Async} annotation, indicates that the
* given executor should be used for all methods within the class. Method-level use
* of {@code Async#value} always overrides any value set at the class level.
* @since 3.1.2
*/
/**
* 在这些注释中有三个非常重要的部分
* 1.使用@Async的方法只能返回Void 或者 Future类型
* 2.表明了@Async是通过org.springframework.core.task.TaskExecutor
* 或者java.util.concurrent.Executor来创建线程池
* 3.写了@Async的作用范围 在类上使用@Async会覆盖方法上的@Async
*/
String value() default "";
}
复制代码2.@EnableAsync源码
/**
* Enables Spring's asynchronous method execution capability, similar to functionality
* found in Spring's {@code <task:*>} XML namespace.
*
* <p>To be used together with @{@link Configuration Configuration} classes as follows,
* enabling annotation-driven async processing for an entire Spring application context:
*
* <pre class="code">
* @Configuration
* @EnableAsync
* public class AppConfig {
*
* }</pre>
* 这里表示需要联合@Configuration注解一起使用，所以@EnableAsync应该
* 添加在线程池Config或者SpringBootApplication 上
* {@code MyAsyncBean} is a user-defined type with one or more methods annotated with
* either Spring's {@code @Async} annotation, the EJB 3.1 {@code @javax.ejb.Asynchronous}
* annotation, or any custom annotation specified via the {@link #annotation} attribute.
* The aspect is added transparently for any registered bean, for instance via this
* configuration:
*
* <pre class="code">
* @Configuration
* public class AnotherAppConfig {
*
* @Bean
* public MyAsyncBean asyncBean() {
* return new MyAsyncBean();
* }
* }</pre>
*
* <p>By default, Spring will be searching for an associated thread pool definition:
* either a unique {@link org.springframework.core.task.TaskExecutor} bean in the context,
* or an {@link java.util.concurrent.Executor} bean named "taskExecutor" otherwise. If
* neither of the two is resolvable, a {@link org.springframework.core.task.SimpleAsyncTaskExecutor}
* 默认情况下spring会先搜索TaskExecutor类型的bean或者名字为
* taskExecutor的Executor类型的bean,都不存在使用
* SimpleAsyncTaskExecutor执行器但是这个SimpleAsyncTaskExecutor实际
* 上是有很大的坑的，建议是自定义一个线程池，这个后面会说
* will be used to process async method invocations. Besides, annotated methods having
*
* @author Chris Beams
* @author Juergen Hoeller
* @author Stephane Nicoll
* @author Sam Brannen
* @since 3.1
* @see Async
* @see AsyncConfigurer
* @see AsyncConfigurationSelector
*/
@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Import(AsyncConfigurationSelector.class)
public @interface EnableAsync {
/**
* Indicate the 'async' annotation type to be detected at either class
* or method level.
* <p>By default, both Spring's @{@link Async} annotation and the EJB 3.1
* {@code @javax.ejb.Asynchronous} annotation will be detected.
* <p>This attribute exists so that developers can provide their own
* custom annotation type to indicate that a method (or all methods of
* a given class) should be invoked asynchronously.
*/
Class<? extends Annotation> annotation() default Annotation.class;
/**
* Indicate whether subclass-based (CGLIB) proxies are to be created as opposed
* to standard Java interface-based proxies.
* <p><strong>Applicable only if the {@link #mode} is set to {@link AdviceMode#PROXY}</strong>.
* <p>The default is {@code false}.
* <p>Note that setting this attribute to {@code true} will affect <em>all</em>
* Spring-managed beans requiring proxying, not just those marked with {@code @Async}.
* For example, other beans marked with Spring's {@code @Transactional} annotation
* will be upgraded to subclass proxying at the same time. This approach has no
* negative impact in practice unless one is explicitly expecting one type of proxy
* vs. another — for example, in tests.
*
* 这个字段用来表示，是否要创建基于CGLIB的代理，实际上在高版本
* 的spring 上（大概3.x)是自动选择使用jdk动态代理还是CGLIB.
* 设置为true时，其它spring管理的bean也会升级到CGLIB代理
*/
boolean proxyTargetClass() default false;
/**
* Indicate how async advice should be applied.
* <p><b>The default is {@link AdviceMode#PROXY}.</b>
* Please note that proxy mode allows for interception of calls through the proxy
* only. Local calls within the same class cannot get intercepted that way; an
* {@link Async} annotation on such a method within a local call will be ignored
* since Spring's interceptor does not even kick in for such a runtime scenario.
* For a more advanced mode of interception, consider switching this to
* {@link AdviceMode#ASPECTJ}.
* 这个字段用来标识异步通知的模式，默认PROXY，当这个字段为
* PROXY的时候，在同一个类中，非异步方法调用异步方法，会导致异
* 步不生效，相反如果，想实现同一个类非异步方法调用异步方法就应
* 该设置为ASPECTJ
*/
AdviceMode mode() default AdviceMode.PROXY;
/**
* Indicate the order in which the {@link AsyncAnnotationBeanPostProcessor}
* should be applied.
* <p>The default is {@link Ordered#LOWEST_PRECEDENCE} in order to run
* after all other post-processors, so that it can add an advisor to
* existing proxies rather than double-proxy.
* 标明异步注解bean处理器应该遵循的执行顺序，默认最低的优先级
*（Integer.MAX_VALUE，值越小优先级越高）
*/
int order() default Ordered.LOWEST_PRECEDENCE;
}
复制代码在上面的源码中，其实最核心的代码只有一句，@Import(AsyncConfigurationSelector.class)，引入了相关的配置。
/**
* Selects which implementation of {@link AbstractAsyncConfiguration} should
* be used based on the value of {@link EnableAsync#mode} on the importing
* {@code @Configuration} class.
*
* @author Chris Beams
* @author Juergen Hoeller
* @since 3.1
* @see EnableAsync
* @see ProxyAsyncConfiguration
*/
public class AsyncConfigurationSelector extends AdviceModeImportSelector<EnableAsync> {
private static final String ASYNC_EXECUTION_ASPECT_CONFIGURATION_CLASS_NAME =
"org.springframework.scheduling.aspectj.AspectJAsyncConfiguration";
/**
* Returns {@link ProxyAsyncConfiguration} or {@code AspectJAsyncConfiguration}
* for {@code PROXY} and {@code ASPECTJ} values of {@link EnableAsync#mode()},
* respectively.
*/
/**
* 这整个方法其实就是一个选择器和ImportSelector接口的selectImports()方法很像，基于不同的代理模式，加载不同的配置类
*/
@Override
@Nullable
public String[] selectImports(AdviceMode adviceMode) {
switch (adviceMode) {
case PROXY:
return new String[] {ProxyAsyncConfiguration.class.getName()};
case ASPECTJ:
return new String[] {ASYNC_EXECUTION_ASPECT_CONFIGURATION_CLASS_NAME};
default:
return null;
}
}
}
复制代码接下来我们看看默认的ProxyAsyncConfiguration.class
/**
* {@code @Configuration} class that registers the Spring infrastructure beans necessary
* to enable proxy-based asynchronous method execution.
*
* @author Chris Beams
* @author Stephane Nicoll
* @author Juergen Hoeller
* @since 3.1
* @see EnableAsync
* @see AsyncConfigurationSelector
*/
@Configuration
@Role(BeanDefinition.ROLE_INFRASTRUCTURE)
//继承了AbstractAsyncConfiguration类
public class ProxyAsyncConfiguration extends AbstractAsyncConfiguration {
@Bean(name = TaskManagementConfigUtils.ASYNC_ANNOTATION_PROCESSOR_BEAN_NAME)
@Role(BeanDefinition.ROLE_INFRASTRUCTURE)
public AsyncAnnotationBeanPostProcessor asyncAdvisor() {
Assert.notNull(this.enableAsync, "@EnableAsync annotation metadata was not injected");
//初始化AsyncAnnotationBeanPostProcessor类型的bean
AsyncAnnotationBeanPostProcessor bpp = new AsyncAnnotationBeanPostProcessor();
//设置执行器和异常处理器
bpp.configure(this.executor, this.exceptionHandler);
Class<? extends Annotation> customAsyncAnnotation = this.enableAsync.getClass("annotation");
//设置annotation
if (customAsyncAnnotation != AnnotationUtils.getDefaultValue(EnableAsync.class, "annotation")) {
bpp.setAsyncAnnotationType(customAsyncAnnotation);
}
//设置注解属性
bpp.setProxyTargetClass(this.enableAsync.getBoolean("proxyTargetClass"));
bpp.setOrder(this.enableAsync.<Integer>getNumber("order"));
return bpp;
}
}
复制代码这一个类继承了AbstractAsyncConfiguration类，其实也就做了一件事初始化AsyncAnnotationBeanPostProcessor，@Async注解的就是通过AsyncAnnotationBeanPostProcessor这个后置处理器生成一个代理对象来实现异步的，我们先看继承的config。
/**
* Abstract base {@code Configuration} class providing common structure for enabling
* Spring's asynchronous method execution capability.
*
* @author Chris Beams
* @author Juergen Hoeller
* @author Stephane Nicoll
* @since 3.1
* @see EnableAsync
*/
@Configuration
public abstract class AbstractAsyncConfiguration implements ImportAware {
@Nullable
protected AnnotationAttributes enableAsync; //;//enableAsync的注解属性
@Nullable
protected Supplier<Executor> executor; //线程执行器
@Nullable
protected Supplier<AsyncUncaughtExceptionHandler> exceptionHandler; //异常处理器 和上面的代码对应
@Override
//设置注解属性
public void setImportMetadata(AnnotationMetadata importMetadata) {
this.enableAsync = AnnotationAttributes.fromMap(
importMetadata.getAnnotationAttributes(EnableAsync.class.getName(), false));
if (this.enableAsync == null) {
throw new IllegalArgumentException(
"@EnableAsync is not present on importing class " + importMetadata.getClassName());
}
}
/**
* Collect any {@link AsyncConfigurer} beans through autowiring.
*/
@Autowired(required = false)
//设置执行器和异常处理器
void setConfigurers(Collection<AsyncConfigurer> configurers) {
if (CollectionUtils.isEmpty(configurers)) {
return;
}
if (configurers.size() > 1) {
throw new IllegalStateException("Only one AsyncConfigurer may exist");
}
AsyncConfigurer configurer = configurers.iterator().next();
this.executor = configurer::getAsyncExecutor;
this.exceptionHandler = configurer::getAsyncUncaughtExceptionHandler;
}
}
复制代码整个代码的结构其实非常明确，我们回到上一个类，看他设置的bean AsyncAnnotationBeanPostProcessor。这个bean很复杂，所以干脆先生成类图。弄清楚baen的生命周期。AsyncAnnotationBeanPostProcessor是一个后置处理器，所以我们先找父类AbstractAdvisingBeanPostProcessor中。
/**
* Base class for {@link BeanPostProcessor} implementations that apply a
* Spring AOP {@link Advisor} to specific beans.
*
* @author Juergen Hoeller
* @since 3.2
*/
@SuppressWarnings("serial")
public abstract class AbstractAdvisingBeanPostProcessor extends ProxyProcessorSupport implements BeanPostProcessor {
@Nullable
protected Advisor advisor;
protected boolean beforeExistingAdvisors = false;
private final Map<Class<?>, Boolean> eligibleBeans = new ConcurrentHashMap<>(256);
public void setBeforeExistingAdvisors(boolean beforeExistingAdvisors) {
this.beforeExistingAdvisors = beforeExistingAdvisors;
}
@Override
public Object postProcessBeforeInitialization(Object bean, String beanName) {
return bean;
}
@Override
public Object postProcessAfterInitialization(Object bean, String beanName) {
// 没有通知，或者是AopInfrastructureBean，那么不进行代理
if (this.advisor == null || bean instanceof AopInfrastructureBean) {
// Ignore AOP infrastructure such as scoped proxies.
return bean;
}
// 添加advisor
if (bean instanceof Advised) {
Advised advised = (Advised) bean;
if (!advised.isFrozen() && isEligible(AopUtils.getTargetClass(bean))) {
// Add our local Advisor to the existing proxy's Advisor chain...
// 这里通过beforeExistingAdvisors决定是将通知添加到所有通知之前还是添加到所有通知之后
// 默认false 在@Async中被设置为true
if (this.beforeExistingAdvisors) {
advised.addAdvisor(0, this.advisor);
}
else {
advised.addAdvisor(this.advisor);
}
return bean;
}
}
//构造ProxyFactory代理工厂
if (isEligible(bean, beanName)) {
ProxyFactory proxyFactory = prepareProxyFactory(bean, beanName);
//添加代理的接口
if (!proxyFactory.isProxyTargetClass()) {
evaluateProxyInterfaces(bean.getClass(), proxyFactory);
}
//设置切面
proxyFactory.addAdvisor(this.advisor);
customizeProxyFactory(proxyFactory);
//返回代理类
return proxyFactory.getProxy(getProxyClassLoader());
}
// No proxy needed.
return bean;
}
//isEligible用于判断这个类或者这个类中的某个方法是否含有注解
protected boolean isEligible(Object bean, String beanName) {
return isEligible(bean.getClass());
}
}
复制代码在上面代码中可以看出来，proxyFactory.addAdvisor(this.advisor);这里持有一个AsyncAnnotationAdvisor类的对象advisor：buildAdvice()方法生成通知，buildPointcut生成切点。定位到这个类的buildPointcut方法中，看看他的切点匹配规则。
@SuppressWarnings("serial")
public class AsyncAnnotationAdvisor extends AbstractPointcutAdvisor implements BeanFactoryAware {
private Advice advice;
private Pointcut pointcut;
/**
* Create a new {@code AsyncAnnotationAdvisor} for bean-style configuration.
*/
public AsyncAnnotationAdvisor() {
this((Supplier<Executor>) null, (Supplier<AsyncUncaughtExceptionHandler>) null);
}
@SuppressWarnings("unchecked")
public AsyncAnnotationAdvisor(
@Nullable Executor executor, @Nullable AsyncUncaughtExceptionHandler exceptionHandler) {
this(SingletonSupplier.ofNullable(executor), SingletonSupplier.ofNullable(exceptionHandler));
}
@SuppressWarnings("unchecked")
public AsyncAnnotationAdvisor(
@Nullable Supplier<Executor> executor, @Nullable Supplier<AsyncUncaughtExceptionHandler> exceptionHandler) {
Set<Class<? extends Annotation>> asyncAnnotationTypes = new LinkedHashSet<>(2);
asyncAnnotationTypes.add(Async.class);
try {
asyncAnnotationTypes.add((Class<? extends Annotation>)
ClassUtils.forName("javax.ejb.Asynchronous", AsyncAnnotationAdvisor.class.getClassLoader()));
}
catch (ClassNotFoundException ex) {
// If EJB 3.1 API not present, simply ignore.
}
this.advice = buildAdvice(executor, exceptionHandler);
this.pointcut = buildPointcut(asyncAnnotationTypes);
}
public void setAsyncAnnotationType(Class<? extends Annotation> asyncAnnotationType) {
Assert.notNull(asyncAnnotationType, "'asyncAnnotationType' must not be null");
Set<Class<? extends Annotation>> asyncAnnotationTypes = new HashSet<>();
asyncAnnotationTypes.add(asyncAnnotationType);
this.pointcut = buildPointcut(asyncAnnotationTypes);
}
/**
* Set the {@code BeanFactory} to be used when looking up executors by qualifier.
*/
@Override
public void setBeanFactory(BeanFactory beanFactory) {
if (this.advice instanceof BeanFactoryAware) {
((BeanFactoryAware) this.advice).setBeanFactory(beanFactory);
}
}
@Override
public Advice getAdvice() {
return this.advice;
}
@Override
public Pointcut getPointcut() {
return this.pointcut;
}
//构建通知，一个简单的拦截器
protected Advice buildAdvice(
@Nullable Supplier<Executor> executor, @Nullable Supplier<AsyncUncaughtExceptionHandler> exceptionHandler) {
AnnotationAsyncExecutionInterceptor interceptor = new AnnotationAsyncExecutionInterceptor(null);
interceptor.configure(executor, exceptionHandler);
return interceptor;
}
protected Pointcut buildPointcut(Set<Class<? extends Annotation>> asyncAnnotationTypes) {
ComposablePointcut result = null;
for (Class<? extends Annotation> asyncAnnotationType : asyncAnnotationTypes) {
// 根据cpc和mpc 匹配器进行匹配
//检查类上是否有@Async注解
Pointcut cpc = new AnnotationMatchingPointcut(asyncAnnotationType, true);
//检查方法是是否有@Async注解。
Pointcut mpc = new AnnotationMatchingPointcut(null, asyncAnnotationType, true);
if (result == null) {
result = new ComposablePointcut(cpc);
}
else {
result.union(cpc);
}
result = result.union(mpc);
}
return (result != null ? result : Pointcut.TRUE);
}
}
复制代码再找到它的通知逻辑buildAdvice，就是一个拦截器，生成AnnotationAsyncExecutionInterceptor对象，对于Interceptor，关注它的核心方法invoke就行了。它的父类AsyncExecutionInterceptor重写了AsyncExecutionInterceptor接口的invoke方法。代码如下
public class AsyncExecutionInterceptor extends AsyncExecutionAspectSupport implements MethodInterceptor, Ordered {
public AsyncExecutionInterceptor(@Nullable Executor defaultExecutor) {
super(defaultExecutor);
}
public AsyncExecutionInterceptor(@Nullable Executor defaultExecutor, AsyncUncaughtExceptionHandler exceptionHandler) {
super(defaultExecutor, exceptionHandler);
}
@Override
@Nullable
//
public Object invoke(final MethodInvocation invocation) throws Throwable {
Class<?> targetClass = (invocation.getThis() != null ? AopUtils.getTargetClass(invocation.getThis()) : null);
Method specificMethod = ClassUtils.getMostSpecificMethod(invocation.getMethod(), targetClass);
final Method userDeclaredMethod = BridgeMethodResolver.findBridgedMethod(specificMethod);
// 获取到一个线程池
AsyncTaskExecutor executor = determineAsyncExecutor(userDeclaredMethod);
if (executor == null) {
throw new IllegalStateException(
"No executor specified and no default executor set on AsyncExecutionInterceptor either");
}
// 然后将这个方法封装成一个 Callable对象传入到线程池中执行
Callable<Object> task = () -> {
try {
Object result = invocation.proceed();
if (result instanceof Future) {
//阻塞等待执行完毕得到结果
return ((Future<?>) result).get();
}
}
catch (ExecutionException ex) {
handleError(ex.getCause(), userDeclaredMethod, invocation.getArguments());
}
catch (Throwable ex) {
handleError(ex, userDeclaredMethod, invocation.getArguments());
}
return null;
};
return doSubmit(task, executor, invocation.getMethod().getReturnType());
}
@Override
@Nullable
protected String getExecutorQualifier(Method method) {
return null;
}
@Override
@Nullable
protected Executor getDefaultExecutor(@Nullable BeanFactory beanFactory) {
Executor defaultExecutor = super.getDefaultExecutor(beanFactory);
return (defaultExecutor != null ? defaultExecutor : new SimpleAsyncTaskExecutor());
}
@Override
public int getOrder() {
return Ordered.HIGHEST_PRECEDENCE;
}
}
复制代码可以看到，invoke首先是包装了一个Callable的对象，然后传入doSubmit，所以代码的核心就在doSubmit这个方法中。
@Nullable
protected Object doSubmit(Callable<Object> task, AsyncTaskExecutor executor, Class<?> returnType) {
//先判断是否存在CompletableFuture这个类，优先使用CompletableFuture执行任务
if (CompletableFuture.class.isAssignableFrom(returnType)) {
return CompletableFuture.supplyAsync(() -> {
try {
return task.call();
}
catch (Throwable ex) {
throw new CompletionException(ex);
}
}, executor);
}
else if (ListenableFuture.class.isAssignableFrom(returnType)) {
return ((AsyncListenableTaskExecutor) executor).submitListenable(task);
}
else if (Future.class.isAssignableFrom(returnType)) {
return executor.submit(task);
}
else {
executor.submit(task);
return null;
}
}
复制代码这里主要是判断不同的返回值，最终都走进了submit方法，而submit根据线程池的不同，其实现也有区别，下面是SimpleAsyncTaskExecutor的实现方式。
/**
* Template method for the actual execution of a task.
* <p>The default implementation creates a new Thread and starts it.
* @param task the Runnable to execute
* @see #setThreadFactory
* @see #createThread
* @see java.lang.Thread#start()
*/
protected void doExecute(Runnable task) {
Thread thread = (this.threadFactory != null ? this.threadFactory.newThread(task) : createThread(task));
thread.start();
}
复制代码@Async的默认线程池
1.使用@Async一定要定义线程池
在上面的源码中写的很清楚，默认情况下spring会先搜索TaskExecutor类型的bean或者名字为taskExecutor的Executor类型的bean,都不存在使 SimpleAsyncTaskExecutor执行器。但是这个SimpleAsyncTaskExecutor不是真的线程池，这个类不重用线程，每次调用都会创建一个新的线程。很有可能导致OOM。
@SuppressWarnings("serial")
public class SimpleAsyncTaskExecutor extends CustomizableThreadCreator
implements AsyncListenableTaskExecutor, Serializable {
/**
* Permit any number of concurrent invocations: that is, don't throttle concurrency.
* @see ConcurrencyThrottleSupport#UNBOUNDED_CONCURRENCY
*/
public static final int UNBOUNDED_CONCURRENCY = ConcurrencyThrottleSupport.UNBOUNDED_CONCURRENCY;
/**
* Switch concurrency 'off': that is, don't allow any concurrent invocations.
* @see ConcurrencyThrottleSupport#NO_CONCURRENCY
*/
public static final int NO_CONCURRENCY = ConcurrencyThrottleSupport.NO_CONCURRENCY;
/** Internal concurrency throttle used by this executor. */
private final ConcurrencyThrottleAdapter concurrencyThrottle = new ConcurrencyThrottleAdapter();
@Nullable
private ThreadFactory threadFactory;
@Nullable
private TaskDecorator taskDecorator;
/**
* Create a new SimpleAsyncTaskExecutor with default thread name prefix.
*/
public SimpleAsyncTaskExecutor() {
super();
}
/**
* Create a new SimpleAsyncTaskExecutor with the given thread name prefix.
* @param threadNamePrefix the prefix to use for the names of newly created threads
*/
public SimpleAsyncTaskExecutor(String threadNamePrefix) {
super(threadNamePrefix);
}
/**
* Create a new SimpleAsyncTaskExecutor with the given external thread factory.
* @param threadFactory the factory to use for creating new Threads
*/
public SimpleAsyncTaskExecutor(ThreadFactory threadFactory) {
this.threadFactory = threadFactory;
}
/**
* Specify an external factory to use for creating new Threads,
* instead of relying on the local properties of this executor.
* <p>You may specify an inner ThreadFactory bean or also a ThreadFactory reference
* obtained from JNDI (on a Java EE 6 server) or some other lookup mechanism.
* @see #setThreadNamePrefix
* @see #setThreadPriority
*/
public void setThreadFactory(@Nullable ThreadFactory threadFactory) {
this.threadFactory = threadFactory;
}
/**
* Return the external factory to use for creating new Threads, if any.
*/
@Nullable
public final ThreadFactory getThreadFactory() {
return this.threadFactory;
}
public final void setTaskDecorator(TaskDecorator taskDecorator) {
this.taskDecorator = taskDecorator;
}
//这里可以设置最大线程数，通过限流去限制线程数
public void setConcurrencyLimit(int concurrencyLimit) {
this.concurrencyThrottle.setConcurrencyLimit(concurrencyLimit);
}
/**
* Return the maximum number of parallel accesses allowed.
*/
public final int getConcurrencyLimit() {
return this.concurrencyThrottle.getConcurrencyLimit();
}
/**
* Return whether this throttle is currently active.
* @return {@code true} if the concurrency limit for this instance is active
* @see #getConcurrencyLimit()
* @see #setConcurrencyLimit
*/
public final boolean isThrottleActive() {
return this.concurrencyThrottle.isThrottleActive();
}
/**
* Executes the given task, within a concurrency throttle
* if configured (through the superclass's settings).
* @see #doExecute(Runnable)
*/
@Override
public void execute(Runnable task) {
execute(task, TIMEOUT_INDEFINITE);
}
/**
* Executes the given task, within a concurrency throttle
* if configured (through the superclass's settings).
* <p>Executes urgent tasks (with 'immediate' timeout) directly,
* bypassing the concurrency throttle (if active). All other
* tasks are subject to throttling.
* @see #TIMEOUT_IMMEDIATE
* @see #doExecute(Runnable)
*/
//
@Override
public void execute(Runnable task, long startTimeout) {
Assert.notNull(task, "Runnable must not be null");
Runnable taskToUse = (this.taskDecorator != null ? this.taskDecorator.decorate(task) : task);
if (isThrottleActive() && startTimeout > TIMEOUT_IMMEDIATE) {
this.concurrencyThrottle.beforeAccess();
doExecute(new ConcurrencyThrottlingRunnable(taskToUse));
}
else {
doExecute(taskToUse);
}
}
@Override
public Future<?> submit(Runnable task) {
FutureTask<Object> future = new FutureTask<>(task, null);
execute(future, TIMEOUT_INDEFINITE);
return future;
}
@Override
public <T> Future<T> submit(Callable<T> task) {
FutureTask<T> future = new FutureTask<>(task);
execute(future, TIMEOUT_INDEFINITE);
return future;
}
@Override
public ListenableFuture<?> submitListenable(Runnable task) {
ListenableFutureTask<Object> future = new ListenableFutureTask<>(task, null);
execute(future, TIMEOUT_INDEFINITE);
return future;
}
@Override
public <T> ListenableFuture<T> submitListenable(Callable<T> task) {
ListenableFutureTask<T> future = new ListenableFutureTask<>(task);
execute(future, TIMEOUT_INDEFINITE);
return future;
}
/**
* Template method for the actual execution of a task.
* <p>The default implementation creates a new Thread and starts it.
* @param task the Runnable to execute
* @see #setThreadFactory
* @see #createThread
* @see java.lang.Thread#start()
*/
//判断是否有工厂，没有的话调用父类创建线程
protected void doExecute(Runnable task) {
Thread thread = (this.threadFactory != null ? this.threadFactory.newThread(task) : createThread(task));
thread.start();
}
/**
* Subclass of the general ConcurrencyThrottleSupport class,
* making {@code beforeAccess()} and {@code afterAccess()}
* visible to the surrounding class.
*/
private static class ConcurrencyThrottleAdapter extends ConcurrencyThrottleSupport {
@Override
protected void beforeAccess() {
super.beforeAccess();
}
@Override
protected void afterAccess() {
super.afterAccess();
}
}
/**
* This Runnable calls {@code afterAccess()} after the
* target Runnable has finished its execution.
*/
private class ConcurrencyThrottlingRunnable implements Runnable {
private final Runnable target;
public ConcurrencyThrottlingRunnable(Runnable target) {
this.target = target;
}
@Override
public void run() {
try {
this.target.run();
}
finally {
concurrencyThrottle.afterAccess();
}
}
}
}
复制代码最主要的就是这段代码
/**
* Template method for the actual execution of a task.
* <p>The default implementation creates a new Thread and starts it.
* @param task the Runnable to execute
* @see #setThreadFactory
* @see #createThread
* @see java.lang.Thread#start()
*/
//判断是否有工厂，没有的话调用父类创建线程
protected void doExecute(Runnable task) {
Thread thread = (this.threadFactory != null ? this.threadFactory.newThread(task) : createThread(task));
thread.start();
}
复制代码这里并不是用线程池，而是直接创建新的线程，所以会大量创建线程导致OOM。其实这个类是可以通过setConcurrencyLimit设置最大线程数，通过synchronized和wati and notify去进行限流，这里不展开讲。所以结论是在使用@Async一定要设置线程池。
@Async异步失效
以下代码已做脱敏处理
在看公司代码的时候，发现这样一段代码
public UserVO saveUser(HttpServletRequest request,
String source) {
String token = RequestUtils.getToken(request);
String uid = checkUserLoginReturnUid(token);
log.info("注册登录, token : {}, uid : {}", token, uid);
//获取用户信息
User User = getLoginUser(uid);
if(User == null){
User = new User();
//获取用户信息
Map<String,String> userMap = redisTemplateMain.getUserMapByToken(token);
//保存用户
saveUser(User, userMap, source);
sendUserSystem(Integer.valueOf(userMap.get("id")));
}
//用户信息放进缓存
setAuth2Redis(User);
return setUser2Redis(User);
}
//通知用户系统,我们这边成功注册了一个用户
@Async
public void sendUserSystem(Integer userId){
Map<String,Object> map = new HashMap<>();
map.put("mainUid", userId);
map.put("source", "");
String json = HttpUtil.post(property.userRegisterSendSystem, map);
log.info("sendUserSystem userId : {}, json : {}", userId, json);
}
复制代码在之前我们看源码的时候已经知道了，由于@Async的AdviceMode默认为PROXY，所以当调用方和被调用方是在同一个类中，无法产生切面，@Async没有被Spring容器管理。所以这个方法跑了这么久一直是同步。
我们可以写一个方法去测试一下。
public void asyncInvalid() {
try {
log.info("service start");
asyncInvalidExample();
log.info("service end");
}catch (InterruptedException e){
e.printStackTrace();
}
}
@Async
public void asyncInvalidExample() throws InterruptedException{
Thread.sleep(10);
log.info(Thread.currentThread().getName()+":处理完成");
}
复制代码调用结果很明显，没有进行异步操作，而是同步。
线程池拒绝导致线程丢失
既然线程池都已一个缓冲队列来保存未被消费的任务，那么就一定存在队列被塞满，导致线程丢失的情况。我们写一段代码模拟一下。
配置文件
spring:
task:
execution:
pool:
# 最大线程数
max-size: 16
# 核心线程数
core-size: 16
# 存活时间
keep-alive: 10s
# 队列大小
queue-capacity: 100
# 是否允许核心线程超时
allow-core-thread-timeout: true
# 线程名称前缀
thread-name-prefix: async-task-
复制代码异步方法
@Async
public void asyncRefuseRun() throws InterruptedException {
Thread.sleep(5000000);
}
复制代码调用方法
public void asyncRefuseRun() {
for (int t = 0;t<2000;t++){
log.info(""+t);
try {
asyncTask.asyncRefuseRun();
}catch (InterruptedException e){
e.printStackTrace();
}
}
}
复制代码这里我循环了2000个线程，理论上来说当线程到达maxPoolSize + queueCapacity时会进行拒绝，也就是16+100。
到了116的时候果然抛出了异常java.util.concurrent.RejectedExecutionException。证明线程执行了它的拒绝策略。
要理解线程池的拒绝策略，先来看看它的接口。
public interface RejectedExecutionHandler {
void rejectedExecution(Runnable r, ThreadPoolExecutor executor);
}
复制代码当线程池出现拒绝的情况，就会来调用你设置的拒绝策略，将当前提交的任务以及线程池实例本身传递给你处理。这里建议根据自己的业务场景，去实现拒绝策略。
当然如果JDK内置的实现可以满足当前业务，可以直接用jdk实现的。
AbortPolicy（中止策略）
这个中止策略就是我们刚刚演示的，触发拒绝策略后，直接中止任务，抛出异常，这个也是ThreadPoolExecutor默认实现。
/**
* A handler for rejected tasks that throws a
* {@code RejectedExecutionException}.
*/
public static class AbortPolicy implements RejectedExecutionHandler {
/**
* Creates an {@code AbortPolicy}.
*/
public AbortPolicy() { }
/**
* Always throws RejectedExecutionException.
*
* @param r the runnable task requested to be executed
* @param e the executor attempting to execute this task
* @throws RejectedExecutionException always
*/
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
throw new RejectedExecutionException("Task " + r.toString() +
" rejected from " +
e.toString());
}
}
复制代码DiscardPolicy（丢弃策略）
/**
* A handler for rejected tasks that silently discards the
* rejected task.
*/
public static class DiscardPolicy implements RejectedExecutionHandler {
/**
* Creates a {@code DiscardPolicy}.
*/
public DiscardPolicy() { }
/**
* Does nothing, which has the effect of discarding task r.
*
* @param r the runnable task requested to be executed
* @param e the executor attempting to execute this task
*/
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
}
}
复制代码很明显，啥也不干，就是一个空实现。
DiscardOldestPolicy（弃老策略）
/**
* A handler for rejected tasks that discards the oldest unhandled
* request and then retries {@code execute}, unless the executor
* is shut down, in which case the task is discarded.
*/
public static class DiscardOldestPolicy implements RejectedExecutionHandler {
/**
* Creates a {@code DiscardOldestPolicy} for the given executor.
*/
public DiscardOldestPolicy() { }
/**
* Obtains and ignores the next task that the executor
* would otherwise execute, if one is immediately available,
* and then retries execution of task r, unless the executor
* is shut down, in which case task r is instead discarded.
*
* @param r the runnable task requested to be executed
* @param e the executor attempting to execute this task
*/
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
e.getQueue().poll();
e.execute(r);
}
}
}
复制代码如果线程池未关闭，就弹出队列头部的元素，然后尝试执行。实际上还是会丢弃任务，如果头部元素执行失败，就丢弃了。区别是优先丢弃的是老的元素。
CallerRunsPolicy（调用者运行策略）
/**
* A handler for rejected tasks that runs the rejected task
* directly in the calling thread of the {@code execute} method,
* unless the executor has been shut down, in which case the task
* is discarded.
*/
public static class CallerRunsPolicy implements RejectedExecutionHandler {
/**
* Creates a {@code CallerRunsPolicy}.
*/
public CallerRunsPolicy() { }
/**
* Executes task r in the caller's thread, unless the executor
* has been shut down, in which case the task is discarded.
*
* @param r the runnable task requested to be executed
* @param e the executor attempting to execute this task
*/
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
r.run();
}
}
}
复制代码当触发拒绝策略时，判断线程池有没有关闭，没有关闭就由提交任务的当前线程处理。但是当大量提交后就会阻塞线程，导致性能降低。
hutool中的线程池拒绝策略实现
hutool作为我们经常使用的一个工具类，也有线程池工具，我们不如来看看它是如何实现的。
/**
* 构建ThreadPoolExecutor
*
* @param builder {@link ExecutorBuilder}
* @return {@link ThreadPoolExecutor}
*/
private static ThreadPoolExecutor build(ExecutorBuilder builder) {
final int corePoolSize = builder.corePoolSize;
final int maxPoolSize = builder.maxPoolSize;
final long keepAliveTime = builder.keepAliveTime;
final BlockingQueue<Runnable> workQueue;
if (null != builder.workQueue) {
workQueue = builder.workQueue;
} else {
// corePoolSize为0则要使用SynchronousQueue避免无限阻塞
workQueue = (corePoolSize <= 0) ? new SynchronousQueue<>() : new LinkedBlockingQueue<>(DEFAULT_QUEUE_CAPACITY);
}
final ThreadFactory threadFactory = (null != builder.threadFactory) ? builder.threadFactory : Executors.defaultThreadFactory();
RejectedExecutionHandler handler = ObjectUtil.defaultIfNull(builder.handler, new ThreadPoolExecutor.AbortPolicy());
final ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(//
corePoolSize, //
maxPoolSize, //
keepAliveTime, TimeUnit.NANOSECONDS, //
workQueue, //
threadFactory, //
handler//
);
if (null != builder.allowCoreThreadTimeOut) {
threadPoolExecutor.allowCoreThreadTimeOut(builder.allowCoreThreadTimeOut);
}
return threadPoolExecutor;
}
复制代码可以很清晰的看到，会判断是否传入线程池拒绝策略，如果没有就用默认的AbortPolicy。
dubbo中的拒绝策略
public class AbortPolicyWithReport extends ThreadPoolExecutor.AbortPolicy {
protected static final Logger logger = LoggerFactory.getLogger(AbortPolicyWithReport.class);
private final String threadName;
private final URL url;
private static volatile long lastPrintTime = 0;
private static Semaphore guard = new Semaphore(1);
public AbortPolicyWithReport(String threadName, URL url) {
this.threadName = threadName;
this.url = url;
}
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
String msg = String.format("Thread pool is EXHAUSTED!" +
" Thread Name: %s, Pool Size: %d (active: %d, core: %d, max: %d, largest: %d), Task: %d (completed: %d)," +
" Executor status:(isShutdown:%s, isTerminated:%s, isTerminating:%s), in %s://%s:%d!",
threadName, e.getPoolSize(), e.getActiveCount(), e.getCorePoolSize(), e.getMaximumPoolSize(), e.getLargestPoolSize(),
e.getTaskCount(), e.getCompletedTaskCount(), e.isShutdown(), e.isTerminated(), e.isTerminating(),
url.getProtocol(), url.getIp(), url.getPort());
logger.warn(msg);
dumpJStack();
throw new RejectedExecutionException(msg);
}
private void dumpJStack() {
//省略实现
}
}
复制代码dubbo的策略实现主要就是想让开发人员知道拒绝任务的情况以及原因。它先输出了线程池的详细设置参数，以及线程池当前的状态，还有当前拒绝任务的信息。然后又输出了当前线程堆栈详情在dumpJStack中实现，最后抛出RejectedExecutionException。
Netty中的线程池拒绝策略
private static final class NewThreadRunsPolicy implements RejectedExecutionHandler {
NewThreadRunsPolicy() {
super();
}
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
try {
final Thread t = new Thread(r, "Temporary task executor");
t.start();
} catch (Throwable e) {
throw new RejectedExecutionException(
"Failed to start a new thread", e);
}
}
}
复制代码Netty的线程池拒绝策略很像CallerRunsPolicy（调用者运行策略），都是不会直接丢弃任务而是继续处理任务，不同的地方是CallerRunsPolicy（调用者运行策略）是在调用线程继续处理，而Netty是new了一个新线程去处理。
activeMq中的线程池拒绝策略
new RejectedExecutionHandler() {
@Override
public void rejectedExecution(final Runnable r, final ThreadPoolExecutor executor) {
try {
executor.getQueue().offer(r, 60, TimeUnit.SECONDS);
} catch (InterruptedException e) {
throw new RejectedExecutionException("Interrupted waiting for BrokerService.worker");
}
throw new RejectedExecutionException("Timed Out while attempting to enqueue Task.");
}
});
复制代码activeMq中的策略属于最大努力执行任务型，当触发拒绝策略时，在尝试一分钟的时间重新将任务塞进任务队列，当一分钟超时还没成功时，就抛出异常。
监控线程池
在开发中，我们线程池的运行状态，线程状态，对我们来说都非常重要。所以我们应该把线程池监控起来。我们可以通过扩展beforeExecute、afterExecute和terminated这三个方法去在执行前或执行后增加一些新的操作。用来记录线程池的情况。
| 方法 | 含义 |
|---|---|
| shutdown() | 线程池延迟关闭时（等待线程池里的任务都执行完毕），统计已执行任务、正在执行任务、未执行任务数量 |
| shutdownNow() | 任务执行之前，记录任务开始时间，startTimes这个HashMap以任务的hashCode为key，开始时间为值 |
| beforeExecute(Thread t, Runnable r) | 线程池延迟关闭时（等待线程池里的任务都执行完毕），统计已执行任务、正在执行任务、未执行任务数量 |
| afterExecute(Runnable r, Throwable t) | 任务执行之后，计算任务结束时间。统计任务耗时、初始线程数、核心线程数、正在执行的任务数量、已完成任务数量、任务总数、队列里缓存的任务数量、池中存在的最大线程数、最大允许的线程数、线程空闲时间、线程池是否关闭、线程池是否终止信息 |
package com.example.threadpool;
import lombok.extern.slf4j.Slf4j;
import java.util.Date;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
/**
* @author kurtl
*/
@Slf4j
public class ThreadPoolMonitor extends ThreadPoolExecutor {
/**
* 保存任务开始执行的时间，当任务结束时，用任务结束时间减去开始时间计算任务执行时间
*/
private final ConcurrentHashMap<String, Date> startTimes;
/**
* 线程池名称，一般以业务名称命名，方便区分
*/
private final String poolName;
/**
* 调用父类的构造方法，并初始化HashMap和线程池名称
*
* @param corePoolSize 线程池核心线程数
* @param maximumPoolSize 线程池最大线程数
* @param keepAliveTime 线程的最大空闲时间
* @param unit 空闲时间的单位
* @param workQueue 保存被提交任务的队列
* @param poolName 线程池名称
*/
public ThreadPoolMonitor(int corePoolSize, int maximumPoolSize, long keepAliveTime,
TimeUnit unit, BlockingQueue<Runnable> workQueue, String poolName) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
Executors.defaultThreadFactory(), poolName);
}
/**
* 调用父类的构造方法，并初始化HashMap和线程池名称
*
* @param corePoolSize 线程池核心线程数
* @param maximumPoolSize 线程池最大线程数
* @param keepAliveTime 线程的最大空闲时间
* @param unit 空闲时间的单位
* @param workQueue 保存被提交任务的队列
* @param threadFactory 线程工厂
* @param poolName 线程池名称
*/
public ThreadPoolMonitor(int corePoolSize, int maximumPoolSize, long keepAliveTime,
TimeUnit unit, BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory, String poolName) {
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);
this.startTimes = new ConcurrentHashMap<>();
this.poolName = poolName;
}
/**
* 线程池延迟关闭时（等待线程池里的任务都执行完毕），统计线程池情况
*/
@Override
public void shutdown() {
// 统计已执行任务、正在执行任务、未执行任务数量
log.info("{} Going to shutdown. Executed tasks: {}, Running tasks: {}, Pending tasks: {}",
this.poolName, this.getCompletedTaskCount(), this.getActiveCount(), this.getQueue().size());
super.shutdown();
}
/**
* 线程池立即关闭时，统计线程池情况
*/
@Override
public List<Runnable> shutdownNow() {
// 统计已执行任务、正在执行任务、未执行任务数量
log.info("{} Going to immediately shutdown. Executed tasks: {}, Running tasks: {}, Pending tasks: {}",
this.poolName, this.getCompletedTaskCount(), this.getActiveCount(), this.getQueue().size());
return super.shutdownNow();
}
/**
* 任务执行之前，记录任务开始时间
*/
@Override
protected void beforeExecute(Thread t, Runnable r) {
startTimes.put(String.valueOf(r.hashCode()), new Date());
}
/**
* 任务执行之后，计算任务结束时间
*/
@Override
protected void afterExecute(Runnable r, Throwable t) {
Date startDate = startTimes.remove(String.valueOf(r.hashCode()));
Date finishDate = new Date();
long diff = finishDate.getTime() - startDate.getTime();
// 统计任务耗时、初始线程数、核心线程数、正在执行的任务数量、
// 已完成任务数量、任务总数、队列里缓存的任务数量、池中存在的最大线程数、
// 最大允许的线程数、线程空闲时间、线程池是否关闭、线程池是否终止
log.info("{}-pool-monitor: " +
"Duration: {} ms, PoolSize: {}, CorePoolSize: {}, Active: {}, " +
"Completed: {}, Task: {}, Queue: {}, LargestPoolSize: {}, " +
"MaximumPoolSize: {}, KeepAliveTime: {}, isShutdown: {}, isTerminated: {}",
this.poolName,
diff, this.getPoolSize(), this.getCorePoolSize(), this.getActiveCount(),
this.getCompletedTaskCount(), this.getTaskCount(), this.getQueue().size(), this.getLargestPoolSize(),
this.getMaximumPoolSize(), this.getKeepAliveTime(TimeUnit.MILLISECONDS), this.isShutdown(), this.isTerminated());
}
/**
* 创建固定线程池，代码源于Executors.newFixedThreadPool方法，这里增加了poolName
*
* @param nThreads 线程数量
* @param poolName 线程池名称
* @return ExecutorService对象
*/
public static ExecutorService newFixedThreadPool(int nThreads, String poolName) {
return new ThreadPoolMonitor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(), poolName);
}
/**
* 创建缓存型线程池，代码源于Executors.newCachedThreadPool方法，这里增加了poolName
*
* @param poolName 线程池名称
* @return ExecutorService对象
*/
public static ExecutorService newCachedThreadPool(String poolName) {
return new ThreadPoolMonitor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<>(), poolName);
}
/**
* 生成线程池所用的线程，只是改写了线程池默认的线程工厂，传入线程池名称，便于问题追踪
*/
static class EventThreadFactory implements ThreadFactory {
private static final AtomicInteger POOL_NUMBER = new AtomicInteger(1);
private final ThreadGroup group;
private final AtomicInteger threadNumber = new AtomicInteger(1);
private final String namePrefix;
/**
* 初始化线程工厂
*
* @param poolName 线程池名称
*/
EventThreadFactory(String poolName) {
SecurityManager s = System.getSecurityManager();
group = Objects.nonNull(s) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup();
namePrefix = poolName + "-pool-" + POOL_NUMBER.getAndIncrement() + "-thread-";
}
@Override
public Thread newThread(Runnable r) {
Thread t = new Thread(group, r, namePrefix + threadNumber.getAndIncrement(), 0);
if (t.isDaemon())
t.setDaemon(false);
if (t.getPriority() != Thread.NORM_PRIORITY)
t.setPriority(Thread.NORM_PRIORITY);
return t;
}
}
}
复制代码线程池负载关注的核心问题是：基于当前线程池参数分配的资源够不够。对于这个问题，我们可以从事前和事中两个角度来看。事前，线程池定义了“活跃度”这个概念，来让用户在发生Reject异常之前能够感知线程池负载问题，线程池活跃度计算公式为：线程池活跃度 = activeCount/maximumPoolSize。这个公式代表当活跃线程数趋向于maximumPoolSize的时候，代表线程负载趋高。事中，也可以从两方面来看线程池的过载判定条件，一个是发生了Reject异常，一个是队列中有等待任务（支持定制阈值）。以上两种情况发生了都会触发告警，告警信息会通过大象推送给服务所关联的负责人。——美团技术文档
核心线程数 最大线程数 如何配置
如何合理的配置线程池参数，比较普遍的说法是。
IO密集型 = 2Ncpu（可以测试后自己控制大小，2Ncpu一般没问题）（常出现于线程中：数据库数据交互、文件上传下载、网络数据传输等等）
计算密集型 = Ncpu（常出现于线程中：复杂算法）
而这种方案没有考虑多线程池的情况，实际使用上也有偏离。
图来自美团技术博客
所以参数的设置应该根据自己实际的应用场景定制。
多线程池的使用
一般在实际业务中，我们会定义不同的线程池来处理不同的业务。利用我们之前完成的ThreadPoolMonitor可以很快的定义不同的线程。
ThreadPoolConfig
@EnableAsync
@Configuration
public class ThreadPoolConfig {
@Bean
public ThreadPoolExecutor test01(){
return new ThreadPoolMonitor(16,32,60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(100),"test01");
}
@Bean
public ThreadPoolExecutor test02(){
return new ThreadPoolMonitor(8,16,60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(100),"test02");
}
}
作者：西西弗的石头
链接：https://juejin.cn/post/6976893903223914527
来源：掘金
著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。
评论