异步线程池

Posted 2021-02-02 chunming-an

1.创建类实现AsyncTaskExecutor, InitializingBean, DisposableBean接口，重写方法。

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.core.task.AsyncTaskExecutor;

import java.util.concurrent.Callable;
import java.util.concurrent.Future;

/**
 * The class Exception handling async task executor.
 *
 */
@Slf4j
public class ExceptionHandlingAsyncTaskExecutor implements AsyncTaskExecutor, InitializingBean, DisposableBean {

   private final AsyncTaskExecutor executor;

   /**
    * Instantiates a new Exception handling async task executor.
    *
    * @param executor the executor
    */
   ExceptionHandlingAsyncTaskExecutor(AsyncTaskExecutor executor) {
      this.executor = executor;
   }

   /**
    * Execute.
    *
    * @param task the task
    */
   @Override
   public void execute(Runnable task) {
      executor.execute(createWrappedRunnable(task));
   }

   /**
    * Execute.
    *
    * @param task         the task
    * @param startTimeout the start timeout
    */
   @Override
   public void execute(Runnable task, long startTimeout) {
      executor.execute(createWrappedRunnable(task), startTimeout);
   }

   private <T> Callable<T> createCallable(final Callable<T> task) {
      return () -> {
         try {
            return task.call();
         } catch (Exception e) {
            handle(e);
            throw e;
         }
      };
   }

   private Runnable createWrappedRunnable(final Runnable task) {
      return () -> {
         try {
            task.run();
         } catch (Exception e) {
            handle(e);
         }
      };
   }

   /**
    * Handle.
    *
    * @param e the e
    */
   private void handle(Exception e) {
      log.error("Caught async exception", e);
   }

   /**
    * Submit future.
    *
    * @param task the task
    *
    * @return the future
    */
   @Override
   public Future<?> submit(Runnable task) {
      return executor.submit(createWrappedRunnable(task));
   }

   /**
    * Submit future.
    *
    * @param <T>  the type parameter
    * @param task the task
    *
    * @return the future
    */
   @Override
   public <T> Future<T> submit(Callable<T> task) {
      return executor.submit(createCallable(task));
   }

   /**
    * Destroy.
    *
    * @throws Exception the exception
    */
   @Override
   public void destroy() throws Exception {
      if (executor instanceof DisposableBean) {
         DisposableBean bean = (DisposableBean) executor;
         bean.destroy();
      }
   }

   /**
    * After properties set.
    *
    * @throws Exception the exception
    */
   @Override
   public void afterPropertiesSet() throws Exception {
      if (executor instanceof InitializingBean) {
         InitializingBean bean = (InitializingBean) executor;
         bean.afterPropertiesSet();
      }
   }
}
2.创建AsyncTaskExecutorConfiguration类，实现AsyncConfigurer接口，重写方法

import org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler;
import org.springframework.aop.interceptor.SimpleAsyncUncaughtExceptionHandler;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.AsyncConfigurer;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.concurrent.Executor;

@Configuration
@EnableAsync
@EnableScheduling
public class AsyncTaskExecutorConfiguration implements AsyncConfigurer {

    @Bean(name = "taskExecutor")
    public Executor getAsyncExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(5);
        executor.setMaxPoolSize(20);
        executor.setQueueCapacity(1000);
        executor.setKeepAliveSeconds(60000);
        executor.setThreadNamePrefix("");
        return new ExceptionHandlingAsyncTaskExecutor(executor);
    }

    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return new SimpleAsyncUncaughtExceptionHandler();
    }
}
3.使用。

@Resource
private TaskExecutor taskExecutor;

taskExecutor.execute(new Runnable() {
     @Override
     public void run() {
         try {
             sout（"异步使用线程池"）;
         } catch (IOException e) {
             e.printStackTrace();
         }
     }
 });