package cn.iocoder.yudao.module.mes.service.pro.bagcode;
|
|
import cn.hutool.core.util.StrUtil;
|
import cn.iocoder.yudao.module.mes.dal.dataobject.pro.bagcode.MesProBagCodeGenerateTaskDO;
|
import cn.iocoder.yudao.module.mes.dal.mysql.pro.bagcode.MesProBagCodeGenerateTaskMapper;
|
import cn.iocoder.yudao.module.mes.enums.pro.MesProBagCodeGenerateTaskStatusEnum;
|
import cn.iocoder.yudao.module.mes.framework.bagcode.MesProBagCodeGenerateTaskRegistry;
|
import jakarta.annotation.Resource;
|
import lombok.extern.slf4j.Slf4j;
|
import org.springframework.dao.DuplicateKeyException;
|
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
|
import org.springframework.stereotype.Service;
|
import org.springframework.validation.annotation.Validated;
|
|
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_GENERATE_TASK_BUSY;
|
import static cn.iocoder.yudao.module.mes.enums.ErrorCodeConstants.PRO_BAG_CODE_GENERATE_TASK_NOT_EXISTS;
|
import static cn.iocoder.yudao.module.mes.enums.ErrorCodeConstants.PRO_BAG_CODE_GENERATE_TASK_STATUS_INVALID;
|
|
/**
|
* 袋码批量生成任务 Service 实现
|
*/
|
@Slf4j
|
@Service
|
@Validated
|
public class MesProBagCodeGenerateTaskServiceImpl implements MesProBagCodeGenerateTaskService {
|
|
/**
|
* 非终态状态集合
|
*/
|
private static final List<Integer> UNFINISHED_STATUSES = List.of(
|
MesProBagCodeGenerateTaskStatusEnum.PENDING.getStatus(),
|
MesProBagCodeGenerateTaskStatusEnum.RUNNING.getStatus(),
|
MesProBagCodeGenerateTaskStatusEnum.CANCELING.getStatus());
|
|
/**
|
* 终态状态集合
|
*/
|
private static final List<Integer> FINAL_STATUSES = List.of(
|
MesProBagCodeGenerateTaskStatusEnum.FINISHED.getStatus(),
|
MesProBagCodeGenerateTaskStatusEnum.CANCELED.getStatus(),
|
MesProBagCodeGenerateTaskStatusEnum.FAILED.getStatus());
|
|
@Resource
|
private MesProBagCodeGenerateTaskMapper taskMapper;
|
@Resource
|
private MesProBagCodeGenerateRunner runner;
|
@Resource
|
private MesProBagCodeGenerateCleaner cleaner;
|
@Resource
|
private MesProBagCodeGenerateTaskRegistry registry;
|
|
@Resource(name = "mesBagCodeGenerateExecutor")
|
private ThreadPoolTaskExecutor executor;
|
|
@Override
|
public MesProBagCodeGenerateTaskDO submit(MesProBagCodeGenerateSpec spec, String requestId) {
|
// 1. 幂等：网络重试复用同一个 requestId 时直接返回已有任务，不再生成
|
MesProBagCodeGenerateTaskDO existing = findExisting(requestId);
|
if (existing != null) {
|
return existing;
|
}
|
// 2. 并发闸门：同一时间只跑一个后台生成任务；被挡住时带上当前任务进度
|
if (!registry.tryAcquire()) {
|
// 并发重复提交时，另一请求可能已抢先落库
|
existing = findExisting(requestId);
|
if (existing != null) {
|
return existing;
|
}
|
throw exception(PRO_BAG_CODE_GENERATE_TASK_BUSY, resolveRunningProgress());
|
}
|
// 权限一旦交给执行器，就由执行线程在收尾时归还；此前任何失败都必须在这里归还
|
boolean handedOver = false;
|
try {
|
MesProBagCodeGenerateTaskDO task = MesProBagCodeGenerateTaskDO.builder()
|
.requestId(requestId)
|
.bizType(spec.bizType().getType())
|
.uuid(spec.uuid())
|
.itemId(spec.itemId())
|
.itemCode(spec.itemCode())
|
.itemName(spec.itemName())
|
.unitType(spec.unitType())
|
.perPallet(spec.perPallet())
|
.palletCount(spec.palletCount())
|
.totalCount(spec.totalCount())
|
.finishedCount(0)
|
.status(MesProBagCodeGenerateTaskStatusEnum.PENDING.getStatus())
|
.remark(spec.remark())
|
.build();
|
try {
|
taskMapper.insert(task);
|
} catch (DuplicateKeyException e) {
|
// 唯一索引兜底并发重复提交
|
existing = findExisting(requestId);
|
if (existing != null) {
|
return existing;
|
}
|
throw e;
|
}
|
// 3. 先注册运行态再投递：避免「已落库但未注册」的窗口被清理任务误回收
|
registry.register(task.getId());
|
handedOver = true;
|
try {
|
executor.execute(() -> runner.run(task.getId(), spec));
|
} catch (RejectedExecutionException e) {
|
registry.unregister(task.getId());
|
handedOver = false;
|
taskMapper.updateStatus(task.getId(),
|
MesProBagCodeGenerateTaskStatusEnum.FAILED.getStatus(),
|
"生成线程繁忙，请稍后重试", LocalDateTime.now());
|
throw exception(PRO_BAG_CODE_GENERATE_TASK_BUSY, 0);
|
}
|
return task;
|
} finally {
|
if (!handedOver) {
|
registry.release();
|
}
|
}
|
}
|
|
@Override
|
public MesProBagCodeGenerateTaskDO getTask(Long id) {
|
MesProBagCodeGenerateTaskDO task = taskMapper.selectById(id);
|
if (task == null) {
|
throw exception(PRO_BAG_CODE_GENERATE_TASK_NOT_EXISTS);
|
}
|
return task;
|
}
|
|
@Override
|
public MesProBagCodeGenerateTaskDO getTaskByRequestId(String requestId) {
|
return findExisting(requestId);
|
}
|
|
@Override
|
public void cancel(Long id) {
|
MesProBagCodeGenerateTaskDO task = getTask(id);
|
if (FINAL_STATUSES.contains(task.getStatus())) {
|
throw exception(PRO_BAG_CODE_GENERATE_TASK_STATUS_INVALID);
|
}
|
if (registry.requestCancel(id)) {
|
// 执行线程会在当前分段结束后停止、清理半成品并置为已取消
|
taskMapper.updateStatus(id, MesProBagCodeGenerateTaskStatusEnum.CANCELING.getStatus(),
|
null, null);
|
return;
|
}
|
// 不在本进程运行（进程重启后的残留任务）：直接清理残留并收尾
|
cleanupQuietly(task);
|
taskMapper.updateStatus(id, MesProBagCodeGenerateTaskStatusEnum.CANCELED.getStatus(),
|
null, LocalDateTime.now());
|
}
|
|
@Override
|
public int recycleStaleTasks() {
|
List<MesProBagCodeGenerateTaskDO> list = taskMapper.selectUnfinishedList(UNFINISHED_STATUSES);
|
// 提交线程「落库 → 注册运行态」之间存在极短窗口，跳过刚创建的任务避免误回收
|
LocalDateTime recent = LocalDateTime.now().minusMinutes(1);
|
int count = 0;
|
for (MesProBagCodeGenerateTaskDO task : list) {
|
if (task.getCreateTime() != null && task.getCreateTime().isAfter(recent)) {
|
continue;
|
}
|
if (registry.isRunning(task.getId())) {
|
continue;
|
}
|
cleanupQuietly(task);
|
taskMapper.updateStatus(task.getId(), MesProBagCodeGenerateTaskStatusEnum.FAILED.getStatus(),
|
"任务因服务重启或异常中断而终止，请重新生成", LocalDateTime.now());
|
count++;
|
}
|
return count;
|
}
|
|
// ==================== 私有方法 ====================
|
|
private MesProBagCodeGenerateTaskDO findExisting(String requestId) {
|
if (StrUtil.isBlank(requestId)) {
|
return null;
|
}
|
return taskMapper.selectByRequestId(requestId);
|
}
|
|
/**
|
* 清理半成品：清理失败不能阻断任务终态回写，否则任务会一直挂在非终态
|
*/
|
private void cleanupQuietly(MesProBagCodeGenerateTaskDO task) {
|
try {
|
cleaner.cleanup(task.getUuid());
|
} catch (Exception e) {
|
log.error("[cleanupQuietly][生成任务 {} 半成品清理失败，uuid={}]",
|
task.getId(), task.getUuid(), e);
|
}
|
}
|
|
/**
|
* 正在运行任务的进度百分比，用于「已有任务进行中」的提示
|
*/
|
private int resolveRunningProgress() {
|
List<MesProBagCodeGenerateTaskDO> tasks = taskMapper.selectUnfinishedList(UNFINISHED_STATUSES);
|
return tasks.stream().findFirst().map(this::calcProgress).orElse(0);
|
}
|
|
private int calcProgress(MesProBagCodeGenerateTaskDO 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);
|
}
|
|
}
|
