package cn.iocoder.yudao.module.mes.service.pro.qrexport;
|
|
import cn.hutool.core.util.StrUtil;
|
import cn.iocoder.yudao.framework.common.util.http.HttpUtils;
|
import cn.iocoder.yudao.module.mes.dal.dataobject.pro.qrexport.MesProQrExportTaskDO;
|
import cn.iocoder.yudao.module.mes.dal.mysql.pro.qrexport.MesProQrExportTaskMapper;
|
import cn.iocoder.yudao.module.mes.enums.pro.MesProQrExportBizTypeEnum;
|
import cn.iocoder.yudao.module.mes.enums.pro.MesProQrExportTaskStatusEnum;
|
import cn.iocoder.yudao.module.mes.framework.qrexport.MesProQrExportProperties;
|
import cn.iocoder.yudao.module.mes.framework.qrexport.MesProQrExportTaskRegistry;
|
import cn.iocoder.yudao.module.mes.framework.qrexport.SystemCpuLoadMonitor;
|
import jakarta.annotation.Resource;
|
import jakarta.servlet.http.HttpServletResponse;
|
import lombok.extern.slf4j.Slf4j;
|
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
|
import org.springframework.stereotype.Service;
|
import org.springframework.util.StreamUtils;
|
import org.springframework.validation.annotation.Validated;
|
|
import java.io.BufferedInputStream;
|
import java.io.File;
|
import java.io.FileInputStream;
|
import java.io.IOException;
|
import java.io.InputStream;
|
import java.nio.file.Files;
|
import java.time.LocalDateTime;
|
import java.util.List;
|
import java.util.concurrent.RejectedExecutionException;
|
|
import static cn.iocoder.yudao.framework.common.exception.util.ServiceExceptionUtil.exception;
|
import static cn.iocoder.yudao.module.mes.enums.ErrorCodeConstants.PRO_BAG_CODE_NOT_EXISTS;
|
import static cn.iocoder.yudao.module.mes.enums.ErrorCodeConstants.PRO_BATCH_NOT_EXISTS;
|
import static cn.iocoder.yudao.module.mes.enums.ErrorCodeConstants.PRO_PALLET_NOT_EXISTS;
|
import static cn.iocoder.yudao.module.mes.enums.ErrorCodeConstants.PRO_QR_EXPORT_SERVER_BUSY;
|
import static cn.iocoder.yudao.module.mes.enums.ErrorCodeConstants.PRO_QR_EXPORT_TASK_BUSY;
|
import static cn.iocoder.yudao.module.mes.enums.ErrorCodeConstants.PRO_QR_EXPORT_TASK_FILE_MISSING;
|
import static cn.iocoder.yudao.module.mes.enums.ErrorCodeConstants.PRO_QR_EXPORT_TASK_NOT_EXISTS;
|
import static cn.iocoder.yudao.module.mes.enums.ErrorCodeConstants.PRO_QR_EXPORT_TASK_STATUS_INVALID;
|
|
/**
|
* MES 二维码导出任务 Service 实现
|
*/
|
@Slf4j
|
@Service
|
@Validated
|
public class MesProQrExportTaskServiceImpl implements MesProQrExportTaskService {
|
|
/**
|
* 非终态状态集合
|
*/
|
private static final List<Integer> UNFINISHED_STATUSES = List.of(
|
MesProQrExportTaskStatusEnum.PENDING.getStatus(),
|
MesProQrExportTaskStatusEnum.RUNNING.getStatus());
|
|
/**
|
* 终态状态集合
|
*/
|
private static final List<Integer> FINAL_STATUSES = List.of(
|
MesProQrExportTaskStatusEnum.FINISHED.getStatus(),
|
MesProQrExportTaskStatusEnum.CANCELED.getStatus(),
|
MesProQrExportTaskStatusEnum.FAILED.getStatus(),
|
MesProQrExportTaskStatusEnum.DOWNLOADED.getStatus());
|
|
@Resource
|
private MesProQrExportTaskMapper taskMapper;
|
@Resource
|
private MesProQrLabelSourceFactory sourceFactory;
|
@Resource
|
private MesProQrExportRunner runner;
|
@Resource
|
private MesProQrExportProperties properties;
|
@Resource
|
private SystemCpuLoadMonitor cpuMonitor;
|
@Resource
|
private MesProQrExportTaskRegistry registry;
|
|
@Resource(name = "mesQrExportExecutor")
|
private ThreadPoolTaskExecutor executor;
|
|
@Override
|
public long countLabels(MesProQrExportBizTypeEnum bizType, String uuid, Long batchId) {
|
return sourceFactory.count(bizType, uuid, batchId);
|
}
|
|
@Override
|
public MesProQrExportTaskDO submit(MesProQrExportBizTypeEnum bizType, String uuid, Long batchId) {
|
// 1. 解析数据源并统计数量：参数不合法或业务数据不存在时，在此直接抛出而不落任务
|
MesProQrLabelSource source = sourceFactory.create(bizType, uuid, batchId);
|
long total = source.count();
|
if (total <= 0) {
|
throw exception(resolveEmptyErrorCode(bizType));
|
}
|
// 2. 并发闸门：同一时间只跑一个大批量任务；小批量同步导出不走这里，因此不会被挡住
|
if (!registry.tryAcquire()) {
|
throw exception(PRO_QR_EXPORT_TASK_BUSY, resolveRunningProgress());
|
}
|
// 一旦任务被交给执行器，并发许可就转由执行线程在收尾时归还；此前的任何失败都必须在这里归还
|
boolean handedOver = false;
|
try {
|
// 3. CPU 闸门：服务器已经很忙时不再叠加导出负载
|
assertCpuAllowSubmit();
|
// 4. 落任务
|
MesProQrExportTaskDO task = MesProQrExportTaskDO.builder()
|
.bizType(bizType.getType())
|
.uuid(uuid)
|
.batchId(batchId)
|
.taskName(source.zipName())
|
.status(MesProQrExportTaskStatusEnum.PENDING.getStatus())
|
.totalCount((int) total)
|
.finishedCount(0)
|
.build();
|
taskMapper.insert(task);
|
// 5. 先注册运行态再投递：避免「已落库但未注册」的窗口被清理任务误回收
|
registry.register(task.getId());
|
handedOver = true;
|
try {
|
executor.execute(() -> runner.run(task.getId(), source));
|
} catch (RejectedExecutionException e) {
|
registry.unregister(task.getId());
|
handedOver = false;
|
taskMapper.updateStatus(task.getId(), MesProQrExportTaskStatusEnum.FAILED.getStatus(),
|
null, null, "导出线程繁忙，请稍后重试", LocalDateTime.now());
|
throw exception(PRO_QR_EXPORT_TASK_BUSY, 0);
|
}
|
return task;
|
} finally {
|
if (!handedOver) {
|
registry.release();
|
}
|
}
|
}
|
|
@Override
|
public MesProQrExportTaskDO getTask(Long id) {
|
MesProQrExportTaskDO task = taskMapper.selectById(id);
|
if (task == null) {
|
throw exception(PRO_QR_EXPORT_TASK_NOT_EXISTS);
|
}
|
return task;
|
}
|
|
@Override
|
public void cancel(Long id) {
|
MesProQrExportTaskDO task = getTask(id);
|
if (isFinalStatus(task.getStatus())) {
|
throw exception(PRO_QR_EXPORT_TASK_STATUS_INVALID);
|
}
|
if (registry.requestCancel(id)) {
|
// 生成线程会自行收尾：删除半成品文件并置为已取消
|
return;
|
}
|
// 不在本进程运行（进程重启后的残留任务）：直接收尾
|
runner.deleteTaskFile(task);
|
taskMapper.updateStatus(id, MesProQrExportTaskStatusEnum.CANCELED.getStatus(),
|
null, null, null, LocalDateTime.now());
|
}
|
|
@Override
|
public void download(Long id, HttpServletResponse response) throws IOException {
|
MesProQrExportTaskDO task = getTask(id);
|
if (!MesProQrExportTaskStatusEnum.FINISHED.getStatus().equals(task.getStatus())) {
|
throw exception(PRO_QR_EXPORT_TASK_STATUS_INVALID);
|
}
|
File file = resolveTaskFile(task);
|
response.setContentType("application/zip");
|
response.setHeader("Content-Disposition",
|
"attachment;filename=" + HttpUtils.encodeUtf8(task.getTaskName() + ".zip"));
|
boolean success = false;
|
try (InputStream in = new BufferedInputStream(new FileInputStream(file))) {
|
StreamUtils.copy(in, response.getOutputStream());
|
response.getOutputStream().flush();
|
success = true;
|
} finally {
|
// 仅在完整传输后删除；中断时保留，交由清理任务在保留期后删除，便于短时间内重试
|
if (success) {
|
deleteAfterDownload(task, file);
|
}
|
}
|
}
|
|
@Override
|
public int cleanupExpiredFiles() {
|
LocalDateTime deadline = LocalDateTime.now()
|
.minusMinutes(Math.max(1, properties.getTempFileRetentionMinutes()));
|
List<MesProQrExportTaskDO> list = taskMapper.selectExpiredFinalList(FINAL_STATUSES, deadline);
|
int count = 0;
|
for (MesProQrExportTaskDO task : list) {
|
if (StrUtil.isBlank(task.getFilePath())) {
|
continue;
|
}
|
runner.deleteTaskFile(task);
|
taskMapper.clearFilePath(task.getId());
|
count++;
|
}
|
return count;
|
}
|
|
@Override
|
public int recycleStaleTasks() {
|
List<MesProQrExportTaskDO> list = taskMapper.selectUnfinishedList(UNFINISHED_STATUSES);
|
int count = 0;
|
for (MesProQrExportTaskDO task : list) {
|
if (registry.isRunning(task.getId())) {
|
continue;
|
}
|
runner.deleteTaskFile(task);
|
taskMapper.clearFilePath(task.getId());
|
taskMapper.updateStatus(task.getId(), MesProQrExportTaskStatusEnum.FAILED.getStatus(),
|
null, null, "任务因服务重启或异常中断而终止，请重新导出", LocalDateTime.now());
|
count++;
|
}
|
return count;
|
}
|
|
// ==================== 私有方法 ====================
|
|
/**
|
* 校验系统 CPU 是否允许提交大批量导出
|
* <p>
|
* 取不到 CPU 指标时按放行处理：不能因为读不到监控值就阻断业务。
|
*/
|
private void assertCpuAllowSubmit() {
|
if (!cpuMonitor.isAvailable()) {
|
return;
|
}
|
double load = cpuMonitor.getCpuLoad();
|
double threshold = properties.getCpuRejectThreshold();
|
if (load >= threshold) {
|
throw exception(PRO_QR_EXPORT_SERVER_BUSY,
|
Math.round(load * 100), Math.round(threshold * 100));
|
}
|
}
|
|
/**
|
* 数据为空时的错误码：按业务类型给出更准确的提示
|
*/
|
private cn.iocoder.yudao.framework.common.exception.ErrorCode resolveEmptyErrorCode(
|
MesProQrExportBizTypeEnum bizType) {
|
return switch (bizType) {
|
case PALLET_LABEL, PALLET_BUSINESS -> PRO_PALLET_NOT_EXISTS;
|
case BATCH_LABEL, BATCH_BUSINESS -> PRO_BATCH_NOT_EXISTS;
|
default -> PRO_BAG_CODE_NOT_EXISTS;
|
};
|
}
|
|
/**
|
* 正在运行任务的进度百分比，用于「已有任务进行中」的提示
|
*/
|
private int resolveRunningProgress() {
|
List<MesProQrExportTaskDO> tasks = taskMapper.selectUnfinishedList(UNFINISHED_STATUSES);
|
return tasks.stream().findFirst().map(this::calcProgress).orElse(0);
|
}
|
|
private int calcProgress(MesProQrExportTaskDO task) {
|
Integer total = task.getTotalCount();
|
if (total == null || total <= 0) {
|
return 0;
|
}
|
int finished = task.getFinishedCount() == null ? 0 : task.getFinishedCount();
|
return (int) Math.min(100L, (long) finished * 100 / total);
|
}
|
|
private boolean isFinalStatus(Integer status) {
|
return status != null && FINAL_STATUSES.contains(status);
|
}
|
|
/**
|
* 定位任务的临时 ZIP，文件缺失时顺手清空路径避免后续反复失败
|
*/
|
private File resolveTaskFile(MesProQrExportTaskDO task) {
|
String filePath = task.getFilePath();
|
if (StrUtil.isBlank(filePath)) {
|
throw exception(PRO_QR_EXPORT_TASK_FILE_MISSING);
|
}
|
File file = new File(filePath);
|
if (!file.isFile()) {
|
taskMapper.clearFilePath(task.getId());
|
throw exception(PRO_QR_EXPORT_TASK_FILE_MISSING);
|
}
|
return file;
|
}
|
|
/**
|
* 下载完成即删除临时文件：单次导出可达上 GB，留着会迅速把磁盘吃满
|
*/
|
private void deleteAfterDownload(MesProQrExportTaskDO task, File file) {
|
try {
|
Files.deleteIfExists(file.toPath());
|
} catch (IOException e) {
|
log.warn("[deleteAfterDownload][删除临时文件失败 {}]", file.getAbsolutePath(), e);
|
}
|
taskMapper.clearFilePath(task.getId());
|
taskMapper.updateStatus(task.getId(), MesProQrExportTaskStatusEnum.DOWNLOADED.getStatus(),
|
null, null, null, null);
|
}
|
|
}
|
