wkc/ccdi
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 实现员工信息异步导入Service层方法

Browse Source 
完成功能:
- 新增异步导入方法 importEmployeeAsync,使用@Async注解实现异步处理
- 新增查询导入状态方法 getImportStatus
- 新增查询导入失败记录方法 getImportFailures
- 实现完整的导入逻辑,包括数据分类、批量操作、进度跟踪
- 使用Redis存储导入状态和失败记录,TTL设置为7天
- 支持增量更新模式,批量插入新数据,批量更新已有数据
- 实时更新导入进度到Redis

技术要点:
- 使用RedisTemplate操作Redis,Hash结构存储状态
- 使用importExecutor线程池异步执行导入任务
- 使用UUID生成唯一任务ID
- 使用CompletableFuture包装返回结果
- 批量操作提高性能(saveBatch每500条一批)
- 失败记录只保存到Redis,不保存成功记录

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
This commit is contained in:
wkc
2026-02-06 09:34:08 +08:00
parent e8a4b53a0e
commit 8bdce0adbf
4 changed files with 2566 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

745
doc/plans/2026-02-06-employee-async-import-design.md Normal file
View File

@@ -0,0 +1,745 @@
# 员工信息异步导入功能设计文档
**创建日期**: 2026-02-06
**设计者**: Claude Code
**状态**: 已确认
---
## 一、需求概述
### 1.1 背景
当前员工信息导入功能为同步处理,存在以下问题:
- 导入大量数据时前端等待时间长,用户体验差
- 导入失败记录无法保留和查询
- 未充分利用批量操作提升性能
### 1.2 目标
- 实现异步导入,提升用户体验
- 失败记录存储在Redis中,保留7天,支持查询
- 新数据批量插入,已有数据使用`ON DUPLICATE KEY UPDATE`批量更新
- 以前端页面按钮方式提供失败记录查询功能
### 1.3 核心决策
- **唯一标识**: 柜员号(employeeId)
- **Redis TTL**: 7天
- **进度反馈**: 后台处理 + 完成通知
- **失败数据格式**: JSON对象列表存储
---
## 二、系统架构设计
### 2.1 整体架构
采用**生产者-消费者模式**:
```
前端 → Controller(立即返回) → 异步Service → Redis
↑ ↓
└──── 轮询查询状态 ←─────────┘
```
**核心流程**:
1. 前端提交Excel文件
2. Controller立即返回taskId,不阻塞
3. 异步线程处理导入逻辑
4. 结果实时写入Redis
5. 前端轮询查询导入状态
6. 完成后通知用户,如有失败显示查询按钮
### 2.2 技术选型
| 技术 | 用途 | 说明 |
|------|------|------|
| Spring @Async | 异步处理 | 独立线程池处理导入任务 |
| Redis | 结果存储 | 存储导入状态和失败记录,7天TTL |
| MyBatis Plus | 批量插入 | saveBatch方法批量插入新数据 |
| 自定义SQL | 批量更新 | INSERT ... ON DUPLICATE KEY UPDATE |
| ThreadPoolExecutor | 线程管理 | 核心线程2,最大线程5 |
---
## 三、数据库设计
### 3.1 表结构修改
确保`ccdi_employee`表的`employee_id`字段有UNIQUE约束:
```sql
ALTER TABLE ccdi_employee
ADD UNIQUE KEY uk_employee_id (employee_id);
```
### 3.2 Redis数据结构
**状态信息存储**:
```
Key: import:employee:{taskId}
Type: Hash
TTL: 604800秒 (7天)
Fields:
- status: PROCESSING | SUCCESS | PARTIAL_SUCCESS | FAILED
- totalCount: 总记录数
- successCount: 成功数
- failureCount: 失败数
- startTime: 开始时间戳
- endTime: 结束时间戳
- message: 状态描述
```
**失败记录存储**:
```
Key: import:employee:{taskId}:failures
Type: List (JSON序列化)
TTL: 604800秒 (7天)
Value: [
{
"employeeId": "1234567",
"name": "张三",
"idCard": "110101199001011234",
"deptId": 100,
"phone": "13800138000",
"status": "0",
"hireDate": "2020-01-01",
"errorMessage": "身份证号格式错误"
},
...
]
```
---
## 四、后端实现设计
### 4.1 异步配置
**AsyncConfig.java**:
```java
@Configuration
@EnableAsync
public class AsyncConfig {
@Bean("importExecutor")
public Executor importExecutor() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(2);
executor.setMaxPoolSize(5);
executor.setQueueCapacity(100);
executor.setThreadNamePrefix("import-async-");
executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
executor.initialize();
return executor;
}
}
```
### 4.2 核心VO类
**ImportResultVO** (导入提交结果):
```java
@Data
public class ImportResultVO {
private String taskId;
private String status;
private String message;
}
```
**ImportStatusVO** (导入状态):
```java
@Data
public class ImportStatusVO {
private String taskId;
private String status;
private Integer totalCount;
private Integer successCount;
private Integer failureCount;
private Integer progress;
private Long startTime;
private Long endTime;
private String message;
}
```
**ImportFailureVO** (失败记录):
```java
@Data
public class ImportFailureVO {
private Long employeeId;
private String name;
private String idCard;
private Long deptId;
private String phone;
private String status;
private String hireDate;
private String errorMessage;
}
```
### 4.3 Service层接口
**ICcdiEmployeeService**新增:
```java
/**
* 异步导入员工数据
* @param excelList Excel数据列表
* @param isUpdateSupport 是否更新已存在的数据
* @return CompletableFuture包含导入结果
*/
CompletableFuture<ImportResultVO> importEmployeeAsync(
List<CcdiEmployeeExcel> excelList,
boolean isUpdateSupport
);
/**
* 查询导入状态
* @param taskId 任务ID
* @return 导入状态信息
*/
ImportStatusVO getImportStatus(String taskId);
/**
* 获取导入失败记录
* @param taskId 任务ID
* @return 失败记录列表
*/
List<ImportFailureVO> getImportFailures(String taskId);
```
### 4.4 核心业务逻辑
**数据分类**:
```java
// 1. 批量查询已存在的柜员号
Set<Long> existingIds = employeeMapper.selectBatchIds(
excelList.stream()
.map(CcdiEmployeeExcel::getEmployeeId)
.collect(Collectors.toList())
).stream()
.map(CcdiEmployee::getEmployeeId)
.collect(Collectors.toSet());
// 2. 分类为新数据和更新数据
List<CcdiEmployee> newRecords = new ArrayList<>();
List<CcdiEmployee> updateRecords = new ArrayList<>();
for (CcdiEmployeeExcel excel : excelList) {
CcdiEmployee employee = convertToEntity(excel);
if (existingIds.contains(excel.getEmployeeId())) {
updateRecords.add(employee);
} else {
newRecords.add(employee);
}
}
```
**批量插入**:
```java
if (!newRecords.isEmpty()) {
employeeService.saveBatch(newRecords, 500);
}
```
**批量更新**:
```java
if (!updateRecords.isEmpty() && isUpdateSupport) {
employeeMapper.insertOrUpdateBatch(updateRecords);
}
```
**失败记录处理**:
```java
List<ImportFailureVO> failures = new ArrayList<>();
for (CcdiEmployeeExcel excel : excelList) {
try {
// 验证和导入逻辑
validateAndImport(excel);
} catch (Exception e) {
ImportFailureVO failure = new ImportFailureVO();
BeanUtils.copyProperties(excel, failure);
failure.setErrorMessage(e.getMessage());
failures.add(failure);
}
}
// 存入Redis
if (!failures.isEmpty()) {
String key = "import:employee:" + taskId + ":failures";
redisTemplate.opsForValue().set(
key,
JSON.toJSONString(failures),
7,
TimeUnit.DAYS
);
}
```
### 4.5 Mapper SQL
**CcdiEmployeeMapper.xml**:
```xml
<!-- 批量插入或更新 -->
<insert id="insertOrUpdateBatch" parameterType="java.util.List">
INSERT INTO ccdi_employee
(employee_id, name, dept_id, id_card, phone, hire_date, status,
create_time, update_by, update_time, remark)
VALUES
<foreach collection="list" item="item" separator=",">
(#{item.employeeId}, #{item.name}, #{item.deptId}, #{item.idCard},
#{item.phone}, #{item.hireDate}, #{item.status}, NOW(),
#{item.updateBy}, NOW(), #{item.remark})
</foreach>
ON DUPLICATE KEY UPDATE
name = VALUES(name),
dept_id = VALUES(dept_id),
phone = VALUES(phone),
hire_date = VALUES(hire_date),
status = VALUES(status),
update_by = VALUES(update_by),
update_time = NOW(),
remark = VALUES(remark)
</insert>
```
---
## 五、Controller层API设计
### 5.1 修改导入接口
**接口**: `POST /ccdi/employee/importData`
**改动**:
- 改为立即返回taskId
- 使用异步处理
**响应示例**:
```json
{
"code": 200,
"msg": "导入任务已提交,正在后台处理",
"data": {
"taskId": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
"status": "PROCESSING",
"message": "任务已创建"
}
}
```
### 5.2 新增状态查询接口
**接口**: `GET /ccdi/employee/importStatus/{taskId}`
**Swagger注解**:
```java
@Operation(summary = "查询员工导入状态")
@GetMapping("/importStatus/{taskId}")
public AjaxResult getImportStatus(@PathVariable String taskId)
```
**响应示例**:
```json
{
"code": 200,
"data": {
"taskId": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
"status": "SUCCESS",
"totalCount": 100,
"successCount": 95,
"failureCount": 5,
"progress": 100,
"startTime": 1707225600000,
"endTime": 1707225900000,
"message": "导入完成"
}
}
```
### 5.3 新增失败记录查询接口
**接口**: `GET /ccdi/employee/importFailures/{taskId}`
**参数**:
- `taskId`: 任务ID (路径参数)
- `pageNum`: 页码 (可选,默认1)
- `pageSize`: 每页条数 (可选,默认10)
**响应格式**: TableDataInfo
**Swagger注解**:
```java
@Operation(summary = "查询导入失败记录")
@GetMapping("/importFailures/{taskId}")
public TableDataInfo getImportFailures(
@PathVariable String taskId,
@RequestParam(defaultValue = "1") Integer pageNum,
@RequestParam(defaultValue = "10") Integer pageSize
)
```
---
## 六、前端实现设计
### 6.1 API定义
**api/ccdiEmployee.js**新增:
```javascript
// 查询导入状态
export function getImportStatus(taskId) {
return request({
url: '/ccdi/employee/importStatus/' + taskId,
method: 'get'
})
}
// 查询导入失败记录
export function getImportFailures(taskId, pageNum, pageSize) {
return request({
url: '/ccdi/employee/importFailures/' + taskId,
method: 'get',
params: { pageNum, pageSize }
})
}
```
### 6.2 导入流程优化
**修改handleFileSuccess方法**:
```javascript
handleFileSuccess(response, file, fileList) {
this.upload.isUploading = false;
this.upload.open = false;
if (response.code === 200) {
const taskId = response.data.taskId;
// 显示后台处理提示
this.$notify({
title: '导入任务已提交',
message: '正在后台处理中,处理完成后将通知您',
type: 'info',
duration: 3000
});
// 开始轮询检查状态
this.startImportStatusPolling(taskId);
} else {
this.$modal.msgError(response.msg);
}
}
```
### 6.3 轮询状态检查
```javascript
data() {
return {
// ...其他data
pollingTimer: null
}
},
methods: {
startImportStatusPolling(taskId) {
this.pollingTimer = setInterval(async () => {
const response = await getImportStatus(taskId);
if (response.data.status !== 'PROCESSING') {
clearInterval(this.pollingTimer);
this.handleImportComplete(response.data);
}
}, 2000); // 每2秒轮询一次
},
handleImportComplete(statusResult) {
if (statusResult.status === 'SUCCESS') {
this.$notify({
title: '导入完成',
message: `全部成功!共导入${statusResult.totalCount}条数据`,
type: 'success',
duration: 5000
});
this.getList();
} else if (statusResult.failureCount > 0) {
this.$notify({
title: '导入完成',
message: `成功${statusResult.successCount}条,失败${statusResult.failureCount}条`,
type: 'warning',
duration: 5000
});
// 显示查看失败记录按钮
this.showFailureButton = true;
this.currentTaskId = statusResult.taskId;
// 刷新列表
this.getList();
}
},
beforeDestroy() {
// 组件销毁时清除定时器
if (this.pollingTimer) {
clearInterval(this.pollingTimer);
}
}
}
```
### 6.4 失败记录查询UI
**页面按钮**:
```vue
<el-row :gutter="10" class="mb8">
<!-- 原有按钮... -->
<el-col :span="1.5" v-if="showFailureButton">
<el-button
type="warning"
icon="el-icon-warning"
size="mini"
@click="viewImportFailures"
>
查看导入失败记录 ({{ currentTaskId }})
</el-button>
</el-col>
</el-row>
```
**失败记录对话框**:
```vue
<el-dialog
title="导入失败记录"
:visible.sync="failureDialogVisible"
width="1200px"
append-to-body
>
<el-table :data="failureList" v-loading="failureLoading">
<el-table-column label="姓名" prop="name" />
<el-table-column label="柜员号" prop="employeeId" />
<el-table-column label="身份证号" prop="idCard" />
<el-table-column label="电话" prop="phone" />
<el-table-column label="失败原因" prop="errorMessage" min-width="200" />
</el-table>
<pagination
v-show="failureTotal > 0"
:total="failureTotal"
:page.sync="failureQueryParams.pageNum"
:limit.sync="failureQueryParams.pageSize"
@pagination="getFailureList"
/>
<div slot="footer">
<el-button @click="failureDialogVisible = false">关闭</el-button>
<el-button type="primary" @click="exportFailures">导出失败记录</el-button>
</div>
</el-dialog>
```
**方法实现**:
```javascript
data() {
return {
// 失败记录相关
showFailureButton: false,
currentTaskId: null,
failureDialogVisible: false,
failureList: [],
failureLoading: false,
failureTotal: 0,
failureQueryParams: {
pageNum: 1,
pageSize: 10
}
}
},
methods: {
viewImportFailures() {
this.failureDialogVisible = true;
this.getFailureList();
},
getFailureList() {
this.failureLoading = true;
getImportFailures(
this.currentTaskId,
this.failureQueryParams.pageNum,
this.failureQueryParams.pageSize
).then(response => {
this.failureList = response.rows;
this.failureTotal = response.total;
this.failureLoading = false;
});
},
exportFailures() {
this.download(
'ccdi/employee/exportFailures/' + this.currentTaskId,
{},
`导入失败记录_${new Date().getTime()}.xlsx`
);
}
}
```
---
## 七、错误处理与边界情况
### 7.1 异常场景处理
| 场景 | 处理方式 |
|------|----------|
| 导入文件格式错误 | 上传阶段校验,不创建任务,返回错误提示 |
| 单条数据验证失败 | 记录到Redis失败列表,继续处理其他数据 |
| Redis连接失败 | 记录日志报警,降级处理,返回警告 |
| 线程池队列满 | CallerRunsPolicy,由提交线程执行 |
| 部分成功 | status=PARTIAL_SUCCESS,显示失败记录按钮 |
| 全部失败 | status=FAILED,显示失败记录按钮 |
| taskId不存在 | 返回404,提示任务不存在或已过期 |
### 7.2 数据一致性
- 使用`@Transactional`保证批量操作原子性
- 新数据插入和已有数据更新在同一事务
- 任意步骤失败,整体回滚
- Redis状态更新在事务提交后执行
### 7.3 幂等性
- taskId使用UUID,全局唯一
- 同一文件多次导入产生多个taskId
- 支持查询历史任务状态和失败记录
- 失败记录独立存储,互不影响
### 7.4 性能优化
- 批量插入每批500条,平衡性能和内存
- 使用ON DUPLICATE KEY UPDATE替代先查后更新
- Redis操作使用Pipeline批量执行
- 线程池复用,避免频繁创建销毁
---
## 八、测试策略
### 8.1 单元测试
- 测试数据分类逻辑(新数据vs已有数据)
- 测试批量插入和批量更新
- 测试异常处理和失败记录收集
- 测试Redis读写操作
### 8.2 集成测试
- 测试完整导入流程(提交→处理→查询)
- 测试并发导入多个文件
- 测试Redis异常降级
- 测试线程池满载情况
### 8.3 性能测试
- 100条数据导入时间 < 2秒
- 1000条数据导入时间 < 10秒
- 10000条数据导入时间 < 60秒
- 导入状态查询响应时间 < 100ms
### 8.4 前端测试
- 测试轮询逻辑正确性
- 测试通知显示和关闭
- 测试失败记录分页查询
- 测试组件销毁时清除定时器
---
## 九、实施检查清单
### 9.1 后端任务
- [ ] 创建AsyncConfig配置类
- [ ] 添加数据库UNIQUE约束
- [ ] 创建VO类(ImportResultVO, ImportStatusVO, ImportFailureVO)
- [ ] 实现Service层异步方法
- [ ] 实现Redis状态存储逻辑
- [ ] 实现数据分类和批量操作
- [ ] 编写Mapper XML SQL
- [ ] 添加Controller接口
- [ ] 更新Swagger文档
### 9.2 前端任务
- [ ] 添加API方法定义
- [ ] 修改导入成功处理逻辑
- [ ] 实现轮询状态检查
- [ ] 添加查看失败记录按钮
- [ ] 创建失败记录对话框
- [ ] 实现分页查询失败记录
- [ ] 添加导出失败记录功能
### 9.3 测试任务
- [ ] 编写单元测试用例
- [ ] 生成测试脚本
- [ ] 执行集成测试
- [ ] 进行性能测试
- [ ] 生成测试报告
---
## 十、API文档更新
更新`doc`目录下的接口文档,包含:
- 修改的导入接口说明
- 新增的状态查询接口
- 新增的失败记录查询接口
- 请求/响应示例
---
## 附录
### A. Redis Key命名规范
```
import:employee:{taskId} # 导入状态
import:employee:{taskId}:failures # 失败记录列表
```
### B. 状态枚举
| 状态值 | 说明 | 前端行为 |
|--------|------|----------|
| PROCESSING | 处理中 | 继续轮询 |
| SUCCESS | 全部成功 | 显示成功通知,刷新列表 |
| PARTIAL_SUCCESS | 部分成功 | 显示警告通知,显示失败按钮 |
| FAILED | 全部失败 | 显示错误通知,显示失败按钮 |
### C. 相关文件清单
**后端**:
- `ruoyi-ccdi/src/main/java/com/ruoyi/ccdi/config/AsyncConfig.java`
- `ruoyi-ccdi/src/main/java/com/ruoyi/ccdi/domain/vo/ImportResultVO.java`
- `ruoyi-ccdi/src/main/java/com/ruoyi/ccdi/domain/vo/ImportStatusVO.java`
- `ruoyi-ccdi/src/main/java/com/ruoyi/ccdi/domain/vo/ImportFailureVO.java`
- `ruoyi-ccdi/src/main/java/com/ruoyi/ccdi/service/ICcdiEmployeeService.java`
- `ruoyi-ccdi/src/main/java/com/ruoyi/ccdi/service/impl/CcdiEmployeeServiceImpl.java`
- `ruoyi-ccdi/src/main/java/com/ruoyi/ccdi/mapper/CcdiEmployeeMapper.java`
- `ruoyi-ccdi/src/main/resources/mapper/ccdi/CcdiEmployeeMapper.xml`
- `ruoyi-ccdi/src/main/java/com/ruoyi/ccdi/controller/CcdiEmployeeController.java`
**前端**:
- `ruoyi-ui/src/api/ccdiEmployee.js`
- `ruoyi-ui/src/views/ccdiEmployee/index.vue`
---
**文档版本**: 1.0
**最后更新**: 2026-02-06

1451
doc/plans/2026-02-06-employee-async-import.md Normal file
View File

File diff suppressed because it is too large Load Diff

29
ruoyi-ccdi/src/main/java/com/ruoyi/ccdi/service/ICcdiEmployeeService.java
View File

@@ -6,8 +6,12 @@ import com.ruoyi.ccdi.domain.dto.CcdiEmployeeEditDTO;
import com.ruoyi.ccdi.domain.dto.CcdiEmployeeQueryDTO;
import com.ruoyi.ccdi.domain.excel.CcdiEmployeeExcel;
import com.ruoyi.ccdi.domain.vo.CcdiEmployeeVO;
import com.ruoyi.ccdi.domain.vo.ImportFailureVO;
import com.ruoyi.ccdi.domain.vo.ImportResultVO;
import com.ruoyi.ccdi.domain.vo.ImportStatusVO;
import java.util.List;
import java.util.concurrent.CompletableFuture;
/**
* 员工信息 服务层
@@ -82,4 +86,29 @@ public interface ICcdiEmployeeService {
* @return 结果
*/
String importEmployee(List<CcdiEmployeeExcel> excelList, Boolean isUpdateSupport);
/**
* 异步导入员工数据
*
* @param excelList Excel数据列表
* @param isUpdateSupport 是否更新已存在的数据
* @return CompletableFuture包含导入结果
*/
CompletableFuture<ImportResultVO> importEmployeeAsync(List<CcdiEmployeeExcel> excelList, Boolean isUpdateSupport);
/**
* 查询导入状态
*
* @param taskId 任务ID
* @return 导入状态信息
*/
ImportStatusVO getImportStatus(String taskId);
/**
* 获取导入失败记录
*
* @param taskId 任务ID
* @return 失败记录列表
*/
List<ImportFailureVO> getImportFailures(String taskId);
}

343
ruoyi-ccdi/src/main/java/com/ruoyi/ccdi/service/impl/CcdiEmployeeServiceImpl.java
View File

@@ -1,5 +1,6 @@
package com.ruoyi.ccdi.service.impl;
import com.alibaba.fastjson2.JSON;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.ruoyi.ccdi.domain.CcdiEmployee;
@@ -8,6 +9,9 @@ import com.ruoyi.ccdi.domain.dto.CcdiEmployeeEditDTO;
import com.ruoyi.ccdi.domain.dto.CcdiEmployeeQueryDTO;
import com.ruoyi.ccdi.domain.excel.CcdiEmployeeExcel;
import com.ruoyi.ccdi.domain.vo.CcdiEmployeeVO;
import com.ruoyi.ccdi.domain.vo.ImportFailureVO;
import com.ruoyi.ccdi.domain.vo.ImportResultVO;
import com.ruoyi.ccdi.domain.vo.ImportStatusVO;
import com.ruoyi.ccdi.enums.EmployeeStatus;
import com.ruoyi.ccdi.mapper.CcdiEmployeeMapper;
import com.ruoyi.ccdi.service.ICcdiEmployeeService;
@@ -15,11 +19,15 @@ import com.ruoyi.common.utils.IdCardUtil;
import com.ruoyi.common.utils.StringUtils;
import jakarta.annotation.Resource;
import org.springframework.beans.BeanUtils;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import java.util.ArrayList;
import java.util.List;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
/**
* 员工信息 服务层处理
@@ -33,6 +41,9 @@ public class CcdiEmployeeServiceImpl implements ICcdiEmployeeService {
@Resource
private CcdiEmployeeMapper employeeMapper;
@Resource
private RedisTemplate<String, Object> redisTemplate;
/**
* 查询员工列表
*
@@ -216,6 +227,114 @@ public class CcdiEmployeeServiceImpl implements ICcdiEmployeeService {
}
}
/**
* 异步导入员工数据
*
* @param excelList Excel数据列表
* @param isUpdateSupport 是否更新已存在的数据
* @return CompletableFuture包含导入结果
*/
@Override
@Async("importExecutor")
public CompletableFuture<ImportResultVO> importEmployeeAsync(List<CcdiEmployeeExcel> excelList, Boolean isUpdateSupport) {
String taskId = UUID.randomUUID().toString();
long startTime = System.currentTimeMillis();
// 初始化Redis状态
String statusKey = "import:employee:" + taskId;
Map<String, Object> statusData = new HashMap<>();
statusData.put("taskId", taskId);
statusData.put("status", "PROCESSING");
statusData.put("totalCount", excelList.size());
statusData.put("successCount", 0);
statusData.put("failureCount", 0);
statusData.put("progress", 0);
statusData.put("startTime", startTime);
statusData.put("message", "正在处理...");
redisTemplate.opsForHash().putAll(statusKey, statusData);
redisTemplate.expire(statusKey, 7, TimeUnit.DAYS);
try {
// 执行导入
ImportResult result = doImport(excelList, isUpdateSupport, taskId);
// 更新最终状态
String finalStatus = result.getFailureCount() == 0 ? "SUCCESS" : "PARTIAL_SUCCESS";
updateImportStatus(taskId, finalStatus, result, startTime);
ImportResultVO resultVO = new ImportResultVO();
resultVO.setTaskId(taskId);
resultVO.setStatus(finalStatus);
resultVO.setMessage("导入任务已提交");
return CompletableFuture.completedFuture(resultVO);
} catch (Exception e) {
// 处理异常
Map<String, Object> errorData = new HashMap<>();
errorData.put("status", "FAILED");
errorData.put("message", "导入失败: " + e.getMessage());
errorData.put("endTime", System.currentTimeMillis());
redisTemplate.opsForHash().putAll(statusKey, errorData);
ImportResultVO resultVO = new ImportResultVO();
resultVO.setTaskId(taskId);
resultVO.setStatus("FAILED");
resultVO.setMessage("导入失败: " + e.getMessage());
return CompletableFuture.completedFuture(resultVO);
}
}
/**
* 查询导入状态
*
* @param taskId 任务ID
* @return 导入状态信息
*/
@Override
public ImportStatusVO getImportStatus(String taskId) {
String key = "import:employee:" + taskId;
Boolean hasKey = redisTemplate.hasKey(key);
if (Boolean.FALSE.equals(hasKey)) {
throw new RuntimeException("任务不存在或已过期");
}
Map<Object, Object> statusMap = redisTemplate.opsForHash().entries(key);
ImportStatusVO statusVO = new ImportStatusVO();
statusVO.setTaskId((String) statusMap.get("taskId"));
statusVO.setStatus((String) statusMap.get("status"));
statusVO.setTotalCount((Integer) statusMap.get("totalCount"));
statusVO.setSuccessCount((Integer) statusMap.get("successCount"));
statusVO.setFailureCount((Integer) statusMap.get("failureCount"));
statusVO.setProgress((Integer) statusMap.get("progress"));
statusVO.setStartTime((Long) statusMap.get("startTime"));
statusVO.setEndTime((Long) statusMap.get("endTime"));
statusVO.setMessage((String) statusMap.get("message"));
return statusVO;
}
/**
* 获取导入失败记录
*
* @param taskId 任务ID
* @return 失败记录列表
*/
@Override
public List<ImportFailureVO> getImportFailures(String taskId) {
String key = "import:employee:" + taskId + ":failures";
Object failuresObj = redisTemplate.opsForValue().get(key);
if (failuresObj == null) {
return Collections.emptyList();
}
return JSON.parseArray(JSON.toJSONString(failuresObj), ImportFailureVO.class);
}
/**
* 构建查询条件
*/
@@ -291,4 +410,224 @@ public class CcdiEmployeeServiceImpl implements ICcdiEmployeeService {
vo.setStatusDesc(EmployeeStatus.getDescByCode(employee.getStatus()));
return vo;
}
/**
* 执行导入逻辑
*/
private ImportResult doImport(List<CcdiEmployeeExcel> excelList, Boolean isUpdateSupport, String taskId) {
List<CcdiEmployee> newRecords = new ArrayList<>();
List<CcdiEmployee> updateRecords = new ArrayList<>();
List<ImportFailureVO> failures = new ArrayList<>();
// 批量查询已存在的柜员号
Set<Long> existingIds = getExistingEmployeeIds(excelList);
// 分类数据
for (int i = 0; i < excelList.size(); i++) {
CcdiEmployeeExcel excel = excelList.get(i);
try {
// 转换为AddDTO进行验证
CcdiEmployeeAddDTO addDTO = new CcdiEmployeeAddDTO();
BeanUtils.copyProperties(excel, addDTO);
// 验证数据(支持更新模式)
validateEmployeeDataForImport(addDTO, isUpdateSupport, existingIds);
CcdiEmployee employee = convertToEntity(excel);
if (existingIds.contains(excel.getEmployeeId())) {
if (isUpdateSupport) {
updateRecords.add(employee);
} else {
throw new RuntimeException("柜员号已存在且未启用更新支持");
}
} else {
newRecords.add(employee);
}
// 更新进度
int progress = (int) ((i + 1) * 100.0 / excelList.size());
updateImportProgress(taskId, progress);
} catch (Exception e) {
ImportFailureVO failure = new ImportFailureVO();
BeanUtils.copyProperties(excel, failure);
failure.setErrorMessage(e.getMessage());
failures.add(failure);
}
}
// 批量插入新数据
if (!newRecords.isEmpty()) {
saveBatch(newRecords, 500);
}
// 批量更新已有数据
if (!updateRecords.isEmpty() && isUpdateSupport) {
employeeMapper.insertOrUpdateBatch(updateRecords);
}
// 保存失败记录到Redis
if (!failures.isEmpty()) {
String failuresKey = "import:employee:" + taskId + ":failures";
redisTemplate.opsForValue().set(failuresKey, failures, 7, TimeUnit.DAYS);
}
ImportResult result = new ImportResult();
result.setTotalCount(excelList.size());
result.setSuccessCount(newRecords.size() + updateRecords.size());
result.setFailureCount(failures.size());
return result;
}
/**
* 批量查询已存在的员工ID
*/
private Set<Long> getExistingEmployeeIds(List<CcdiEmployeeExcel> excelList) {
List<Long> employeeIds = excelList.stream()
.map(CcdiEmployeeExcel::getEmployeeId)
.filter(Objects::nonNull)
.collect(Collectors.toList());
if (employeeIds.isEmpty()) {
return Collections.emptySet();
}
List<CcdiEmployee> existingEmployees = employeeMapper.selectBatchIds(employeeIds);
return existingEmployees.stream()
.map(CcdiEmployee::getEmployeeId)
.collect(Collectors.toSet());
}
/**
* 转换为实体对象
*/
private CcdiEmployee convertToEntity(CcdiEmployeeExcel excel) {
CcdiEmployee employee = new CcdiEmployee();
BeanUtils.copyProperties(excel, employee);
return employee;
}
/**
* 批量保存
*/
private void saveBatch(List<CcdiEmployee> list, int batchSize) {
for (int i = 0; i < list.size(); i += batchSize) {
int end = Math.min(i + batchSize, list.size());
List<CcdiEmployee> subList = list.subList(i, end);
for (CcdiEmployee employee : subList) {
employeeMapper.insert(employee);
}
}
}
/**
* 更新导入进度
*/
private void updateImportProgress(String taskId, Integer progress) {
String key = "import:employee:" + taskId;
redisTemplate.opsForHash().put(key, "progress", progress);
}
/**
* 更新导入状态
*/
private void updateImportStatus(String taskId, String status, ImportResult result, Long startTime) {
String key = "import:employee:" + taskId;
Map<String, Object> statusData = new HashMap<>();
statusData.put("status", status);
statusData.put("successCount", result.getSuccessCount());
statusData.put("failureCount", result.getFailureCount());
statusData.put("progress", 100);
statusData.put("endTime", System.currentTimeMillis());
if ("SUCCESS".equals(status)) {
statusData.put("message", "全部成功!共导入" + result.getTotalCount() + "条数据");
} else {
statusData.put("message", "成功" + result.getSuccessCount() + "条,失败" + result.getFailureCount() + "条");
}
redisTemplate.opsForHash().putAll(key, statusData);
}
/**
* 验证员工数据(导入模式,支持更新)
*/
private void validateEmployeeDataForImport(CcdiEmployeeAddDTO addDTO, Boolean isUpdateSupport, Set<Long> existingIds) {
// 验证必填字段
if (StringUtils.isEmpty(addDTO.getName())) {
throw new RuntimeException("姓名不能为空");
}
if (addDTO.getEmployeeId() == null) {
throw new RuntimeException("柜员号不能为空");
}
if (addDTO.getDeptId() == null) {
throw new RuntimeException("所属部门不能为空");
}
if (StringUtils.isEmpty(addDTO.getIdCard())) {
throw new RuntimeException("身份证号不能为空");
}
if (StringUtils.isEmpty(addDTO.getPhone())) {
throw new RuntimeException("电话不能为空");
}
if (StringUtils.isEmpty(addDTO.getStatus())) {
throw new RuntimeException("状态不能为空");
}
// 验证身份证号格式
String idCardError = IdCardUtil.getErrorMessage(addDTO.getIdCard());
if (idCardError != null) {
throw new RuntimeException(idCardError);
}
// 如果柜员号不存在,检查身份证号唯一性
if (!existingIds.contains(addDTO.getEmployeeId())) {
// 检查身份证号唯一性
LambdaQueryWrapper<CcdiEmployee> wrapper = new LambdaQueryWrapper<>();
wrapper.eq(CcdiEmployee::getIdCard, addDTO.getIdCard());
if (employeeMapper.selectCount(wrapper) > 0) {
throw new RuntimeException("该身份证号已存在");
}
}
// 验证状态
if (!"0".equals(addDTO.getStatus()) && !"1".equals(addDTO.getStatus())) {
throw new RuntimeException("状态只能填写'在职'或'离职'");
}
}
/**
* 导入结果内部类
*/
private static class ImportResult {
private Integer totalCount;
private Integer successCount;
private Integer failureCount;
public Integer getTotalCount() {
return totalCount;
}
public void setTotalCount(Integer totalCount) {
this.totalCount = totalCount;
}
public Integer getSuccessCount() {
return successCount;
}
public void setSuccessCount(Integer successCount) {
this.successCount = successCount;
}
public Integer getFailureCount() {
return failureCount;
}
public void setFailureCount(Integer failureCount) {
this.failureCount = failureCount;
}
}
}