169 lines
9.0 KiB
Java
169 lines
9.0 KiB
Java
package digital.application.ScheduleTask;
|
||
|
||
import com.alibaba.fastjson.JSON;
|
||
import com.alibaba.fastjson.JSONObject;
|
||
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
|
||
import digital.application.entity.SysHistory;
|
||
import digital.application.entity.SysTaskQueue;
|
||
import digital.application.mapper.MwTrainTaskMapper;
|
||
import digital.application.mapper.SysHistoryMapper;
|
||
import digital.application.service.SysTaskQueueService;
|
||
import digital.base.constant.CommonConstant;
|
||
import digital.base.constant.Face3dSeverHTTPConstant;
|
||
import digital.base.constant.ZitaCommonConstant;
|
||
import digital.bean.task.MwTrainTask;
|
||
import digital.config.oss.OssConfiguration;
|
||
import digital.util.oss.OssBootUtil;
|
||
import digital.util.util.RestUtil;
|
||
import lombok.extern.slf4j.Slf4j;
|
||
import org.springframework.beans.factory.annotation.Autowired;
|
||
import org.springframework.http.HttpMethod;
|
||
import org.springframework.http.ResponseEntity;
|
||
import org.springframework.scheduling.annotation.Scheduled;
|
||
import org.springframework.stereotype.Component;
|
||
import org.springframework.transaction.annotation.Transactional;
|
||
|
||
import java.io.File;
|
||
import java.time.LocalTime;
|
||
import java.time.format.DateTimeFormatter;
|
||
import java.util.UUID;
|
||
import java.util.concurrent.*;
|
||
|
||
@Component
|
||
@Slf4j
|
||
public class ClothTaskService {
|
||
|
||
@Autowired
|
||
OssConfiguration ossConfiguration;
|
||
@Autowired
|
||
private SysTaskQueueService sysTaskQueueService;
|
||
@Autowired
|
||
private MwTrainTaskMapper trainTaskMapper;
|
||
@Autowired
|
||
private SysHistoryMapper sysHistoryMapper;
|
||
|
||
private ScheduledExecutorService executorService;
|
||
private BlockingQueue<SysTaskQueue> clothTaskQueue = new LinkedBlockingQueue<>();
|
||
|
||
public ClothTaskService() {
|
||
this.executorService = Executors.newSingleThreadScheduledExecutor();
|
||
}
|
||
|
||
@Scheduled(fixedRate = 3000) // 每隔3秒执行一次
|
||
@Transactional
|
||
public synchronized void pollClothTaskTable() {
|
||
String taskId = UUID.randomUUID().toString();
|
||
QueryWrapper<SysTaskQueue> queryWrapper = new QueryWrapper<>();
|
||
queryWrapper.lambda()
|
||
.eq(SysTaskQueue::getStatus, Face3dSeverHTTPConstant.CREATE_STATUS_DRAFT)
|
||
.orderByAsc(SysTaskQueue::getId)
|
||
.eq(SysTaskQueue::getTryOnType, CommonConstant.TRY_ON_TYPE_SINGLE)
|
||
.and(wrapper -> wrapper.isNull(SysTaskQueue::getTryOnAlgo)
|
||
.or().eq(SysTaskQueue::getTryOnAlgo, CommonConstant.TRY_ON_TYPE_ALGO_CLOTH))
|
||
.last("LIMIT 1"); // 添加这个条件来限制查询结果为1条
|
||
// log.info(Thread.currentThread().getName() + "放入【换装】队列前，clothTaskQueue = " + clothTaskQueue.size() + " taskId == " + taskId);
|
||
SysTaskQueue sysTaskQueue = sysTaskQueueService.getOne(queryWrapper);
|
||
if (null == sysTaskQueue) {
|
||
QueryWrapper<SysTaskQueue> qw = new QueryWrapper<>();
|
||
qw.lambda()
|
||
.eq(SysTaskQueue::getStatus, Face3dSeverHTTPConstant.CREATE_STATUS_DRAFT)
|
||
.orderByAsc(SysTaskQueue::getId)
|
||
.eq(SysTaskQueue::getTryOnType, CommonConstant.TRY_ON_TYPE_BATCH)
|
||
.and(wrapper -> wrapper.isNull(SysTaskQueue::getTryOnAlgo)
|
||
.or().eq(SysTaskQueue::getTryOnAlgo, CommonConstant.TRY_ON_TYPE_ALGO_CLOTH)) // 假设 "cloth" 是你要比较的值
|
||
.last("LIMIT 1"); // 限制只返回一条记录
|
||
sysTaskQueue = sysTaskQueueService.getOne(qw);
|
||
}
|
||
if (sysTaskQueue != null && clothTaskQueue.size() < 1) {
|
||
clothTaskQueue.offer(sysTaskQueue);
|
||
sysTaskQueue.setStatus(Face3dSeverHTTPConstant.CREATE_STATUS_IN_PROGRESS);
|
||
sysTaskQueueService.updateById(sysTaskQueue);
|
||
}
|
||
// log.info(Thread.currentThread().getName() + " 放入【换装】队列后，clothTaskQueue = " + clothTaskQueue.size() + " taskId == " + taskId);
|
||
}
|
||
|
||
public void startClothProcessingTasks() {
|
||
executorService.scheduleAtFixedRate(() -> {
|
||
// log.info("当前【换装】队列大小=" + clothTaskQueue.size());
|
||
SysTaskQueue clothTask = clothTaskQueue.poll();
|
||
// log.info("当前【换装】队列大小=" + clothTaskQueue.size());
|
||
if (clothTask != null) {
|
||
// 执行任务
|
||
processTask(clothTask);
|
||
}
|
||
}, 0, 3, TimeUnit.SECONDS);
|
||
}
|
||
|
||
|
||
private synchronized void processTask(SysTaskQueue sysTaskQueue) {
|
||
String taskId = sysTaskQueue.getTaskId();
|
||
long startTime = System.currentTimeMillis();
|
||
log.info(Thread.currentThread().getName() + " 开始【换装】=========================================：taskId==" + taskId + LocalTime.now().format(DateTimeFormatter.ofPattern("HH:mm:ss")));
|
||
String jsonInfo = sysTaskQueue.getJsonInfo();
|
||
JSONObject clothParam = JSON.parseObject(jsonInfo);
|
||
MwTrainTask trainTask = trainTaskMapper.selectOne(new QueryWrapper<MwTrainTask>().lambda().eq(MwTrainTask::getTaskId, taskId));
|
||
//防止数据删除空指针
|
||
SysHistory sysHistory = sysHistoryMapper.selectHistoryOne(taskId);
|
||
try {
|
||
String url;
|
||
if (ZitaCommonConstant.sex_male.equals(sysTaskQueue.getSex())) {
|
||
url = Face3dSeverHTTPConstant.SWAP_CLOTH_MALE;
|
||
} else {
|
||
url = Face3dSeverHTTPConstant.SWAP_CLOTH;
|
||
}
|
||
ResponseEntity<JSONObject> clothResult = RestUtil.zitaRequest(url,
|
||
HttpMethod.POST, null, clothParam, JSONObject.class);
|
||
String clothState = clothResult.getBody().get("status").toString();
|
||
log.info(taskId + " 返回结果信息成功 ==" + taskId + " clothResult=" + clothResult.getBody());
|
||
if (!ZitaCommonConstant.STATUS_NORMAL.equals(clothState)) {
|
||
trainTask.setStatus(Face3dSeverHTTPConstant.result_fail);
|
||
trainTask.setResult(Face3dSeverHTTPConstant.result_fail);
|
||
trainTask.setRequestResult(String.format("digital cloth 【换装】任务失败！{taskId: %s, resp: %s}", taskId, clothResult.getBody().toJSONString()));
|
||
trainTaskMapper.updateById(trainTask);
|
||
log.info(String.format("digital cloth 【换装】任务失败！{taskId: %s, resp: %s}", taskId, clothResult.getBody().toJSONString()));
|
||
sysHistory.setFileUrl(CommonConstant.TRY_ON_FAIL);
|
||
sysHistory.setStatus(Face3dSeverHTTPConstant.result_fail);
|
||
sysHistoryMapper.updateById(sysHistory);
|
||
sysTaskQueue.setStatus(Face3dSeverHTTPConstant.result_fail);
|
||
sysTaskQueueService.updateById(sysTaskQueue);
|
||
return;
|
||
}
|
||
// 尝试将返回结果转为JSON
|
||
JSONObject resultBody = clothResult.getBody();
|
||
String imagePath = (String) resultBody.get("data");
|
||
File file = new File(imagePath);
|
||
String newFileName = UUID.randomUUID().toString() + file.getName().substring(file.getName().lastIndexOf("."));
|
||
String fileUrl = OssBootUtil.uploadOssVideo(file, CommonConstant.PUTON_SWAP, newFileName);
|
||
String savePath = OssBootUtil.getDigitalObjectURL(ossConfiguration.getBucketName(), fileUrl, null);
|
||
trainTask.setResultImages(savePath);
|
||
trainTask.setStatus(Face3dSeverHTTPConstant.CREATE_STATUS_SUCCESS);
|
||
trainTask.setResult(Face3dSeverHTTPConstant.CREATE_STATUS_SUCCESS);
|
||
long endTime = System.currentTimeMillis();
|
||
long duration = (endTime - startTime) / 1000; // 除以1000，将毫秒转换为秒
|
||
trainTask.setCostTime((int) duration);
|
||
trainTaskMapper.updateById(trainTask);
|
||
//替换换装后的照片
|
||
sysHistory.setFileUrl(savePath);
|
||
sysHistory.setStatus(Face3dSeverHTTPConstant.CREATE_STATUS_SUCCESS);
|
||
sysHistoryMapper.updateById(sysHistory);
|
||
sysTaskQueue.setStatus(Face3dSeverHTTPConstant.CREATE_STATUS_SUCCESS);
|
||
sysTaskQueueService.updateById(sysTaskQueue);
|
||
} catch (Exception ex) {
|
||
String message = ex.getMessage();
|
||
trainTask.setStatus(Face3dSeverHTTPConstant.result_fail);
|
||
trainTask.setResult(Face3dSeverHTTPConstant.result_fail);
|
||
trainTask.setRequestResult("digital cloth 【换装】失败！" + String.format("%s {taskId: %s}", message, taskId));
|
||
trainTaskMapper.updateById(trainTask);
|
||
log.error("digital cloth 【换装】任务失败！" + String.format("%s {taskId: %s}", message, taskId));
|
||
sysHistory.setFileUrl(CommonConstant.TRY_ON_FAIL);
|
||
sysHistory.setStatus(Face3dSeverHTTPConstant.result_fail);
|
||
sysHistoryMapper.updateById(sysHistory);
|
||
sysTaskQueue.setStatus(Face3dSeverHTTPConstant.result_fail);
|
||
sysTaskQueueService.updateById(sysTaskQueue);
|
||
return;
|
||
}
|
||
log.info(Thread.currentThread().getName() + "结束【换装】=========================================：" + LocalTime.now().format(DateTimeFormatter.ofPattern("HH:mm:ss")));
|
||
}
|
||
|
||
}
|