springboot实现异步导入Excel
需求前言
前文介绍了使用断点续传优化同步导入Excel，即使并发上传分片，依然会因上传耗时对用户体验感不好，故上文只起到抛砖引玉的作用，生产上使用较多的还是异步导入的方式。
异步导入面临的问题
先说说流程：
1、异步导入Excel，后端使用springboot和easyExcel接收处理，并入库，同时异步导入结果会保存到消息通知表；
2、通知前端页面的收件箱有红点表示有导入结果，并且用户点击收件箱查看完，就会取消红点表示已查看过了，下次用户打开前端页面就不会有红点了；
面临的问题：
1、如何实现异步？
2、如何导入大Excel文件避免OOM？
3、异步操作后，如何通知导入结果？
4、如何加快导入效率？
5、将导入结果通知给用户后，如何避免重复通知？
往下看就知道如何解决上述问题了
实现异步
使用spring的Async注解实现异步，但要注意配置线程池，否则在并发导入时，创建多个线程处理异步容易出现OOM，代码如下：
@Configuration
@EnableAsync
public class AsyncConfig implements AsyncConfigurer {
@Override
public Executor getAsyncExecutor() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(5);
executor.setMaxPoolSize(10);
executor.setQueueCapacity(100);
executor.setThreadNamePrefix("ExcelImport-");
executor.initialize();
return executor;
}
}
@Service
@RequiredArgsConstructor
public class ExcelImportService {
private final ImportTaskMapper taskMapper;
private final ImportMessageMapper messageMapper;
private final UserService userService;
@Async
public void asyncImport(String taskId, File excelFile) {
.................
}
}
如何导入大Excel文件避免OOM？
使用阿里的easyExcel工具即可避免，前文也介绍过其操作原理，这里就不多做解释了。
异步操作后，如何通知导入结果？
在导入的大Excel文件，处理数据是异步的，所以需要将处理是成功还是失败的结果保存到一个消息通知表中，供用户访问，示例如csdn有一个消息通知的功能
消息通知表设计：
CREATE TABLE import_message (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
task_id VARCHAR(32) NOT NULL COMMENT '关联任务ID',
user_id BIGINT NOT NULL COMMENT '用户ID',
is_read TINYINT DEFAULT 0 COMMENT '是否已读(0:未读,1:已读)',
create_time DATETIME DEFAULT CURRENT_TIMESTAMP
);
代码如下：
@Service
@RequiredArgsConstructor
public class ExcelImportService {
private final ImportTaskMapper taskMapper;
private final ImportMessageMapper messageMapper;
private final UserService userService;
@Async
public void asyncImport(String taskId, File excelFile) {
ImportTask task = taskMapper.selectById(taskId);
try {
// 使用EasyExcel解析
ImportResult result = EasyExcel.read(excelFile)
.head(ExcelData.class)
.registerReadListener(new DataListener(task))
.sheet().doRead();
// 更新任务状态
task.setStatus(1);
task.setSuccessCount(result.getSuccessCount());
task.setErrorCount(result.getErrorCount());
if (result.hasErrors()) {
task.setErrorFile(generateErrorFile(result));
}
} catch (Exception e) {
task.setStatus(2);
task.setErrorCount(-1); // 表示系统错误
} finally {
taskMapper.updateById(task);
createMessage(task); // 创建通知消息
}
}
private void createMessage(ImportTask task) {
ImportMessage message = new ImportMessage();
message.setTaskId(task.getId());
message.setUserId(task.getUserId());
messageMapper.insert(message);
}
}
// 数据监听器
public class DataListener extends AnalysisEventListener<ExcelData> {
private final ImportTask task;
private final List<ExcelData> cachedData = new ArrayList<>();
private final List<ErrorRow> errors = new ArrayList<>();
@Override
public void invoke(ExcelData data, AnalysisContext context) {
// 数据校验逻辑...
cachedData.add(data);
if (cachedData.size() >= 100) {
saveBatch(cachedData);
cachedData.clear();
}
}
@Override
public void doAfterAllAnalysed(AnalysisContext context) {
if (!cachedData.isEmpty()) {
saveBatch(cachedData);
}
}
private void saveBatch(List<ExcelData> list) {
try {
// 批量入库逻辑
} catch (Exception e) {
// 记录错误行
errors.add(new ErrorRow(/* 行信息 */));
}
}
}
@RestController
@RequestMapping("/api/import")
@RequiredArgsConstructor
public class ImportController {
private final ExcelImportService importService;
// 获取未读消息数量
@GetMapping("/unread-count")
public ResponseEntity<Integer> getUnreadCount(@AuthenticationPrincipal User user) {
int count = messageMapper.countUnread(user.getId());
return ResponseEntity.ok(count);
}
// 标记消息已读
@PostMapping("/mark-read")
public ResponseEntity<?> markAsRead(@RequestBody List<Long> messageIds) {
messageMapper.updateReadStatus(messageIds, 1);
return ResponseEntity.ok().build();
}
// 启动异步导入
@PostMapping
public ResponseEntity<?> startImport(@RequestParam MultipartFile file) {
String taskId = IdUtil.simpleUUID();
File tempFile = saveToTemp(file); // 保存临时文件
ImportTask task = new ImportTask();
task.setId(taskId);
task.setUserId(SecurityUtils.getCurrentUserId());
task.setFileName(file.getOriginalFilename());
task.setStatus(0);
taskMapper.insert(task);
importService.asyncImport(taskId, tempFile);
return ResponseEntity.ok(Map.of("taskId", taskId));
}
}
如何加快导入效率？
避免for循环每次与DB建立一个连接，应该使用MyBatis-Plus的批量插入
List<User> userList = new ArrayList<>();
User user;
for(int i = 0 ;i < 10000; i++) {
user = new User();
user.setUsername("name" + i);
user.setPassword("password" + i);
userList.add(user);
}
saveBatch(userList);
MyBatis-Plus的saveBatch方法默认是使用JDBC的addBatch()和executeBatch()方法实现批量插入。但是部分数据库的JDBC驱动并不支持addBatch()，这样每次插入都会发送一条SQL语句，严重影响了批量插入的性能。设置rewriteBatchedStatements=true后，MyBatis-Plus会重写插入语句，将其合并为一条SQL语句，从而减少网络交互次数，提高批量插入的效率。
将导入结果通知给用户后，如何避免重复通知？
前面也说过，导入Excel的结果会保存到消息表中，前端在登录后，通过访问“标记消息已读”接口标记为已读，至于前端如何发现有消息结果，直接使用定时轮训即可（csdn也是用这种来发现消息通知），前端访问代码如下：
<template>
<!-- 上传组件 -->
<el-upload
action="/api/import"
:show-file-list="false"
:before-upload="beforeUpload"
@success="handleSuccess"
>
<el-button type="primary">导入Excel</el-button>
</el-upload>
<!-- 通知红点 -->
<el-badge :value="unreadCount" :max="99" class="notification-badge">
<el-button icon="bell" @click="showMessages"></el-button>
</el-badge>
<!-- 消息弹窗 -->
<el-dialog v-model="messageVisible">
<el-table :data="messages">
<el-table-column prop="fileName" label="文件名"></el-table-column>
<el-table-column prop="status" label="状态">
<template #default="{row}">
<el-tag :type="statusType(row)">{{ statusText(row) }}</el-tag>
</template>
</el-table-column>
<el-table-column label="操作">
<template #default="{row}">
<el-button @click="downloadError(row)" v-if="row.errorFile">
下载错误报告
</el-button>
</template>
</el-table-column>
</el-table>
</el-dialog>
</template>
<script setup>
import { ref, onMounted } from 'vue'
import { ElMessage } from 'element-plus'
const unreadCount = ref(0)
const messageVisible = ref(false)
const messages = ref([])
// 初始化获取未读数量
onMounted(async () => {
const res = await fetch('/api/import/unread-count')
unreadCount.value = await res.json()
})
// 显示消息弹窗
const showMessages = async () => {
const res = await fetch('/api/import/messages')
messages.value = await res.json()
messageVisible.value = true
// 标记所有消息为已读
const ids = messages.value.map(m => m.id)
await fetch('/api/import/mark-read', {
method: 'POST',
body: JSON.stringify(ids)
})
unreadCount.value = 0
}
// 定时刷新未读数量
setInterval(async () => {
const res = await fetch('/api/import/unread-count')
unreadCount.value = await res.json()
}, 30000)
</script>
优化点
1、通知导入Excel的结果，如果是导入失败，需要知道是什么原因，例如是校验某一行数据的参数不合法之类的，那也要知道是哪一行数据才行，可能会有多行数据有问题，可以通过导出一个Excel的方式，生成导出的Excel路径保存到消息通知表，前端查看红点收件箱即可下载；
2、前端将定时访问优化成websocket，避免长时间轮训，浪费带宽，当然这种优化是看业务场景是否需要，如果需要频繁导入Excel频繁通知导入结果的场景；