快捷導(dǎo)航

Java實(shí)現(xiàn)Excel通用異步導(dǎo)出框架方式

更新時(shí)間：2025年10月23日 10:08:00 作者：假裝開心的撒旦

我最近要做一個(gè)需求,就是設(shè)計(jì)一個(gè)通用的異步導(dǎo)出框架,如果數(shù)據(jù)量小于5W那么直接進(jìn)行導(dǎo)出,數(shù)據(jù)大于5W則創(chuàng)建異步導(dǎo)出任務(wù),并且前端可以刷新該任務(wù)進(jìn)行,然后后面所有相關(guān)功能的導(dǎo)出都使用此框架進(jìn)行操作,思來想去實(shí)現(xiàn)后,想著該需求應(yīng)該符合大部分實(shí)際工作的場景,特此分享！

1、審題

1.小于5W直接導(dǎo)出，大于5W則需要?jiǎng)?chuàng)建任務(wù)

2.異步導(dǎo)出

3.框架

分析1

小于5W直接導(dǎo)出，所以我們框架需要從使用者手里知道本次導(dǎo)出的數(shù)據(jù)總量是多少

分析2

既然是異步導(dǎo)出，所以我們不能讓主線程去執(zhí)行導(dǎo)出操作，所以主線程只管將導(dǎo)出任務(wù)的相關(guān)信息交給其他線程即可

分析3

既然是框架，那么我們不能關(guān)心具體的導(dǎo)出實(shí)現(xiàn)邏輯，例如怎么獲取表的數(shù)據(jù)，怎么查詢總量，但是我們需要關(guān)心一些通用的邏輯，例如創(chuàng)建文件，使用導(dǎo)出組件進(jìn)行導(dǎo)出，等等

好了，現(xiàn)在開始實(shí)現(xiàn)。

2、創(chuàng)建任務(wù)表，記錄任務(wù)的狀態(tài)，進(jìn)度，以及任務(wù)的類型

-- public.export_task definition

-- Drop table

-- DROP TABLE public.export_task;

CREATE TABLE public.export_task (
	id text NOT NULL,
	export_key text NOT NULL,  -- 任務(wù)類型
	params text NOT NULL, -- 任務(wù)所需要的參數(shù)
	status int2 NOT NULL, -- 任務(wù)狀態(tài)
	progress text NOT NULL, -- 任務(wù)進(jìn)度
	create_user text NOT NULL, -- 創(chuàng)建人
	gmt_create text NOT NULL, -- 創(chuàng)建時(shí)間
	single_export_num int4 NOT NULL DEFAULT 50000, -- 單次導(dǎo)出條數(shù)
	file_name text NOT NULL, -- 導(dǎo)出的文件名稱
	file_path text NULL, -- 導(dǎo)出的文件路徑
	complete_time text NULL, -- 導(dǎo)出完成時(shí)間
	CONSTRAINT export_task_pkey PRIMARY KEY (id) -- 主鍵
);

-- 索引
CREATE INDEX export_task_export_key_index ON public.export_task (export_key text_ops);

3、有了任務(wù)表后，就可以開始考慮怎么執(zhí)行任務(wù)了

我的思路是：新增任務(wù) -> 必要校驗(yàn)完成 -> 通過任務(wù)類型(也就是上表中的export_key)找到需要處理的處理類 -> 調(diào)用獲取總量的方法，判斷總量是否大于5W

總量大于5W -> 將當(dāng)前任務(wù)插入數(shù)據(jù)庫 -> 然后將當(dāng)前任務(wù)丟給隊(duì)列，等待被消費(fèi)處理
總量小于5W -> 調(diào)用處理類的導(dǎo)出方法直接進(jìn)行導(dǎo)出

上代碼：controller類，新增任務(wù)（公司代碼里面有記錄日志的邏輯你們不能復(fù)用）

 @POST
@Path("/task-manage")
@Produces({ MediaType.APPLICATION_JSON })
@ApiOperation(value = "新建導(dǎo)出任務(wù)", notes = "新建導(dǎo)出任務(wù)", httpMethod = "POST", tags = {"新建導(dǎo)出任務(wù)"})
public void taskCreate(TaskInfoDTO condition, @Context HttpServletResponse response) throws Exception {
ServiceBaseInfoBean infoBean = new ServiceBaseInfoBean(UserThreadLocal.getuserName(),
UserThreadLocal.getRemoteHost(),UserThreadLocal.getLanguageOption());

ExportTaskMgr.Operation operation = ExportTaskMgr.Operation.DIRECT_EXPORT;
ExportTaskMgr.initOperlogBean(infoBean, operation);
String detailZh = String.format(operation.getDetailZh(), condition.getFileName());
String detailEn = String.format(operation.getDetailEn(), condition.getFileName());
try{
condition.checkFileName();
Long allDataTotal = taskCommandApplicationService.exportOrCreateTask(condition, response);
if(allDataTotal > GlobalConstants.EXPORT_DEFAULT_COUNT){
operation = ExportTaskMgr.Operation.CREATE_EXPORT_TASK;
ExportTaskMgr.initOperlogBean(infoBean, operation);
}
detailZh = String.format(operation.getDetailZh(), condition.getFileName());
detailEn = String.format(operation.getDetailEn(), condition.getFileName());
ExportTaskMgr.refreshSuccessDetail(infoBean,operation, detailZh,detailEn);
}
catch (Exception e){
ResponseUtils.resetContentType(response);
ExportTaskMgr.refreshFailDetail(infoBean,operation, detailZh,detailEn,e);
throw e;
}
finally {
// 發(fā)送操作記錄日志
String operlogMsg = JSON.toJSONString(infoBean.getOperlogBean());
msgSenderService.sendMsgAsync(OperlogBean.IMOP_LOG_MANAGE_TOPIC, operlogMsg);
}
}

service: 通過當(dāng)前任務(wù)類型調(diào)用工廠找到處理類然后... 看注釋把

 @Override
public Long exportOrCreateTask(TaskInfoDTO condition, HttpServletResponse response) throws Exception {
BiFunction<ExportTaskPO, Map<String, Future<?>>, ExportHandler> handlerFuc = ExportHandlerFactory
.getHandlerByExportKey(condition.getExportKey());
if(null == handlerFuc){
throw new MonitorException(-1, I18nConstants.UNKNOWN_EXPORT_TYPE);
}

// 新建任務(wù)相關(guān)信息 暫不入庫
ExportTaskPO taskPO = new ExportTaskPO();
taskPO.setId(UUID.randomUUID().toString());
taskPO.setProgress(GlobalConstants.DEFAULT_VALUE);
taskPO.setStatus(ExportStatusEnums.INIT.getStatus());
taskPO.setCreateUser(UserThreadLocal.getuserName());
taskPO.setFileName(condition.getFileName());
taskPO.setExportKey(condition.getExportKey());
taskPO.setParams(condition.getParams());
taskPO.setGmtCreate(dateTimeService.getCurrentTime());
taskPO.setSingleExportNum(GlobalConstants.EXPORT_DEFAULT_COUNT);
taskPO.setFilePath(GlobalConstants.DEFAULT_VALUE);
taskPO.setCompleteTime(GlobalConstants.DEFAULT_VALUE);

// 構(gòu)建handler
ExportHandler handler = handlerFuc.apply(taskPO, new HashMap<>());

// 獲取total 如果大于單次導(dǎo)出條數(shù)則創(chuàng)建任務(wù) 然后后臺(tái)異步導(dǎo)出
Long total = handler.getTotal();
if(total > taskPO.getSingleExportNum()){
exportTaskRepository.save(taskPO);
ExportHandlerFactory.addTaskToQue(new ArrayList<>(Collections.singletonList(taskPO)));
}
// 滿足單次導(dǎo)出 則直接導(dǎo)出
else {
handler.directExport(response);
}

return total;
}

工廠類的具體實(shí)現(xiàn):

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.*;
import java.util.concurrent.*;
import java.util.function.BiFunction;

@Slf4j
@Component
public class ExportHandlerFactory {
private static final ExportTaskRepository EXPORT_TASK_REPOSITORY = SpringUtil.getBean(ExportTaskRepository.class);
/**
 * 線程池  3個(gè)線程消息導(dǎo)出任務(wù)
 */
private static final ExecutorService THREAD_POOL = Executors.newFixedThreadPool(3);

/**
 * 記錄執(zhí)行中的線程
 */
private static final Map<String, Future<?>> RUNNING_THREAD = new ConcurrentHashMap<>();

/**
 * 任務(wù)隊(duì)列 輪訓(xùn)消費(fèi)此隊(duì)列
 */
private static final Queue<ExportTaskPO> TASK_QUE = new ConcurrentLinkedQueue<>();

/**
 * 定義處理器 例如DEMO類型導(dǎo)出任務(wù)由DEMO_HANDLER進(jìn)行處理
 */
private static final Map<String, BiFunction<ExportTaskPO,Map<String, Future<?>>,ExportHandler>> HANDLER_MAP = new ConcurrentHashMap<>();
static {
HANDLER_MAP.put(ExportTypeEnums.DEMO_EXPORT.getExportKey(), DemoExportHandler::new);
}

@PostConstruct
@SuppressWarnings("InfiniteLoopStatement")
public static void init() {
Runnable runnable = () -> {
while (true) {
try {
// 每隔50ms拉取一次導(dǎo)出任務(wù) 如果已經(jīng)消費(fèi)完 則不再執(zhí)行
TimeUnit.MILLISECONDS.sleep(50L);
ExportTaskPO taskInfo = TASK_QUE.poll();
if(taskInfo == null) {
continue;
}

// 根據(jù)導(dǎo)出類型key 找到handler 開始處理
BiFunction<ExportTaskPO,Map<String, Future<?>>,ExportHandler> biFunction =
HANDLER_MAP.get(taskInfo.getExportKey());
log.info("begin consume taskInfo={},biFunction={}",taskInfo,biFunction);

if(biFunction != null) {
// 設(shè)置任務(wù)信息 以及記錄運(yùn)行狀態(tài)的map 得到處理者
ExportHandler exportHandler = biFunction.apply(taskInfo, RUNNING_THREAD);
log.info("begin consume exportHandler={}",exportHandler);
// 執(zhí)行任務(wù)
Future<Boolean> exceuteFuture = THREAD_POOL.submit(exportHandler);
// 將正在執(zhí)行的任務(wù)放入map中
RUNNING_THREAD.put(taskInfo.getId(),exceuteFuture);
}
}
catch (Exception e) {
log.error("consume export task error.",e);
}
}
};

Executors.newFixedThreadPool(1).execute(runnable);
}

public static void stopTaskByTaskId(String taskId){
// 從運(yùn)行中的map 拿到執(zhí)行中的線程
Future<?> future = RUNNING_THREAD.get(taskId);

// 如果存在  直接停止運(yùn)行
if(future != null){
try{
future.cancel(Boolean.TRUE);
}
catch (Exception e){
log.debug("destroy thread error",e);
}
finally {
RUNNING_THREAD.remove(taskId);
}
}
}

public static BiFunction<ExportTaskPO,Map<String, Future<?>>,ExportHandler> getHandlerByExportKey(String ExportKey){
// 根據(jù)導(dǎo)出類型key 找到handler 開始處理
return HANDLER_MAP.get(ExportKey);
}


public static void addTaskToQue(List<ExportTaskPO> needExecuteTask){
if(CollectionUtils.isNotEmpty(needExecuteTask)){
TASK_QUE.addAll(needExecuteTask);
}
}

public static void initTaskWhenAppRun(){
try{
List<ExportTaskPO> needExecuteTask = EXPORT_TASK_REPOSITORY
.lambdaQuery()
.in(ExportTaskPO::getStatus, Arrays.asList(ExportStatusEnums.INIT.getStatus(), ExportStatusEnums.RUNNING.getStatus()))
.orderByAsc(ExportTaskPO::getGmtCreate)
.list();
log.info("initAllWaitExecuteTask={}",needExecuteTask);
addTaskToQue(needExecuteTask);
}
catch (Exception e){
log.error("initAllWaitExecuteTask error",e);
}
}
}

工廠類定了輪訓(xùn)消費(fèi)，定義了哪個(gè)類型的任務(wù)由哪個(gè)處理器處理，定義了執(zhí)行中的任務(wù)以便刪除任務(wù)時(shí)立刻任務(wù)執(zhí)行，等等

工廠類相當(dāng)于總管，但是具體的實(shí)現(xiàn)是由ExportHandler這個(gè)類來完成處理，那么這個(gè)類又做了什么些事情呢~~~

import javax.servlet.http.HttpServletResponse;
import java.io.BufferedOutputStream;
import java.io.File;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.net.URLEncoder;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.util.*;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import java.util.zip.ZipEntry;
import java.util.zip.ZipOutputStream;

@Slf4j
public abstract class ExportHandler implements Callable<Boolean> {
public final ExportTaskPO taskInfo;
private final Map<String, Future<?>> runMap;
private static final ExportTaskRepository EXPORT_TASK_REPOSITORY = SpringUtil.getBean(ExportTaskRepository.class);
private static final FileConfig FILE_CONFIG = SpringUtil.getBean(FileConfig.class);
private static final DateTimeService DATE_TIME_SERVICE = SpringUtil.getBean(DateTimeService.class);

public ExportHandler(ExportTaskPO taskInfo,Map<String,Future<?>> runMap){
this.taskInfo = taskInfo;
this.runMap = runMap;
}

/**
 * 數(shù)據(jù)小于5w條時(shí)直接導(dǎo)出
 * @param response 導(dǎo)出response
 * @throws Exception 異常時(shí)直接拋出
 */
public void directExport(HttpServletResponse response) throws Exception {
FileUtils.checkFileName(taskInfo.getFileName());

// 設(shè)置響應(yīng)類型以及文件名
response.reset();
response.setContentType(GlobalConstants.CONTENT_TYPE_XLSX);
String encodeName = URLEncoder.encode(taskInfo.getFileName(),GlobalConstants.ENCODER_UTF8);
response.setHeader(GlobalConstants.CONTENT_DISPOSITION_KEY, GlobalConstants.CONTENT_DISPOSITION_FILE_PREFIX.concat(encodeName));

// 寫入數(shù)據(jù)
EasyExcel
.write(response.getOutputStream())
.excelType(ExcelTypeEnum.XLSX)
.charset(StandardCharsets.UTF_8)
.sheet(FileNameUtil.mainName(taskInfo.getFileName()))
.head(getHeader())
.doWrite(getData(null,null));
}

/**
 * 默認(rèn)的數(shù)據(jù)大于5W時(shí)的異步導(dǎo)出處理邏輯 默認(rèn)單sheet頁 簡單實(shí)現(xiàn)更新任務(wù)進(jìn)度
 * 如果有復(fù)雜的多sheet頁導(dǎo)出 請(qǐng)?jiān)谀愕膆andler里面重寫此方法
 */
public void asyncExport(){
ZipOutputStream zipOut = null;
File file = null;
// 執(zhí)行導(dǎo)出邏輯
try {
// 任務(wù)執(zhí)行中
log.info("begin export,taskId:{}",taskInfo.getId());
updateTaskInfo(BigDecimal.ZERO + GlobalConstants.PER_CENT,ExportStatusEnums.RUNNING.getStatus(),null);

// 檢查文件名稱 獲取沒有后綴的文件名
FileUtils.checkFileName(taskInfo.getFileName());
String fileMainName = FileNameUtil.mainName(taskInfo.getFileName());

// 先在服務(wù)器生成文件，得到輸出流
String filePath = FileUtils.getDateFileName(fileMainName, GlobalConstants.FILE_TYPE_ZIP);
file = FileUtils.getAbsoluteFile(FILE_CONFIG.getExportPath(), filePath);
zipOut = new ZipOutputStream(new BufferedOutputStream(Files.newOutputStream(file.toPath())));

// 根據(jù)總量數(shù)據(jù)以及分頁每條的數(shù)據(jù)  得到要導(dǎo)出多少次并且遍歷 開始導(dǎo)入數(shù)據(jù)
long exportCount = calculateExportCount(getTotal(),taskInfo.getSingleExportNum());

// 遍歷導(dǎo)出的次數(shù) 每次生成一個(gè)excel文件
dealExport(exportCount,zipOut);

// 導(dǎo)出完成后更新文件路徑，完成時(shí)間以及狀態(tài)
taskInfo.setFilePath(filePath);
updateTaskInfo(null,ExportStatusEnums.SUCCESS.getStatus(),DATE_TIME_SERVICE.getCurrentTime());
log.info("end export,taskId:{}",taskInfo.getId());
}
// 異常打印異常信息 更新任務(wù)進(jìn)度為-- 狀態(tài)為失敗  如果生成了文件 還需要把文件刪除
catch (Exception e) {
log.error("export error,taskId:{}",taskInfo.getId(),e);
updateTaskInfo(GlobalConstants.DEFAULT_VALUE,ExportStatusEnums.FAILED.getStatus(),DATE_TIME_SERVICE.getCurrentTime());

if(file != null){
boolean delete = file.delete();
log.info("export error,taskId:{},file delete：{}",taskInfo.getId(),delete);
}
}
// 最終從執(zhí)行中的map里移除當(dāng)前任務(wù)
finally {
if(zipOut != null){
FileUtils.safeClose(zipOut);
}
}
}

private void dealExport(long exportCount,ZipOutputStream zipOut) throws Exception{
// 遍歷導(dǎo)出的次數(shù) 每次生成一個(gè)excel文件
for (long i = 1; i <= exportCount; i++) {
// 新建zip中的其中一個(gè)文件
String eachName = String.format("%02d", i) + ExcelTypeEnum.XLSX.getValue();
zipOut.putNextEntry(new ZipEntry(new String(eachName.getBytes(StandardCharsets.UTF_8))));

// 新建好了后往zip寫入數(shù)據(jù)
List<List<String>> singleData = getData(i, taskInfo.getSingleExportNum());
EasyExcel
.write(zipOut)
.autoCloseStream(Boolean.FALSE)
.excelType(ExcelTypeEnum.XLSX)
.charset(StandardCharsets.UTF_8)
.sheet()
.head(getHeader())
.doWrite(singleData);

// 關(guān)閉entry 代表zip當(dāng)前的其中一個(gè)文件已經(jīng)寫入結(jié)束
zipOut.closeEntry();

// 更新進(jìn)度
updateTaskInfo(calculateProgress(i,exportCount),ExportStatusEnums.RUNNING.getStatus(),null);
singleData.clear();
}
}

private void updateTaskInfo(String progress,Integer status,String completeTime){
Optional.ofNullable(progress).ifPresent(taskInfo::setProgress);
Optional.ofNullable(status).ifPresent(taskInfo::setStatus);
Optional.ofNullable(completeTime).ifPresent(taskInfo::setCompleteTime);
EXPORT_TASK_REPOSITORY.updateById(taskInfo);
}

private Long calculateExportCount(Long total,Long singleExportNum){
long exportCount = total / singleExportNum;
long mod = total % singleExportNum;
if(mod != 0){
exportCount += 1;
}

return exportCount;
}

private String calculateProgress(Long thisCount,Long exportCount){
BigDecimal progress = BigDecimal.valueOf(thisCount)
.divide(BigDecimal.valueOf(exportCount), 2, RoundingMode.HALF_UP)
.multiply(new BigDecimal(GlobalConstants.PROGRESS_OVER))
.setScale(BigDecimal.ZERO.intValue(),RoundingMode.HALF_UP);

return progress + GlobalConstants.PER_CENT;
}

@Override
public Boolean call(){
try{
asyncExport();
return Boolean.TRUE;
}
catch (Exception e){
log.error("export task:{} error",taskInfo.getId(),e);
return Boolean.FALSE;
}
finally {
if(taskInfo != null && runMap != null){
runMap.remove(taskInfo.getId());
}
}
}

/**
 * 需要實(shí)現(xiàn)的查詢total的方法
 */
public abstract Long getTotal() throws Exception;

/**
 * 需要實(shí)現(xiàn)的組裝表頭的方法
 */
public abstract List<List<String>> getHeader();

/**
 * 需要實(shí)現(xiàn)的組裝表體數(shù)據(jù)的方法
 */
public abstract List<List<String>> getData(Long pageNo,Long pageSize) throws Exception;
}

噢，原來是一個(gè)抽象類，里面把需要用到的一些邏輯，例如生成文件，導(dǎo)入等都已經(jīng)實(shí)現(xiàn)了，那么所有用到該架子導(dǎo)出的人只需要繼承當(dāng)前類，實(shí)現(xiàn)里面的getTotal getHeader getData三個(gè)方法就可以啦，比如我們工廠類里面寫的這段代碼，demo_export由DemoExportHandler進(jìn)行處理，現(xiàn)在最后來看看這個(gè)DemoExportHandler怎么實(shí)現(xiàn)的把

static {
HANDLER_MAP.put(ExportTypeEnums.DEMO_EXPORT.getExportKey(), DemoExportHandler::new);
}

DemoExportHandler代碼：

import java.util.*;
import java.util.concurrent.Future;
import java.util.stream.Collectors;

@Slf4j
public class DemoExportHandler extends ExportHandler {

private static final StandardPointRepository STANDARD_POINT_REPOSITORY = SpringUtil.getBean(StandardPointRepository.class);
private static final JsonService JSON_SERVICE = SpringUtil.getBean(JsonService.class);
private static final I18nUtilService I_18_N_UTIL_SERVICE = SpringUtil.getBean(I18nUtilService.class);

public DemoExportHandler(ExportTaskPO taskInfo, Map<String, Future<?>> runMap) {
super(taskInfo, runMap);
}

@Override
@SuppressWarnings("unchecked")
public Long getTotal() throws Exception{

// 導(dǎo)出參數(shù) 反序列化為自己想要的類型的對(duì)象
Map<String,Object> paramMap = getParamObj(taskInfo.getParams());

// 取出自己想要的數(shù)據(jù)，根據(jù)條件進(jìn)行導(dǎo)出
List<String> pointList = (List<String>)paramMap.get("pointList");

// 此處demo使用 total請(qǐng)根據(jù)自己的邏輯查詢
List<PointInfoBean> pointBeans = STANDARD_POINT_REPOSITORY.selectPointByIdList(pointList);
return (long) pointBeans.size();
}

@Override
public List<List<String>> getHeader() {
// 組裝表頭
return Arrays.asList(Collections.singletonList("測(cè)點(diǎn)id"),
Collections.singletonList("測(cè)點(diǎn)名稱"),Collections.singletonList("測(cè)點(diǎn)單位"));
}

@Override
@SuppressWarnings("unchecked")
public List<List<String>> getData(Long pageNo, Long pageSize) throws Exception {
// 模擬執(zhí)行導(dǎo)出邏輯執(zhí)行20S
Thread.sleep(20000L);

// 導(dǎo)出參數(shù) 反序列化為自己想要的類型的對(duì)象
Map<String,Object> paramMap = getParamObj(taskInfo.getParams());

// 取出自己想要的數(shù)據(jù)，根據(jù)條件進(jìn)行導(dǎo)出
String lang = (String)paramMap.get("language-option");
List<String> pointList = (List<String>)paramMap.get("pointList");

// 查詢數(shù)據(jù)
List<PointInfoBean> pointBeans = STANDARD_POINT_REPOSITORY.selectPointByIdList(pointList);
List<List<String>> allData = pointBeans
.stream()
.map(bean -> Arrays.asList(bean.getId(), I_18_N_UTIL_SERVICE
.getMapFieldByLanguageOption(bean.getNameI18n(), lang), bean.getUnit()))
.collect(Collectors.toList());

// 分頁或者不分頁返回
if(pageNo == null || pageSize == null){
return allData;
}

return PageUtils.getPageList(allData,pageNo.intValue(),pageSize.intValue());
}

@SuppressWarnings("unchecked")
private Map<String,Object> getParamObj(String params) throws UedmException {
Map<String,Object> paramMap = JSON_SERVICE.jsonToObject(params,Map.class);
log.debug("getParamObj params={},obj={}",params,paramMap);

return paramMap;
}
}