Spring boot异步任务原理全面分析

  org.springframework.bootspring-boot-starter-web2.1.3.RELEASE org.springframework.bootspring-boot-starter-test2.1.3.RELEASEtest

@RestController @SpringBootApplication public class AsyncMain { public static void main(String[] args) { SpringApplication.run(AsyncMain.class, args); } @Autowired private TaskService taskService; @RequestMapping(value = "/async-task", method = RequestMethod.GET) public String asyncMapping(){ System.out.println(Thread.currentThread().getThreadGroup() + "http-------" + Thread.currentThread().getName()); taskService.doTask(); return "exec http ok--------------"; } }

@EnableAsync @Service public class TaskService { @Async public String doTask(){ System.out.println(Thread.currentThread().getThreadGroup() + "-------" + Thread.currentThread().getName()); return "do task done"; } }

/** * {@link EnableAutoConfiguration Auto-configuration} for {@link TaskExecutor}. * * @author Stephane Nicoll * @author Camille Vienot * @since 2.1.0 */ @ConditionalOnClass(ThreadPoolTaskExecutor.class) @Configuration @EnableConfigurationProperties(TaskExecutionProperties.class) public class TaskExecutionAutoConfiguration { /** * Bean name of the application {@link TaskExecutor}. */ public static final String APPLICATION_TASK_EXECUTOR_BEAN_NAME = "applicationTaskExecutor"; private final TaskExecutionProperties properties; private final ObjectProvider taskExecutorCustomizers; private final ObjectProvider taskDecorator; public TaskExecutionAutoConfiguration(TaskExecutionProperties properties, ObjectProvider taskExecutorCustomizers, ObjectProvider taskDecorator) { this.properties = properties; this.taskExecutorCustomizers = taskExecutorCustomizers; this.taskDecorator = taskDecorator; } @Bean @ConditionalOnMissingBean public TaskExecutorBuilder taskExecutorBuilder() { TaskExecutionProperties.Pool pool = this.properties.getPool(); TaskExecutorBuilder builder = new TaskExecutorBuilder(); builder = builder.queueCapacity(pool.getQueueCapacity()); builder = builder.corePoolSize(pool.getCoreSize()); builder = builder.maxPoolSize(pool.getMaxSize()); builder = builder.allowCoreThreadTimeOut(pool.isAllowCoreThreadTimeout()); builder = builder.keepAlive(pool.getKeepAlive()); builder = builder.threadNamePrefix(this.properties.getThreadNamePrefix()); builder = builder.customizers(this.taskExecutorCustomizers); builder = builder.taskDecorator(this.taskDecorator.getIfUnique()); return builder; } @Lazy @Bean(name = { APPLICATION_TASK_EXECUTOR_BEAN_NAME, AsyncAnnotationBeanPostProcessor.DEFAULT_TASK_EXECUTOR_BEAN_NAME }) @ConditionalOnMissingBean(Executor.class) public ThreadPoolTaskExecutor applicationTaskExecutor(TaskExecutorBuilder builder) { return builder.build(); } }

@ConfigurationProperties("spring.task.execution") public class TaskExecutionProperties { private final Pool pool = new Pool(); /** * Prefix to use for the names of newly created threads. */ private String threadNamePrefix = "task-"; public Pool getPool() { return this.pool; } public String getThreadNamePrefix() { return this.threadNamePrefix; } public void setThreadNamePrefix(String threadNamePrefix) { this.threadNamePrefix = threadNamePrefix; } public static class Pool { /** * Queue capacity. An unbounded capacity does not increase the pool and therefore * ignores the "max-size" property. */ private int queueCapacity = Integer.MAX_VALUE; /** * Core number of threads. */ private int coreSize = 8; /** * Maximum allowed number of threads. If tasks are filling up the queue, the pool * can expand up to that size to accommodate the load. Ignored if the queue is * unbounded. */ private int maxSize = Integer.MAX_VALUE; /** * Whether core threads are allowed to time out. This enables dynamic growing and * shrinking of the pool. */ private boolean allowCoreThreadTimeout = true; /** * Time limit for which threads may remain idle before being terminated. */ private Duration keepAlive = Duration.ofSeconds(60);

    @Lazy     @Bean(name = { APPLICATION_TASK_EXECUTOR_BEAN_NAME,             AsyncAnnotationBeanPostProcessor.DEFAULT_TASK_EXECUTOR_BEAN_NAME })     @ConditionalOnMissingBean(Executor.class)     public ThreadPoolTaskExecutor applicationTaskExecutor(TaskExecutorBuilder builder) {         return builder.build();     }

@Configuration @EnableAsync public class TaskAsyncConfig { @Bean public Executor initExecutor(){ ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); //定制线程名称，还可以定制线程group executor.setThreadFactory(new ThreadFactory() { private final AtomicInteger threadNumber = new AtomicInteger(1); @Override public Thread newThread(Runnable r) { Thread t = new Thread(Thread.currentThread().getThreadGroup(), r, "async-task-" + threadNumber.getAndIncrement(), 0); return t; } }); executor.setCorePoolSize(10); executor.setMaxPoolSize(20); executor.setKeepAliveSeconds(5); executor.setQueueCapacity(100); //        executor.setRejectedExecutionHandler(null); return executor; } }

@EnableAsync @Service public class TaskService { @Async public String doTask(){ System.out.println(Thread.currentThread().getThreadGroup() + "-------" + Thread.currentThread().getName()); throw new RuntimeException(" I`m a demo test exception-----------------"); } }

public class SimpleAsyncUncaughtExceptionHandler implements AsyncUncaughtExceptionHandler { private static final Log logger = LogFactory.getLog(SimpleAsyncUncaughtExceptionHandler.class); @Override public void handleUncaughtException(Throwable ex, Method method, Object... params) { if (logger.isErrorEnabled()) { logger.error("Unexpected exception occurred invoking async method: " + method, ex); } } }

/** * Interface to be implemented by @{@link org.springframework.context.annotation.Configuration * Configuration} classes annotated with @{@link EnableAsync} that wish to customize the * {@link Executor} instance used when processing async method invocations or the * {@link AsyncUncaughtExceptionHandler} instance used to process exception thrown from * async method with {@code void} return type. * * 
Consider using {@link AsyncConfigurerSupport} providing default implementations for * both methods if only one element needs to be customized. Furthermore, backward compatibility * of this interface will be insured in case new customization options are introduced * in the future. * * 
See @{@link EnableAsync} for usage examples. * * @author Chris Beams * @author Stephane Nicoll * @since 3.1 * @see AbstractAsyncConfiguration * @see EnableAsync * @see AsyncConfigurerSupport */ public interface AsyncConfigurer { /** * The {@link Executor} instance to be used when processing async * method invocations. */ @Nullable default Executor getAsyncExecutor() { return null; } /** * The {@link AsyncUncaughtExceptionHandler} instance to be used * when an exception is thrown during an asynchronous method execution * with {@code void} return type. */ @Nullable default AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() { return null; } }

@Configuration @EnableAsync public class TaskAsyncConfig implements AsyncConfigurer { @Override public Executor getAsyncExecutor() { ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); //定制线程名称，还可以定制线程group executor.setThreadFactory(new ThreadFactory() { private final AtomicInteger threadNumber = new AtomicInteger(1); @Override public Thread newThread(Runnable r) { //重新定义一个名称 Thread t = new Thread(Thread.currentThread().getThreadGroup(), r, "async-task-all" + threadNumber.getAndIncrement(), 0); return t; } }); executor.setCorePoolSize(10); executor.setMaxPoolSize(20); executor.setKeepAliveSeconds(5); executor.setQueueCapacity(100); //        executor.setRejectedExecutionHandler(null); executor.initialize(); return executor; } @Override public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() { return new AsyncUncaughtExceptionHandler() { @Override public void handleUncaughtException(Throwable ex, Method method, Object... params) { System.out.println("do exception by myself"); } }; } }

executor.initialize();

@Override protected ExecutorService initializeExecutor( ThreadFactory threadFactory, RejectedExecutionHandler rejectedExecutionHandler) { BlockingQueue queue = createQueue(this.queueCapacity); ThreadPoolExecutor executor; if (this.taskDecorator != null) { executor = new ThreadPoolExecutor( this.corePoolSize, this.maxPoolSize, this.keepAliveSeconds, TimeUnit.SECONDS, queue, threadFactory, rejectedExecutionHandler) { @Override public void execute(Runnable command) { Runnable decorated = taskDecorator.decorate(command); if (decorated != command) { decoratedTaskMap.put(decorated, command); } super.execute(decorated); } }; } else { executor = new ThreadPoolExecutor( this.corePoolSize, this.maxPoolSize, this.keepAliveSeconds, TimeUnit.SECONDS, queue, threadFactory, rejectedExecutionHandler); } if (this.allowCoreThreadTimeOut) { executor.allowCoreThreadTimeOut(true); } this.threadPoolExecutor = executor; return executor; }

/** * AOP Alliance {@code MethodInterceptor} that processes method invocations * asynchronously, using a given {@link org.springframework.core.task.AsyncTaskExecutor}. * Typically used with the {@link org.springframework.scheduling.annotation.Async} annotation. * * 
In terms of target method signatures, any parameter types are supported. * However, the return type is constrained to either {@code void} or * {@code java.util.concurrent.Future}. In the latter case, the Future handle * returned from the proxy will be an actual asynchronous Future that can be used * to track the result of the asynchronous method execution. However, since the * target method needs to implement the same signature, it will have to return * a temporary Future handle that just passes the return value through * (like Spring's {@link org.springframework.scheduling.annotation.AsyncResult} * or EJB 3.1's {@code javax.ejb.AsyncResult}). * * 
When the return type is {@code java.util.concurrent.Future}, any exception thrown * during the execution can be accessed and managed by the caller. With {@code void} * return type however, such exceptions cannot be transmitted back. In that case an * {@link AsyncUncaughtExceptionHandler} can be registered to process such exceptions. * * 
As of Spring 3.1.2 the {@code AnnotationAsyncExecutionInterceptor} subclass is * preferred for use due to its support for executor qualification in conjunction with * Spring's {@code @Async} annotation. * * @author Juergen Hoeller * @author Chris Beams * @author Stephane Nicoll * @since 3.0 * @see org.springframework.scheduling.annotation.Async * @see org.springframework.scheduling.annotation.AsyncAnnotationAdvisor * @see org.springframework.scheduling.annotation.AnnotationAsyncExecutionInterceptor */ public class AsyncExecutionInterceptor extends AsyncExecutionAspectSupport implements MethodInterceptor, Ordered {

In terms of target method signatures, any parameter types are supported.  * However, the return type is constrained to either {@code void} or  * {@code java.util.concurrent.Future}. In the latter case, the Future handle  * returned from the proxy will be an actual asynchronous Future that can be used  * to track the result of the asynchronous method execution. However, since the  * target method needs to implement the same signature, it will have to return  * a temporary Future handle that just passes the return value through  * (like Spring's {@link org.springframework.scheduling.annotation.AsyncResult}  * or EJB 3.1's {@code javax.ejb.AsyncResult}).