diff --git a/src/main/java/com/ai/da/common/RabbitMQ/MQConfig.java b/src/main/java/com/ai/da/common/RabbitMQ/MQConfig.java index bf0a360a..d4e24508 100644 --- a/src/main/java/com/ai/da/common/RabbitMQ/MQConfig.java +++ b/src/main/java/com/ai/da/common/RabbitMQ/MQConfig.java @@ -16,7 +16,7 @@ import org.springframework.beans.factory.annotation.Value; public class MQConfig { public static final String GENERATE_EXCHANGE_FANOUT = "generate-exchange"; - public static final String GENERATE_QUEUE = "generate-queue"; + public static final String GENERATE_QUEUE = "generate-queue-prod"; public MQConfig() { } diff --git a/src/main/java/com/ai/da/common/RabbitMQ/MQConsumer.java b/src/main/java/com/ai/da/common/RabbitMQ/MQConsumer.java index eb30d23a..77f46dbd 100644 --- a/src/main/java/com/ai/da/common/RabbitMQ/MQConsumer.java +++ b/src/main/java/com/ai/da/common/RabbitMQ/MQConsumer.java @@ -63,12 +63,12 @@ public class MQConsumer { // 2.2 将该消息从取消列表中删除 // redisUtil.removeFromSet(cancelSetKey, uniqueId); } else { + /*try { + Thread.sleep(15000); + } catch (InterruptedException e) { + throw new RuntimeException(e); + }*/ GenerateCollectionVO generateCollectionVO = generateService.generateThroughImageText(generateThroughImageTextDTO); -// try { -// Thread.sleep(15000); -// } catch (InterruptedException e) { -// throw new RuntimeException(e); -// } // 将消息从redis排队队列中删除,需保证被消费的消息存储到db之后再从redis删除 redisUtil.removeFromZSet(consumptionOrderKey, uniqueId); if (!Objects.isNull(generateCollectionVO)) { @@ -100,8 +100,8 @@ public class MQConsumer { } long end = System.currentTimeMillis(); - log.info(" task_id： " + uniqueId + "----------" + consumerName + " 执行时长：" + (end - start) + "毫秒"); - log.info("============end listening=========="); + log.info(" task_id： " + uniqueId + "----------" + consumerName + " 执行时长：" + (end - start) + "毫秒"); + log.info("=============end listening==========="); } @RabbitListener(queues = MQConfig.GENERATE_QUEUE) diff --git a/src/main/java/com/ai/da/common/utils/AsyncCallerUtil.java b/src/main/java/com/ai/da/common/utils/AsyncCallerUtil.java index 7ef9e3a0..35039431 100644 --- a/src/main/java/com/ai/da/common/utils/AsyncCallerUtil.java +++ b/src/main/java/com/ai/da/common/utils/AsyncCallerUtil.java @@ -30,21 +30,20 @@ public class AsyncCallerUtil { public List generate(GenerateToPythonDTO generateToPython) { ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(5); String taskId = generateToPython.getTasks_id(); - waitingStatus.put(taskId, true); ScheduledFuture timeoutTask = null; + if (!waitingStatus.containsKey(taskId)) waitingStatus.put(taskId, true); try { CompletableFuture> generateResult = callGenerateAsync(generateToPython); - // 10秒后第一次确认，之后每隔10秒确认一次用户选择结果 + // 5秒后第一次确认，之后每隔10秒确认一次用户选择结果 timeoutTask = scheduledExecutorService.scheduleAtFixedRate(() -> { // 调用另一个接口获取用户的选择 if (!waitingStatus.get(taskId)) { - // 如果用户选择取消，则取消对generate的调用，cancel判断是否成功取消 + // 如果用户选择取消，则取消对generate的调用 generateResult.cancel(true); waitingStatus.remove(taskId); - } - log.info("===============持续等待==============="); - }, 10, 10, TimeUnit.SECONDS); + } else log.info("===============持续等待==============="); + }, 5, 10, TimeUnit.SECONDS); log.info("阻塞等待结果..."); // 阻塞，等待结果 @@ -53,6 +52,10 @@ public class AsyncCallerUtil { timeoutTask.cancel(true); waitingStatus.remove(taskId); return result; + } catch (CancellationException e) { + // generateResult.cancel(true);通过抛出异常取消该任务 + log.info("==========成功取消generate任务=========="); + return null; } catch (InterruptedException | ExecutionException | BusinessException e) { // 处理异常 log.error("发生错误 ： " + e); diff --git a/src/main/java/com/ai/da/python/PythonService.java b/src/main/java/com/ai/da/python/PythonService.java index b8ecb4e2..5a37b3b7 100644 --- a/src/main/java/com/ai/da/python/PythonService.java +++ b/src/main/java/com/ai/da/python/PythonService.java @@ -107,7 +107,7 @@ public class PythonService { if (Objects.nonNull(response.body())) { responseBody = response.body().string(); JSONObject responseObj = JSON.parseObject(responseBody); - log.info("moodboard与printboard图片合成 python返回###{}",responseObj); + log.info("moodboard与printboard图片合成 python返回###{}", responseObj); return responseObj.get("data").toString(); } } catch (IOException | JSONException e) { @@ -389,7 +389,7 @@ public class PythonService { all.addAll(new ArrayList<>(DesignPythonItem.SKIRT_TROUSERS)); return all; } - }else if (modelSex.equals(Sex.MALE.getValue())) { + } else if (modelSex.equals(Sex.MALE.getValue())) { Long randomIndex = RandomsUtil.randomSysFile(0L, 3L); if (randomIndex == 0) { return DesignPythonItem.TOPS; @@ -422,11 +422,11 @@ public class PythonService { long noPinNum = printBoardElements.stream().filter(f -> f.getHasPin() == 0).count(); if (noPinNum == 0L) { return 0; - }else { + } else { long pinNum = printBoardElements.stream().filter(f -> f.getHasPin() == 1).count(); if (8 - pinNum < 4) { return RandomsUtil.randomSysFile(0L, 8 - pinNum + 1); - }else { + } else { return RandomsUtil.randomSysFile(0L, 4L + 1); } } @@ -1600,7 +1600,7 @@ public class PythonService { printBoardElements = elementVO.getPrintBoardElements() .stream() .filter(f -> !elementVO.getHasUseMd5List().contains(f.getMd5())).collect(Collectors.toList()); - }else { + } else { printBoardElements = elementVO.getPrintBoardElements(); } if (CollectionUtil.isEmpty(printBoardElements)) { @@ -2081,7 +2081,7 @@ public class PythonService { skirt.setPrint(designPythonItemPrint); skirt.setPath("aida-sys-image/images/female/trousers/trousers_974.jpg"); response.add(skirt); - }else { + } else { DesignPythonItem top = new DesignPythonItem(); top.setType(MalePosition.TOPS.getValue()); top.setColor("none"); @@ -2195,7 +2195,9 @@ public class PythonService { throw new BusinessException("design.interface.exception"); } - /** 暂时未用 */ + /** + * 暂时未用 + */ public String generateSketchCaption(String url) { //限流校验 AccessLimitUtils.validate("generateSketchCaption", 5); @@ -2259,7 +2261,7 @@ public class PythonService { .addHeader("Content-Type", "application/json") .build(); Response response = null; - String bodyString ; + String bodyString; try { log.info("generateSketchOrPrint请求入参content###{}", JSON.toJSONString(generateToPythonDTO, SerializerFeature.WriteMapNullValue)); response = client.newCall(request).execute(); @@ -2276,7 +2278,7 @@ public class PythonService { log.error("PythonService##generateSketchOrPrint异常###{}", "response or body is empty!"); // throw new BusinessException("generate.interface.error"); throw new BusinessException("PythonService##generateSketchOrPrint异常###: response or body is empty!"); - } else if (response.code() != HttpURLConnection.HTTP_OK){ + } else if (response.code() != HttpURLConnection.HTTP_OK) { log.error("PythonService##generateSketchOrPrint异常###{}", "Response error!Response code ## " + response.code() + " ##"); // throw new BusinessException("generate.interface.error"); throw new BusinessException("PythonService##generateSketchOrPrint异常### Response error!Response code ## " + response.code() + " ##"); @@ -2356,7 +2358,7 @@ public class PythonService { if (Objects.isNull(response) || Objects.isNull(response.body())) { log.error("PythonService##composeLayers异常###{}", "response or body is empty!"); throw new BusinessException("compose-layer.interface.exception"); - }else if (response.code() != HttpURLConnection.HTTP_OK){ + } else if (response.code() != HttpURLConnection.HTTP_OK) { log.error("PythonService##composeLayers异常###{}", "Response error!Response code ## " + response.code() + " ##"); throw new BusinessException("compose-layer.interface.exception"); } else { @@ -2381,10 +2383,10 @@ public class PythonService { return item0.getString("synthesis_url"); } - public String getClothCategory(String path,String gender){ + public String getClothCategory(String path, String gender) { HashMap content = new HashMap<>(); - content.put("sketch_img_url",path); - content.put("colony",gender); + content.put("sketch_img_url", path); + content.put("colony", gender); List> contents = Collections.singletonList(content); String jsonString = JSON.toJSONString(contents, SerializerFeature.WriteNullStringAsEmpty); @@ -2397,7 +2399,7 @@ public class PythonService { if (Objects.isNull(response) || Objects.isNull(response.body())) { log.error("PythonService##GetClothCategory###{}", "response or body is empty!"); throw new BusinessException("cloth-classification.interface.exception"); - } else if (response.code() != HttpURLConnection.HTTP_OK){ + } else if (response.code() != HttpURLConnection.HTTP_OK) { log.error("PythonService##GetClothCategory###{}", "Response error!Response code ## " + response.code() + " ##"); throw new BusinessException("cloth-classification.interface.exception"); } else { @@ -2408,7 +2410,7 @@ public class PythonService { } } JSONObject jsonObject = JSON.parseObject(bodyString); - try{ + try { Boolean result = JSON.parseObject(JSON.toJSONString(response)).getBoolean("successful"); if (result && jsonObject.get("msg").equals("OK!")) { JSONObject data = jsonObject.getJSONObject("data"); @@ -2416,7 +2418,7 @@ public class PythonService { JSONObject map = (JSONObject) list.get(0); return map.get("category").toString(); } - }catch (NullPointerException e){ + } catch (NullPointerException e) { log.info("getClothCategory 失败###{}，未返回category", jsonObject); throw new BusinessException("cloth-classification.interface.exception"); } @@ -2425,14 +2427,14 @@ public class PythonService { throw new BusinessException("cloth-classification.interface.exception"); } - public Boolean cancelGenerateTask(String taskId){ + public Boolean cancelGenerateTask(String taskId) { OkHttpClient client = new OkHttpClient().newBuilder() .connectTimeout(30, TimeUnit.SECONDS) .pingInterval(5, TimeUnit.SECONDS)//websocket轮训间隔(单位:秒) .readTimeout(60, TimeUnit.SECONDS)//读取超时(单位:秒) .writeTimeout(60, TimeUnit.SECONDS)//写入超时(单位:秒) .build(); - String url = accessPythonIp + ":" + accessPythonPort + "/api/generate_cancel/" + taskId + "/"; + String url = accessPythonIp + ":" + accessPythonPort + "/api/generate_cancel/" + taskId; Request request = new Request.Builder() .url(url) // .addHeader("Authorization", "Basic YWlkbGFiOjEyMw==") @@ -2444,12 +2446,14 @@ public class PythonService { response = client.newCall(request).execute(); } catch (IOException ioException) { log.error("PythonService##cancelGenerateTask异常###{}", ExceptionUtil.getThrowableList(ioException)); - throw new BusinessException("generate.interface.error"); + return null; } - if (response.code() != HttpURLConnection.HTTP_OK){ + if (response.code() != HttpURLConnection.HTTP_OK) { + log.info("generate-python 取消请求失败"); return Boolean.FALSE; } + log.info("generate-python 取消请求成功"); return Boolean.TRUE; } diff --git a/src/main/java/com/ai/da/service/impl/GenerateServiceImpl.java b/src/main/java/com/ai/da/service/impl/GenerateServiceImpl.java index 5ff34fb5..81d48ddc 100644 --- a/src/main/java/com/ai/da/service/impl/GenerateServiceImpl.java +++ b/src/main/java/com/ai/da/service/impl/GenerateServiceImpl.java @@ -123,7 +123,7 @@ public class GenerateServiceImpl extends ServiceImpl i generateThroughImageTextDTO.getLevel1Type().equals(PRINT_BOARD.getRealName()) ? "print" : "moodboard"; AsyncCallerUtil asyncCallerUtil = new AsyncCallerUtil(); List generatedSketchUrl = asyncCallerUtil.generate(new GenerateToPythonDTO(accountId, Objects.isNull(collectionElement) ? "" : collectionElement.getUrl(), - category, text, mode, "1", generateThroughImageTextDTO.getGender() ,generateThroughImageTextDTO.getUniqueId())); + category, text, mode, "1", generateThroughImageTextDTO.getGender(), generateThroughImageTextDTO.getUniqueId())); // List generatedSketchUrl = pythonService.generateSketchOrPrint(new GenerateToPythonDTO(accountId, Objects.isNull(elementId) ? null : collectionElement.getUrl(), // category, text, mode, "1", generateThroughImageTextDTO.getGender())); log.info("generate 响应 ： " + generatedSketchUrl); @@ -336,7 +336,7 @@ public class GenerateServiceImpl extends ServiceImpl i num++; } // 无依据确定的数字 - if (num > 10){ + if (num > 10) { try { Thread.sleep(1000); } catch (InterruptedException e) { @@ -412,9 +412,9 @@ public class GenerateServiceImpl extends ServiceImpl i return new GenerateCollectionVO(generateId, null, generatedCollectionItems); } - public Generate selectByUniqueId(String uniqueId){ + public Generate selectByUniqueId(String uniqueId) { QueryWrapper qw = new QueryWrapper<>(); - qw.eq("unique_id",uniqueId); + qw.eq("unique_id", uniqueId); return getOne(qw); } @@ -423,19 +423,22 @@ public class GenerateServiceImpl extends ServiceImpl i public void cancelGenerate(String uniqueId) { // 1、确认当前消息是否还在排队中 Boolean exists = redisUtil.isElementExistsInZSet(consumptionOrderKey, uniqueId); - if (exists) { + Boolean flag = Boolean.FALSE; + if (exists) flag = redisUtil.getRank(consumptionOrderKey, uniqueId) > 1L ? Boolean.TRUE : Boolean.FALSE; + // 不管flag的默认值是true还是false，只要exists为false，&& 将短路 + if (exists && flag) { // 1.1、将需要取消的唯一id加入redis，以便及时取消生成 redisUtil.addToSet(cancelSetKey, uniqueId); // 1.2 将需要取消的id从redis的ConsumptionOrder中删除 redisUtil.removeFromZSet(consumptionOrderKey, uniqueId); - }else { + } else { // 2、判断该消息是否异常 boolean hasKey = redisUtil.isElementExistsInMap(exceptionMapKey, uniqueId); // 3、判断该消息是否已经消费结束 Boolean existsInResult = redisUtil.isElementExistsInMap(resultMapKey, uniqueId); - if (!hasKey && !existsInResult){ + if (!hasKey && !existsInResult) { // 设置取等待状态为false - AsyncCallerUtil.waitingStatus.put(uniqueId,false); + AsyncCallerUtil.waitingStatus.put(uniqueId, false); // 3、直接发送取消请求到python端 pythonService.cancelGenerateTask(uniqueId); } diff --git a/src/main/resources/application-prod.properties b/src/main/resources/application-prod.properties index 3abdcf3b..a9510207 100644 --- a/src/main/resources/application-prod.properties +++ b/src/main/resources/application-prod.properties @@ -54,3 +54,24 @@ minio.bucketName.sysImage=aida-sys-image minio.bucketName.users=aida-users minio.bucketName.collectionElement=aida-collection-element redirect_url=http://18.167.251.121:7788 + +spring.rabbitmq.host=18.167.251.121 +spring.rabbitmq.port=5672 +spring.rabbitmq.username=rabbit +spring.rabbitmq.password=123456 +spring.rabbitmq.virtual-host=/ + +spring.redis.host=172.31.11.32 +#spring.redis.host=18.167.251.121 +spring.redis.port=6379 +spring.redis.database=1 +spring.redis.password=Aidlab +spring.redis.lettuce.pool.max-active=8 +spring.redis.lettuce.pool.max-idle=8 +spring.redis.lettuce.pool.min-idle=0 +spring.redis.lettuce.pool.max-wait=5 + +redis.key.consumptionOrder=ConsumptionOrder +redis.key.cancelSet=CancelSet +redis.key.exceptionMap=ExceptionMap +redis.key.resultMap=ResultMap \ No newline at end of file diff --git a/src/main/resources/application.properties b/src/main/resources/application.properties index ec94939b..8550475d 100644 --- a/src/main/resources/application.properties +++ b/src/main/resources/application.properties @@ -1,8 +1,8 @@ #����application-test�ļ�(���Ի���) -spring.profiles.active=test +#spring.profiles.active=test #����application-prod�ļ�(��������) -#spring.profiles.active=prod +spring.profiles.active=prod #����application-dev�ļ�(��������) #spring.profiles.active=dev