使用AOP實現(xiàn)異步上傳

前言

相信很多系統(tǒng)里都有這一種場景：用戶上傳Excel，后端解析Excel生成相應(yīng)的數(shù)據(jù)，校驗數(shù)據(jù)并落庫。這就引發(fā)了一個問題：如果Excel的行非常多，或者解析非常復(fù)雜，那么解析+校驗的過程就非常耗時。

如果接口是一個同步的接口，則非常容易出現(xiàn)接口超時，進而返回的校驗錯誤信息也無法展示給前端，這就需要從功能上解決這個問題。一般來說都是啟動一個子線程去做解析工作，主線程正常返回，由子線程記錄上傳狀態(tài)+校驗結(jié)果到數(shù)據(jù)庫。同時提供一個查詢頁面用于實時查詢上傳的狀態(tài)和校驗信息。

進一步的，如果我們每一個上傳的任務(wù)都寫一次線程池異步+日志記錄的代碼就顯得非常冗余。同時，非業(yè)務(wù)代碼也侵入了業(yè)務(wù)代碼導(dǎo)致代碼可讀性下降。

從通用性的角度上講，這種業(yè)務(wù)場景非常適合模板方法的設(shè)計模式。即設(shè)計一個抽象類，定義上傳的抽象方法，同時實現(xiàn)記錄日志的方法，例如：

//偽代碼，省略了一些步驟
@Slf4j
publicabstractclassAbstractUploadService<T>{
publicstaticThreadFactorycommonThreadFactory=newThreadFactoryBuilder().setNameFormat("-upload-pool-%d")
.setPriority(Thread.NORM_PRIORITY).build();
publicstaticExecutorServiceuploadExecuteService=newThreadPoolExecutor(10,20,300L,
TimeUnit.SECONDS,newLinkedBlockingQueue<>(1024),commonThreadFactory,newThreadPoolExecutor.AbortPolicy());

protectedabstractStringupload(Listdata);

protectedvoidexecute(StringuserName,Listdata){
//生成一個唯一編號
Stringuuid=UUID.randomUUID().toString().replace("-","");
uploadExecuteService.submit(()->{
//記錄日志
writeLogToDb(uuid,userName,updateTime,"導(dǎo)入中");
//一個字符串，用于記錄upload的校驗信息
StringerrorLog="";
//執(zhí)行上傳
try{
errorLog=upload(data);
writeSuccess(uuid,"導(dǎo)入中",updateTime);
}catch(Exceptione){
LOGGER.error("導(dǎo)入錯誤",e);
//計入導(dǎo)入錯誤日志
writeFailToDb(uuid,"導(dǎo)入失敗",e.getMessage(),updateTime);
}
/**
*檢查一下upload是不是返回了錯誤日志，如果有，需要注意記錄
*
*因為錯誤日志可能比較長，
*可以寫入一個文件然后上傳到公司的文件服務(wù)器，
*然后在查看結(jié)果的時候允許用戶下載該文件，
*這里不展開只做示意
*/
if(StringUtils.isNotEmpty(errorLog)){
writeFailToDb(uuid,"導(dǎo)入失敗",errorLog,updateTime);
}

});
}
}

如上文所示，模板方法的方式雖然能夠極大地減少重復(fù)代碼，但是仍有下面兩個問題：

• upload方法得限定死參數(shù)結(jié)構(gòu)，一旦有變化，不是很容易更改參數(shù)類型or數(shù)量
• 每個上傳的service還是要繼承一下這個抽象類，還是不夠簡便和優(yōu)雅

為解決上面兩個問題，我也經(jīng)常進行思考，結(jié)果在某次自定義事務(wù)提交or回滾的方法的時候得到了啟發(fā)。這個上傳的邏輯過程和事務(wù)提交的邏輯過程非常像，都是在實際操作前需要做初始化操作，然后在異?；蛘叱晒Φ臅r候做進一步操作。這種完全可以通過環(huán)裝切面的方式實現(xiàn)，由此，我寫了一個小輪子給團隊使用。

當(dāng)然了，這個小輪子在本人所在的大團隊內(nèi)部使用的很好，但是不一定適合其他人，但是思路一樣，大家可以擴展自己的功能

多說無益，上代碼！

代碼與實現(xiàn)

首先定義一個日志實體

publicclassFileUploadLog{
privateIntegerid;
//唯一編碼
privateStringbatchNo;
//上傳到文件服務(wù)器的文件key
privateStringkey;
//錯誤日志文件名
privateStringfileName;
//上傳狀態(tài)
privateIntegerstatus;
//上傳人
privateStringcreateName;
//上傳類型
privateStringuploadType;
//結(jié)束時間
privateDateendTime;
//開始時間
privateDatestartTime;
}

然后定義一個上傳的類型枚舉，用于記錄是哪里操作的

publicenumUploadType{
未知(1,"未知"),
類型2(2,"類型2"),
類型1(3,"類型1");

privateintcode;
privateStringdesc;
privatestaticMapmap=newHashMap<>();
static{
for(UploadTypevalue:UploadType.values()){
map.put(value.code,value);
}
}

UploadType(intcode,Stringdesc){
this.code=code;
this.desc=desc;
}

publicintgetCode(){
returncode;
}

publicStringgetDesc(){
returndesc;
}

publicstaticUploadTypegetByCode(Integercode){
returnmap.get(code);
}
}

最后，定義一個注解，用于標(biāo)識切點

@Retention(RetentionPolicy.RUNTIME)
@Target({ElementType.METHOD})
public@interfaceUpload{
//記錄上傳類型
UploadTypetype()defaultUploadType.未知;
}

然后，編寫切面

@Component
@Aspect
@Slf4j
publicclassUploadAspect{
publicstaticThreadFactorycommonThreadFactory=newThreadFactoryBuilder().setNameFormat("upload-pool-%d")
.setPriority(Thread.NORM_PRIORITY).build();
publicstaticExecutorServiceuploadExecuteService=newThreadPoolExecutor(10,20,300L,
TimeUnit.SECONDS,newLinkedBlockingQueue<>(1024),commonThreadFactory,newThreadPoolExecutor.AbortPolicy());


@Pointcut("@annotation(com.aaa.bbb.Upload)")
publicvoiduploadPoint(){}

@Around(value="uploadPoint()")
publicObjectuploadControl(ProceedingJoinPointpjp){
//獲取方法上的注解，進而獲取uploadType
MethodSignaturesignature=(MethodSignature)pjp.getSignature();
Uploadannotation=signature.getMethod().getAnnotation(Upload.class);
UploadTypetype=annotation==null?UploadType.未知:annotation.type();
//獲取batchNo
StringbatchNo=UUID.randomUUID().toString().replace("-","");
//初始化一條上傳的日志，記錄開始時間
writeLogToDB(batchNo,type,newDate)
//線程池啟動異步線程，開始執(zhí)行上傳的邏輯，pjp.proceed()就是你實現(xiàn)的上傳功能
uploadExecuteService.submit(()->{
try{
StringerrorMessage=pjp.proceed();
//沒有異常直接成功
if(StringUtils.isEmpty(errorMessage)){
//成功，寫入數(shù)據(jù)庫，具體不展開了
writeSuccessToDB(batchNo);
}else{
//失敗，因為返回了校驗信息
fail(errorMessage,batchNo);
}
}catch(Throwablee){
LOGGER.error("導(dǎo)入失?。?,e);
//失敗，拋了異常，需要記錄
fail(e.toString(),batchNo);
}
});
returnnewObject();
}

privatevoidfail(Stringmessage,StringbatchNo){
//生成上傳錯誤日志文件的文件key
Strings3Key=UUID.randomUUID().toString().replace("-","");
//生成文件名稱
StringfileName="錯誤日志_"+
DateUtil.dateToString(newDate(),"yyyy年MM月dd日HH時mm分ss秒")+ExportConstant.txtSuffix;
StringfilePath="/home/xxx/xxx/"+fileName;
//生成一個文件，寫入錯誤數(shù)據(jù)
Filefile=newFile(filePath);
OutputStreamoutputStream=null;
try{
outputStream=newFileOutputStream(file);
outputStream.write(message.getBytes());

}catch(Exceptione){
LOGGER.error("寫入文件錯誤",e);
}finally{
try{
if(outputStream!=null)
outputStream.close();
}catch(Exceptione){
LOGGER.error("關(guān)閉錯誤",e);
}
}
//上傳錯誤日志文件到文件服務(wù)器，我們用的是s3
upFileToS3(file,s3Key);
//記錄上傳失敗，同時記錄錯誤日志文件地址到數(shù)據(jù)庫，方便用戶查看錯誤信息
writeFailToDB(batchNo,s3Key,fileName);
//刪除文件，防止硬盤爆炸
deleteFile(file)
}

}

至此整個異步上傳功能就完成了，是不是很簡單？（笑）

那么怎么使用呢？更簡單，只需要在service層加入注解即可，頂多就是把錯誤信息return出去。

@Upload(type=UploadType.類型1)
publicStringupload(Listitems){
if(items==null||items.size()==0){
return;
}
//校驗
Stringerror=uploadCheck(items);
if(StringUtils.isNotEmpty){
returnerror;
}
//刪除舊的
deleteAll();
//插入新的
batchInsert(items);
}

結(jié)語

寫了個小輪子提升團隊整體開發(fā)效率感覺真不錯。程序員的最高品質(zhì)就是解放雙手（偷懶？），然后成功的用自己寫的代碼把自己干畢業(yè)。。。。。。

審核編輯：湯梓紅