調用接口，異步處理業務

調用接口時，如果后台需要處理的時間過長，需要采取異步處理，先把結果返回給前台。

1、原生的

接口定義：

@RequestMapping(value="/test")
    public Object test(){
         MyExecutor  myExecutor = new MyExecutor();
         try {
            myExecutor.work();
        }catch(Exception e) {
            System.out.println("myExecutor.work()"+e.getMessage());
        }
         System.out.println("返回結果:  "+new Date());
        return "成功";
    }

業務執行：

import java.util.Date;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class MyExecutor {

    private ExecutorService executor = Executors.newCachedThreadPool() ;

    public void work() throws Exception {
        executor.submit(new Runnable(){
            public void run() {
                try {
                    System.out.println("開始處理業務。。。    "+new Date());
                    Thread.sleep(5000);
                    System.out.println("業務結束。。。   "+new Date());
                }catch(Exception e) {
                    System.out.println(e.getMessage());
                }
            }
        });
    }

}

控制台輸出：

返回結果:  Sat Jul 27 09:28:33 GMT+08:00 2019
開始處理業務。。。    Sat Jul 27 09:28:33 GMT+08:00 2019
業務結束。。。   Sat Jul 27 09:28:38 GMT+08:00 2019

 

2、在springboot  中使用

controller層

@RequestMapping(value="/async2")
    public Object testAsync2(){
        System.out.println(1111);
        userService.testAsync();
        System.out.println(12);
        return "asdfas";
    }

異步方法加注解    @Async （org.springframework.scheduling.annotation.Async;） 

    @Override
    @Async
    public void testAsync() {
        System.out.println("開始異步處理業務。。");
        try {
            Thread.sleep(5000);
            System.out.println("結束。。。。。");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

啟動類加注解  @EnableAsync  開啟異步支持

 

擴展：

異步方法有返回值

@Async
    public Future<String> asyncMethodWithReturnType() {
         System.out.println("Execute method asynchronously - "+ Thread.currentThread().getName());
         try {
             Thread.sleep(5000);
             return new AsyncResult<String>("hello world !!!!");
         } catch (InterruptedException e) {
             //
         }
     
         return null;
    }

 

異步操作的執行器

默認情況下，Spring 使用SimpleAsyncTaskExecutor去執行這些異步方法（此執行器沒有限制線程數）。此默認值可以從兩個層級進行覆蓋：

1. 方法級別覆蓋

@Async("threadPoolTaskExecutor")
    public void asyncMethodWithConfiguredExecutor() {
        System.out.println("Execute method with configured executor - "+ Thread.currentThread().getName());
    }

2. 應用級別覆蓋

配置類應該實現AsyncConfigurer接口——這意味着它擁有getAsyncExecutor()方法的實現。在這里，我們將返回整個應用程序的執行器——這現在成為運行帶有@Async注釋的方法的默認執行器:(異步中的線程名稱-- ThreadPoolTaskExecutor-1 ，與所使用的執行相關)

import java.util.concurrent.Executor;

import org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler;
import org.springframework.scheduling.annotation.AsyncConfigurer;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

@Configuration
@EnableAsync
public class SpringAsyncConfig implements AsyncConfigurer{

    @Override
    public Executor getAsyncExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.initialize();
        executor.setCorePoolSize(5);
        executor.setMaxPoolSize(10);
        executor.setQueueCapacity(25);
        return executor;
    }

    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return new CustomAsyncExceptionHandler();
    }
}

CustomAsyncExceptionHandler 是一個自定義的異常捕捉，當方法返回值是Future的時候，異常捕獲是沒問題的 - Future.get()方法會拋出異常。但是，如果返回類型是Void，那么異常在當前線程就捕獲不到。因此，我們需要添加額外的配置來處理異常。
通過實現AsyncUncaughtExceptionHandler接口創建一個定制的async異常處理程序。handleUncaughtException()方法在存在任何未捕獲的異步異常時調用:

import java.lang.reflect.Method;

import org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler;

public class CustomAsyncExceptionHandler implements AsyncUncaughtExceptionHandler {

    @Override
    public void handleUncaughtException(
            Throwable throwable, Method method, Object... obj) {
        System.out.println("Exception message - " + throwable.getMessage());
        System.out.println("Method name - " + method.getName());
        for (Object param : obj) {
            System.out.println("Parameter value - " + param);
        }

    }

}

 

 

3、spring MVC 中使用

需要加配置

相關的配置：

<task:annotation-driven />配置：

• executor：指定一個缺省的executor給@Async使用。

例子：

<task:annotation-driven executor="asyncExecutor" />

 

<task:executor />配置參數：

• id：當配置多個executor時，被@Async("id")指定使用；也被作為線程名的前綴。
• pool-size：

• core size：最小的線程數，缺省：1
• max size：最大的線程數，缺省：Integer.MAX_VALUE
• queue-capacity：當最小的線程數已經被占用滿后，新的任務會被放進queue里面，當這個queue的capacity也被占滿之后，pool里面會創建新線程處理這個任務，直到總線程數達到了max size，這時系統會拒絕這個任務並拋出TaskRejectedException異常（缺省配置的情況下，可以通過rejection-policy來決定如何處理這種情況）。缺省值為：Integer.MAX_VALUE
• keep-alive：超過core size的那些線程，任務完成后，再經過這個時長（秒）會被結束掉
• rejection-policy：當pool已經達到max size的時候，如何處理新任務
  • ABORT（缺省）：拋出TaskRejectedException異常，然后不執行
  • DISCARD：不執行，也不拋出異常
  • DISCARD_OLDEST：丟棄queue中最舊的那個任務
  • CALLER_RUNS：不在新線程中執行任務，而是有調用者所在的線程來執行

配置例子：

<task:annotation-driven executor="asyncExecutor" />
 <task:executor id="asyncExecutor" pool-size="100-10000" queue-capacity="10"/>

 

實例：

<!-- 缺省的異步任務線程池 -->   
<task:annotation-driven executor="asyncExecutor" />  
<task:executor id="asyncExecutor" pool-size="100-10000" queue-capacity="10" />  
  
<!-- 處理log的線程池 -->  
<task:executor id="logExecutor" pool-size="15-1000" queue-capacity="5" keep-alive="5"/>  

@Override  
@Async("logExecutor")    //如果不指定名字，會使用缺省的“asyncExecutor”  
public void saveUserOpLog(TabUserOpLog tabUserOpLog) {  
  
 userOpLogDAO.insertTabUserOpLog(tabUserOpLog);  
}

（注意：如果在同一個類中調用的話，不會生效，原因請參考：http://blog.csdn.net/clementad/article/details/47339519）

線程的優先級和類型：

優先級：NORM_PRIORITY

類型：非守護線程

 

用戶線程（User Thread）：JVM會等待所有的用戶線程結束后才退出；當系統中沒有用戶線程了，JVM也就退出了

守護線程（Daemon Thread）：一般是為其他線程提供服務的線程，比如GC垃圾回收器；JVM退出時，不會管守護線程是否存在，而是直接退出

所以，對於文件、數據庫的操作，不適宜使用守護線程，不然可能會丟失數據！

 

Web應用停止時，Spring容器會被關閉，調用者如果是Spring bean，就會停止生成新任務。然而，線程池中已經在運行的任務，由於缺省是用戶線程，所以JVM會等待它們結束后才退出。

 

 附：Java編程方式的配置方法：

@Configuration  
@EnableAsync  
public class SpringConfig {  
  
    /** Set the ThreadPoolExecutor's core pool size. */  
    private int corePoolSize = 10;  
    /** Set the ThreadPoolExecutor's maximum pool size. */  
    private int maxPoolSize = 200;  
    /** Set the capacity for the ThreadPoolExecutor's BlockingQueue. */  
    private int queueCapacity = 10;  
  
    private String ThreadNamePrefix = "MyLogExecutor-";  
  
    @Bean  
    public Executor logExecutor() {  
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();  
        executor.setCorePoolSize(corePoolSize);  
        executor.setMaxPoolSize(maxPoolSize);  
        executor.setQueueCapacity(queueCapacity);  
        executor.setThreadNamePrefix(ThreadNamePrefix);  
  
        // rejection-policy：當pool已經達到max size的時候，如何處理新任務  
        // CALLER_RUNS：不在新線程中執行任務，而是有調用者所在的線程來執行  
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());  
        executor.initialize();  
        return executor;  
    }  
  
}  

 

參考:  https://www.cnblogs.com/panxuejun/p/7838970.html

https://blog.csdn.net/xiaoxiaole0313/article/details/104666789

SimpleAsyncTaskExecutor