前言

        在之前的博客中，曾將講了在SpringBoot中如何使用Event來進行大文件上傳的解耦，原文地址：使用SpringEvent解決WebUploader大文件上傳解耦問題，在這篇博客當中，我們使用Event機制成功的將大文件的上傳和解析的功能進行分離，已經(jīng)實現(xiàn)了解耦的需求。但是在真實項目中會存在一個問題，就是解耦是解耦了。但是我們期望程序能夠做到異步，也就是將文件的上傳和解析進行徹底的異步化。后臺程序在接收前端請求的文件時，文件上傳完成后就結(jié)束。而對于上傳文件的處理和解析等操作則放到解析程序中。整個過程給人的感覺就是到上傳就完成了，解析則可以在后臺慢慢運行，等待執(zhí)行完成即可。

        這里我們?nèi)匀灰源笪募蟼鳛槔?，首先講解在未進行程序異步化的時候，程序的運行機制和表現(xiàn)。然后講解如何進行異步化的改造，讓程序進行異步執(zhí)行。通過本文不僅能讓你掌握如何進行Event的事件開發(fā)，同時還能掌握在Spring中如何進行異步開發(fā)，熟悉@Async的具體用法。

一、場景再現(xiàn)

        為了能讓大家對故事的場景有更加直觀的認識，這里我們將場景進行再現(xiàn)，讓大家看到具體的問題。帶著問題，我們一起來尋找解決辦法，這樣對前因后果更加清楚。

1、Event的同步機制

        首先我們來看一下原來的事件分離處理代碼，關(guān)鍵代碼如下：

@EventListener
public void fileUploadEventRegister(FileUploadEvent event){
	try {
		sys_user_logger.info("當前處理線程名稱:" + Thread.currentThread().getName());
		FileEntity fileEntity = event.getFileEntity();
		if(StringUtils.isNotEmpty(fileEntity.getTablename())){
			FileUploadServiceRegisterEnum rigisterEnum = null;
			if(StringUtils.isNotBlank(fileEntity.getBizType())) {//業(yè)務(wù)類型不為空，則根據(jù)表名和業(yè)務(wù)名稱來查找執(zhí)行service
				rigisterEnum = FileUploadServiceRegisterEnum.getEnumByTableNameAndBizType(fileEntity.getTablename(), fileEntity.getBizType());
			}else {
				rigisterEnum = FileUploadServiceRegisterEnum.getEnumByTableName(fileEntity.getTablename());
			}
			if(null != rigisterEnum && StringUtils.isNotEmpty(rigisterEnum.getExecService())){
				String execService = rigisterEnum.getExecService();
				IFileUploadCallbackService service = SpringUtils.getBean(execService);
				service.process(fileEntity);
			}else{
				sys_user_logger.info("未注冊文件上傳監(jiān)聽回調(diào)處理器.");
			}
		}
	} catch (Exception e) {
		sys_user_logger.error("文件上傳事件監(jiān)聽發(fā)生錯誤.",e);
	}
}

        在這里為了讓文件處理的時間加長，我們可以把文件處理的邏輯加上一個線程的等待時間比如休眠等待35秒鐘。代碼如下：

@Override
@Transactional(propagation=Propagation.REQUIRED,rollbackFor=Exception.class)
public void process(FileEntity fileEntity) throws Exception {
	if(null != fileEntity && StringUtils.isNotEmpty(fileEntity.getBid())){
		String pkId = fileEntity.getBid();
		Student stu = studentService.selectStudentById(Long.valueOf(pkId));
		//System.out.println(fileEntity.getPath());
		//System.out.println(stu.getName() + "\t" + stu.getAddress());
		logger.info("開始處理........");
		Thread.sleep(35 * 1000);//休眠35秒測試
		logger.info("執(zhí)行結(jié)束");
	}
}

        在這里，我們使用Thread.sheep這個方法來讓線程進行休眠等待。模擬文件上傳后，解析很慢的場景。由于卡頓變慢，下面的界面也一直處于等待的狀態(tài)。

        下面是原來的處理線程信息，通過日志可以看到，相關(guān)的執(zhí)行線程都是[http-nio-8080-exec-26]：

20:27:11.280 [http-nio-8080-exec-26] INFO  sys-user - [fileUploadEventRegister,32] - 當前處理線程名稱:http-nio-8080-exec-26
20:27:11.281 [http-nio-8080-exec-26] DEBUG c.y.p.e.s.m.S.selectById - [debug,137] - ==>  Preparing: SELECT id,name,sex,birthday,address,remark,create_by,create_time,update_by,update_time FROM biz_student WHERE id=?
20:27:11.284 [http-nio-8080-exec-26] DEBUG c.y.p.e.s.m.S.selectById - [debug,137] - ==> Parameters: 1811048705298571266(Long)
20:27:11.287 [http-nio-8080-exec-26] DEBUG c.y.p.e.s.m.S.selectById - [debug,137] - <==      Total: 1
20:27:11.288 [http-nio-8080-exec-26] INFO  sys-user - [process,32] - 開始處理........
20:27:36.235 [schedule-pool-2] INFO  c.y.f.s.w.s.OnlineWebSessionManager - [validateSessions,100] - invalidation sessions...
20:27:36.240 [schedule-pool-2] DEBUG c.y.p.m.o.m.U.selectOnlineByExpired - [debug,137] - <==      Total: 0
20:27:36.240 [schedule-pool-2] INFO  c.y.f.s.w.s.OnlineWebSessionManager - [validateSessions,165] - Finished invalidation session. No sessions were stopped.
20:27:46.289 [http-nio-8080-exec-26] INFO  sys-user - [process,34] - 執(zhí)行結(jié)束

        大約經(jīng)過了40秒鐘之后，后臺執(zhí)行完成，前端的界面才響應結(jié)束。 試想一下，如果您是操作的用戶，估計早就奪門而出了。

        使用這種方法開發(fā)完成后，可以看到，前面哪怕上傳一個很小的txt，它的處理時間都是要35秒以上。而我們的預期是這部分解析的功能是相對獨立的，在文件成功上傳后，前臺界面就可以關(guān)閉，文件解析的工作在后臺自動運行。

二、性能優(yōu)化

        在遇到上面的性能和執(zhí)行流程的問題后，我們應該怎么來解決這個事情呢？試想一下，可能有以下幾種方法。比如在后臺加快處理能力，加大運算能力。當然這種方法在遇到海量的文件處理的青提下，性能不一定有明顯的提升。是一種花了成本未必有很好的效果的做法。還有一種可能的做法就是將后臺的處理異步，文件的上傳和解析完全異步。文件上傳成功后，返回響應給前端，然后前端可以繼續(xù)去做其它的事情，而后臺自動去做后續(xù)的文件處理的相關(guān)事宜。這是一個非常友好，也是推薦的做法。因此這里引出我們今天的主角，異步Async調(diào)用。

1、異步支持配置

        要想在spring中實現(xiàn)異步的支持，有很多種方式，這里介紹一種比較簡單易用的方式。首先在我們的事件監(jiān)聽器中增加@Async的注解，第一版代碼如下：

@Async
@EventListener
public void fileUploadEventRegister(FileUploadEvent event){
	try {
		// do something
	} catch (Exception e) {
		sys_user_logger.error("文件上傳事件監(jiān)聽發(fā)生錯誤.",e);
	}
}

        通過上述的代碼呢就實現(xiàn)了一個簡單的異步申明，請記住，如果只是在這里聲明這個方法是異步的，并沒有什么效果。同時還要在應用程序的入口增加開啟異步工作的注解：

@EnableAsync

        你可以把這個注解增加到application的主入口當中。當然，如果到這里，其實也是可以的了。因為我們已經(jīng)實現(xiàn)了整個異步工作的閉環(huán)，開啟異步處理的支持，同時在方法中申明了異步的方法。上面的做法似乎是比較完美的一種做法，仔細想一下，是否真的是這樣呢？還有更好的方法嗎？可以在執(zhí)行的時候看一下，當前方法的工作線程是什么？

2、自定義處理線程池擴展

        為了保證這個業(yè)務(wù)的可用性，其實我們可以自己定義一個線程池來執(zhí)行獨立的文件解析處理的服務(wù)。如果在應用程序中還有其它的服務(wù)的話，彼此之間是不會不想影響的。這也是獨立線程池的好處，同時性能也是得到了大大的提升。

        那么這里將重點說一下如何自定義線程池，如何把線程池應用到處理方法中呢？首先我們來定義個線程池，在Java中創(chuàng)建線程池的關(guān)鍵代碼如下：

package com.yelang.framework.config;
import java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor;
import org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler;
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.concurrent.ThreadPoolTaskExecutor;
/**
 * 文件上傳處理線程池對象，切記：@EnableAsync一定要標記上
 * @author 夜郎king
 *
 */
@Configuration
@EnableAsync
public class FileUploadThreadPoolConfig implements AsyncConfigurer{
	// 核心線程池大小
    private int corePoolSize = 50;
    // 最大可創(chuàng)建的線程數(shù)
    private int maxPoolSize = 200;
    // 隊列最大長度
    private int queueCapacity = 1000;
    // 線程池維護線程所允許的空閑時間
    private int keepAliveSeconds = 300;
    @Bean(name = "fileUploadTaskExecutor")
    public ThreadPoolTaskExecutor threadPoolTaskExecutor()
    {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setMaxPoolSize(maxPoolSize);
        executor.setCorePoolSize(corePoolSize);
        executor.setQueueCapacity(queueCapacity);
        executor.setKeepAliveSeconds(keepAliveSeconds);
        executor.setThreadNamePrefix("Async-FileUpload-Thread-");
        // 線程池對拒絕任務(wù)(無線程可用)的處理策略
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        return executor;
    }
	@Override
	public Executor getAsyncExecutor() {
		return AsyncConfigurer.super.getAsyncExecutor();
	}
	@Override
	public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
		return AsyncConfigurer.super.getAsyncUncaughtExceptionHandler();
	}
}

        在這里，我們定義了線程對象的前綴，主要是用于確定在執(zhí)行文件解析的時候，是否是使用線程池中的線程進行解析工作的。同時我們將@EnableAsync這個注解從application的主入口轉(zhuǎn)移到了這個配置類中。請注意，這個線程池中的初始容量，最大容量，隊列長度都是固定的，實際情況下可以根據(jù)配置和任務(wù)情況進行調(diào)整，同時設(shè)置了線程的拒絕策略。

3、將線程池配置類綁定到異步方法

        將線程池配置類定義好之后，為了讓程序能夠運行起來，我們還需要將線程池配置類綁定到異步方法的注解中，如下所示：

//在這里指定用fileUploadTaskExecutor這個線程池去處理
@Async(value="fileUploadTaskExecutor")
@EventListener
public void fileUploadEventRegister(FileUploadEvent event){
	try {
		// do somethings
	} catch (Exception e) {
		sys_user_logger.error("文件上傳事件監(jiān)聽發(fā)生錯誤.",e);
	}
}

        在執(zhí)行的監(jiān)聽器中綁定注冊線程池之后，我們來看一下實際的執(zhí)行效果。同時主要觀察在實際的執(zhí)行過程中，是否是使用設(shè)置的線程池中的線程來進行執(zhí)行相應的業(yè)務(wù)的。從前臺的上傳來看，界面很快有了返回。速度是很快的。同時在后臺可以看到相關(guān)的線程處理信息，入下所示：

20:52:48.959 [Async-FileUpload-Thread-1] INFO  sys-user - [fileUploadEventRegister,32] - 當前處理線程名稱:Async-FileUpload-Thread-1
20:52:48.967 [Async-FileUpload-Thread-1] DEBUG c.y.p.e.s.m.S.selectById - [debug,137] - ==>  Preparing: SELECT id,name,sex,birthday,address,remark,create_by,create_time,update_by,update_time FROM biz_student WHERE id=?
20:52:48.968 [Async-FileUpload-Thread-1] DEBUG c.y.p.e.s.m.S.selectById - [debug,137] - ==> Parameters: 1811048705298571266(Long)
20:52:48.971 [Async-FileUpload-Thread-1] DEBUG c.y.p.e.s.m.S.selectById - [debug,137] - <==      Total: 1
20:52:48.972 [Async-FileUpload-Thread-1] INFO  sys-user - [process,32] - 開始處理........
20:52:48.983 [http-nio-8080-exec-71] DEBUG c.y.p.w.m.F.selectList_COUNT - [debug,137] - ==>  Preparing: SELECT count(0) FROM biz_file WHERE (f_state = ? AND b_id = ? AND table_name = ? AND biz_type = ?)
20:52:48.984 [http-nio-8080-exec-71] DEBUG c.y.p.w.m.F.selectList_COUNT - [debug,137] - ==> Parameters: 1(Integer), 1811048705298571266(String), biz_student(String), 123a(String)
20:52:49.003 [http-nio-8080-exec-71] DEBUG c.y.p.w.m.F.selectList_COUNT - [debug,137] - <==      Total: 1
20:52:49.008 [http-nio-8080-exec-71] DEBUG c.y.p.w.m.F.selectList - [debug,137] - ==>  Preparing: SELECT id, f_id, b_id, f_type AS type, f_name AS name, f_desc AS desc, f_state AS state, f_size AS size, f_path AS path, table_name, md5code, directory, biz_type, create_by, create_time, update_by, update_time FROM biz_file WHERE (f_state = ? AND b_id = ? AND table_name = ? AND biz_type = ?) order by create_time desc LIMIT ?
20:53:23.973 [Async-FileUpload-Thread-1] INFO  sys-user - [process,34] - 執(zhí)行結(jié)束

        可以很明顯的看到，文件的解析程序是使用Async-FileUpload-Thread開頭的線程來進行處理的，即表名是正常的使用線程池來進行處理相關(guān)的業(yè)務(wù)。也說明的我們的設(shè)計達到了預期，即實現(xiàn)了程序的完全異步化。

三、總結(jié)

        以上就是本文的主要內(nèi)容，本文以大文件上傳為例，首先講解在未進行程序異步化的時候，程序的運行機制和具體表現(xiàn)。然后講解如何進行異步化的改造，讓程序進行異步執(zhí)行。通過本文不僅能讓你掌握如何進行Event的事件開發(fā)，同時還能掌握在Spring中如何進行異步開發(fā)，熟悉@Async的具體用法。行文倉促，難免有不足之處，如果有表達不當?shù)牡胤交蛘卟蛔阒?，還請各位專家批評指正，在評論區(qū)留下您的真知灼見，萬分榮幸。

到此這篇關(guān)于Spring中使用Async進行異步功能開發(fā)實戰(zhàn)的文章就介紹到這了,更多相關(guān)Spring Async異步開發(fā)內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評論