前言

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

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

一、場(chǎng)景再現(xiàn)

        為了能讓大家對(duì)故事的場(chǎng)景有更加直觀(guān)的認(rèn)識(shí)，這里我們將場(chǎng)景進(jìn)行再現(xiàn)，讓大家看到具體的問(wèn)題。帶著問(wèn)題，我們一起來(lái)尋找解決辦法，這樣對(duì)前因后果更加清楚。

1、Event的同步機(jī)制

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

@EventListener
public void fileUploadEventRegister(FileUploadEvent event){
	try {
		sys_user_logger.info("當(dāng)前處理線(xiàn)程名稱(chēng):" + Thread.currentThread().getName());
		FileEntity fileEntity = event.getFileEntity();
		if(StringUtils.isNotEmpty(fileEntity.getTablename())){
			FileUploadServiceRegisterEnum rigisterEnum = null;
			if(StringUtils.isNotBlank(fileEntity.getBizType())) {//業(yè)務(wù)類(lèi)型不為空，則根據(jù)表名和業(yè)務(wù)名稱(chēng)來(lái)查找執(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("未注冊(cè)文件上傳監(jiān)聽(tīng)回調(diào)處理器.");
			}
		}
	} catch (Exception e) {
		sys_user_logger.error("文件上傳事件監(jiān)聽(tīng)發(fā)生錯(cuò)誤.",e);
	}
}

        在這里為了讓文件處理的時(shí)間加長(zhǎng)，我們可以把文件處理的邏輯加上一個(gè)線(xiàn)程的等待時(shí)間比如休眠等待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("開(kāi)始處理........");
		Thread.sleep(35 * 1000);//休眠35秒測(cè)試
		logger.info("執(zhí)行結(jié)束");
	}
}

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

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

20:27:11.280 [http-nio-8080-exec-26] INFO  sys-user - [fileUploadEventRegister,32] - 當(dāng)前處理線(xiàn)程名稱(chēng):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] - 開(kāi)始處理........
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)過(guò)了40秒鐘之后，后臺(tái)執(zhí)行完成，前端的界面才響應(yīng)結(jié)束。 試想一下，如果您是操作的用戶(hù)，估計(jì)早就奪門(mén)而出了。

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

二、性能優(yōu)化

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

1、異步支持配置

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

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

        通過(guò)上述的代碼呢就實(shí)現(xiàn)了一個(gè)簡(jiǎn)單的異步申明，請(qǐng)記住，如果只是在這里聲明這個(gè)方法是異步的，并沒(méi)有什么效果。同時(shí)還要在應(yīng)用程序的入口增加開(kāi)啟異步工作的注解：

@EnableAsync

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

2、自定義處理線(xiàn)程池?cái)U(kuò)展

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

        那么這里將重點(diǎn)說(shuō)一下如何自定義線(xiàn)程池，如何把線(xiàn)程池應(yīng)用到處理方法中呢？首先我們來(lái)定義個(gè)線(xiàn)程池，在Java中創(chuàng)建線(xiàn)程池的關(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;
/**
 * 文件上傳處理線(xiàn)程池對(duì)象，切記：@EnableAsync一定要標(biāo)記上
 * @author 夜郎king
 *
 */
@Configuration
@EnableAsync
public class FileUploadThreadPoolConfig implements AsyncConfigurer{
	// 核心線(xiàn)程池大小
    private int corePoolSize = 50;
    // 最大可創(chuàng)建的線(xiàn)程數(shù)
    private int maxPoolSize = 200;
    // 隊(duì)列最大長(zhǎng)度
    private int queueCapacity = 1000;
    // 線(xiàn)程池維護(hù)線(xiàn)程所允許的空閑時(shí)間
    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-");
        // 線(xiàn)程池對(duì)拒絕任務(wù)(無(wú)線(xiàn)程可用)的處理策略
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        return executor;
    }
	@Override
	public Executor getAsyncExecutor() {
		return AsyncConfigurer.super.getAsyncExecutor();
	}
	@Override
	public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
		return AsyncConfigurer.super.getAsyncUncaughtExceptionHandler();
	}
}

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

3、將線(xiàn)程池配置類(lèi)綁定到異步方法

        將線(xiàn)程池配置類(lèi)定義好之后，為了讓程序能夠運(yùn)行起來(lái)，我們還需要將線(xiàn)程池配置類(lèi)綁定到異步方法的注解中，如下所示：

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

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

20:52:48.959 [Async-FileUpload-Thread-1] INFO  sys-user - [fileUploadEventRegister,32] - 當(dāng)前處理線(xiàn)程名稱(chēng):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] - 開(kāi)始處理........
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開(kāi)頭的線(xiàn)程來(lái)進(jìn)行處理的，即表名是正常的使用線(xiàn)程池來(lái)進(jìn)行處理相關(guān)的業(yè)務(wù)。也說(shuō)明的我們的設(shè)計(jì)達(dá)到了預(yù)期，即實(shí)現(xiàn)了程序的完全異步化。

三、總結(jié)

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

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

最新評(píng)論