wordpress 支付宝捐赠/seo初学教程

异常描述

存在一个业务，需要做批量的更新操作，每个操作会调用一个变更记录的接口用来记录更新操作的信息，这个接口使用了异步注解——@Async，并且没有指定线程池，使用的时默认的配置。

异常信息

java.util.concurrent.RejectedExecutionException: Task java.util.concurrent.FutureTask@2665b018[Not completed, task = java.util.concurrent.Executors$RunnableAdapter@63673210[Wrapped task = null]] rejected from java.util.concurrent.ThreadPoolExecutor@1b4a92e[Running, pool size = 10, active threads = 10, queued tasks = 0, completed tasks = 579]at java.base/java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2065)at java.base/java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:833)at java.base/java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1365)at java.base/java.util.concurrent.AbstractExecutorService.submit(AbstractExecutorService.java:123)at com.netflix.hystrix.strategy.concurrency.HystrixContextScheduler$ThreadPoolWorker.schedule(HystrixContextScheduler.java:172)at com.netflix.hystrix.strategy.concurrency.HystrixContextScheduler$HystrixContextSchedulerWorker.schedule(HystrixContextScheduler.java:106)at rx.internal.operators.OperatorSubscribeOn.call(OperatorSubscribeOn.java:50)at rx.internal.operators.OperatorSubscribeOn.call(OperatorSubscribeOn.java:30)at rx.internal.operators.OnSubscribeLift.call(OnSubscribeLift.java:48)at rx.internal.operators.OnSubscribeLift.call(OnSubscribeLift.java:30)at rx.Observable.unsafeSubscribe(Observable.java:10327)at rx.internal.operators.OnSubscribeDoOnEach.call(OnSubscribeDoOnEach.java:41)at rx.internal.operators.OnSubscribeDoOnEach.call(OnSubscribeDoOnEach.java:30)at rx.Observable.unsafeSubscribe(Observable.java:10327)at rx.internal.operators.OnSubscribeDoOnEach.call(OnSubscribeDoOnEach.java:41)at rx.internal.operators.OnSubscribeDoOnEach.call(OnSubscribeDoOnEach.java:30)at rx.internal.operators.OnSubscribeLift.call(OnSubscribeLift.java:48)at rx.internal.operators.OnSubscribeLift.call(OnSubscribeLift.java:30)at rx.Observable.unsafeSubscribe(Observable.java:10327)at rx.internal.operators.OnSubscribeDoOnEach.call(OnSubscribeDoOnEach.java:41)at rx.internal.operators.OnSubscribeDoOnEach.call(OnSubscribeDoOnEach.java:30)at rx.Observable.unsafeSubscribe(Observable.java:10327)at rx.internal.operators.OnSubscribeDoOnEach.call(OnSubscribeDoOnEach.java:41)at rx.internal.operators.OnSubscribeDoOnEach.call(OnSubscribeDoOnEach.java:30)at rx.Observable.unsafeSubscribe(Observable.java:10327)at rx.internal.operators.OnSubscribeDoOnEach.call(OnSubscribeDoOnEach.java:41)at rx.internal.operators.OnSubscribeDoOnEach.call(OnSubscribeDoOnEach.java:30)at rx.internal.operators.OnSubscribeLift.call(OnSubscribeLift.java:48)at rx.internal.operators.OnSubscribeLift.call(OnSubscribeLift.java:30)at rx.Observable.unsafeSubscribe(Observable.java:10327)at rx.internal.operators.OnSubscribeDefer.call(OnSubscribeDefer.java:51)at rx.internal.operators.OnSubscribeDefer.call(OnSubscribeDefer.java:35)at rx.Observable.unsafeSubscribe(Observable.java:10327)at rx.internal.operators.OnSubscribeMap.call(OnSubscribeMap.java:48)at rx.internal.operators.OnSubscribeMap.call(OnSubscribeMap.java:33)at rx.Observable.unsafeSubscribe(Observable.java:10327)at rx.internal.operators.OnSubscribeDoOnEach.call(OnSubscribeDoOnEach.java:41)at rx.internal.operators.OnSubscribeDoOnEach.call(OnSubscribeDoOnEach.java:30)at rx.internal.operators.OnSubscribeLift.call(OnSubscribeLift.java:48)at rx.internal.operators.OnSubscribeLift.call(OnSubscribeLift.java:30)at rx.Observable.unsafeSubscribe(Observable.java:10327)at rx.internal.operators.OnSubscribeDoOnEach.call(OnSubscribeDoOnEach.java:41)at rx.internal.operators.OnSubscribeDoOnEach.call(OnSubscribeDoOnEach.java:30)at rx.Observable.unsafeSubscribe(Observable.java:10327)at rx.internal.operators.OnSubscribeDefer.call(OnSubscribeDefer.java:51)at rx.internal.operators.OnSubscribeDefer.call(OnSubscribeDefer.java:35)at rx.internal.operators.OnSubscribeLift.call(OnSubscribeLift.java:48)at rx.internal.operators.OnSubscribeLift.call(OnSubscribeLift.java:30)at rx.Observable.subscribe(Observable.java:10423)at rx.Observable.subscribe(Observable.java:10390)at rx.internal.operators.BlockingOperatorToFuture.toFuture(BlockingOperatorToFuture.java:51)at rx.observables.BlockingObservable.toFuture(BlockingObservable.java:410)at com.netflix.hystrix.HystrixCommand.queue(HystrixCommand.java:378)at com.netflix.hystrix.HystrixCommand.execute(HystrixCommand.java:344)at feign.hystrix.HystrixInvocationHandler.invoke(HystrixInvocationHandler.java:170)at jdk.proxy2/jdk.proxy2.$Proxy123.action(Unknown Source)at com.tct.relation.service.dataupdate.impl.DataUpdateNoteServiceImpl.qrySomeNameById(DataUpdateNoteServiceImpl.java:631)at com.tct.relation.service.dataupdate.impl.DataUpdateNoteServiceImpl.pubSetFieldContents(DataUpdateNoteServiceImpl.java:198)at com.tct.relation.service.dataupdate.impl.DataUpdateNoteServiceImpl.addDataUpdateNote(DataUpdateNoteServiceImpl.java:90)at com.tct.relation.service.dataupdate.impl.DataUpdateNoteServiceImpl$$FastClassBySpringCGLIB$$ed28d9c8.invoke(<generated>)at org.springframework.cglib.proxy.MethodProxy.invoke(MethodProxy.java:218)at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.invokeJoinpoint(CglibAopProxy.java:771)at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163)at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.proceed(CglibAopProxy.java:749)at org.springframework.aop.interceptor.AsyncExecutionInterceptor.lambda$invoke$0(AsyncExecutionInterceptor.java:115)at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)at java.base/java.lang.Thread.run(Thread.java:833)

错误原因

SpringBoot 默认@Async的默认线程池大小时8，并且超过时的处理策略是AbortPolicy，该策略会抛出RejectExecutionException异常，导致程序报错会影响执行结果。

解决方法

配置自定义线程池，当任务超出线程池大小后进入队列等待，而不是报错。具体的配置线程池方法有很多，下面贴出我的配置。

package com.tct.relation.config;import lombok.extern.slf4j.Slf4j;
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 org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;import java.util.concurrent.ThreadPoolExecutor;/*** @author Huhailong* @Description 线程池配置* @Date 2021/10/9.*/
@Slf4j
@Configuration
@EnableAsync
public class ThreadPoolTaskConfig {/**   默认情况下，在创建了线程池后，线程池中的线程数为0，当有任务来之后，就会创建一个线程去执行任务，*  当线程池中的线程数目达到corePoolSize后，就会把到达的任务放到缓存队列当中；*  当队列满了，就继续创建线程，当线程数量大于等于maxPoolSize后，开始使用拒绝策略拒绝*//*** 获得Java虚拟机可用的处理器个数 + 1*/private static final int THREADS = Runtime.getRuntime().availableProcessors() + 1;@Value("${async.executor.thread.core_pool_size:0}")public static int corePoolSizeConfig;// 核心线程池大小public static int corePoolSize = corePoolSizeConfig ==0 ? THREADS : corePoolSizeConfig;// 最大可创建的线程数//@Value("${async.executor.thread.max_pool_size}")private final int maxPoolSize = 2 * THREADS;//线程池名前缀//@Value("${async.executor.thread.threadNamePrefix}")private static final String threadNamePrefix = "Async-Service-";@Bean(name = "threadPoolTaskExecutor")public ThreadPoolTaskExecutor threadPoolTaskExecutor(){log.info("core thread count : {}",corePoolSize);log.info("max thread pool count : {}",maxPoolSize);ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();executor.setMaxPoolSize(maxPoolSize);executor.setCorePoolSize(corePoolSize);// 队列最大长度//@Value("${async.executor.thread.queue_capacity}")int queueCapacity = 1024;executor.setQueueCapacity(queueCapacity);// 线程池维护线程所允许的空闲时间//@Value("${async.executor.thread.keep_alive_seconds}")int keepAliveSeconds = 300;executor.setKeepAliveSeconds(keepAliveSeconds);executor.setThreadNamePrefix(threadNamePrefix);// 线程池对拒绝任务(无线程可用)的处理策略// CallerRunsPolicy：由调用线程（提交任务的线程）处理该任务executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());// 初始化executor.initialize();return executor;}
}

使用的时候改为@Async(“threadPoolTaskExecutor”)。
在使用类似CompletableFuture的时候需要指定线程池，此时可以通过自动注入ThreadPoolTaskExecutor 即可使用。