TASK:异步调用generate及取消generate

src/main/java/com/ai/da/common/RabbitMQ/MQConsumer.java

@@ -48,12 +48,11 @@ public class MQConsumer {
         log.info("============start listening==========");
 
         GenerateThroughImageTextDTO generateThroughImageTextDTO = JSONObject.parseObject(msg.getBody(), GenerateThroughImageTextDTO.class);
-        Long uniqueId = generateThroughImageTextDTO.getUniqueId();
+        String uniqueId = generateThroughImageTextDTO.getUniqueId();
-        // 1、将消息从redis排队队列中删除
+
-        redisUtil.removeFromZSet(consumptionOrderKey, String.valueOf(uniqueId));
         try {
             // 2、判断当前消息是否在取消列表中
-            Boolean isMember = redisUtil.isElementExistsInSet(cancelSetKey, String.valueOf(uniqueId));
+            Boolean isMember = redisUtil.isElementExistsInSet(cancelSetKey, uniqueId);
             if (isMember) {
                 try {
                     // 2.1 手动确认该消息
@@ -62,40 +61,43 @@ public class MQConsumer {
                     log.error("手动确认，不返回队列重新消费");
                 }
                 // 2.2 将该消息从取消列表中删除
-                redisUtil.removeFromSet(cancelSetKey, String.valueOf(uniqueId));
+//                redisUtil.removeFromSet(cancelSetKey, uniqueId);
             } else {
                 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)){
                     HashMap<String, String> generateResult = new HashMap<>();
-                    generateResult.put(String.valueOf(uniqueId), JSONObject.toJSONString(generateCollectionVO));
+                    generateResult.put(uniqueId, JSONObject.toJSONString(generateCollectionVO));
                     // 将结果存在redis中 ,为空时不要存
                     redisUtil.addToMap(resultMapKey, generateResult);
                 }
 
             }
         } catch (BusinessException e) {
-            log.error(e.getMessage());
+            log.error(e.getMsg());
             // channel.basicNack() 为不确认deliveryTag对应的消息，第二个参数是否应用于多消息，第三个参数是否requeue
             try {
                 // 第二个参数，是否批量确认消息，当传false时，只确认当前 deliveryTag对应的消息;当传true时，会确认当前及之前所有未确认的消息。
                 channel.basicAck(msg.getMessageProperties().getDeliveryTag(), false);
+                // 将消息从redis排队队列中删除,需保证被消费的消息存储到db之后再从redis删除
+                redisUtil.removeFromZSet(consumptionOrderKey, uniqueId);
             } catch (IOException exception) {
                 log.error("手动确认，取消返回队列，不再重新消费");
             }
             // 将入参和错误信息存入数据库
-            String exceptionMessage = JSONObject.toJSONString(generateThroughImageTextDTO) + " Exception message ： " + e.getMessage();
+            String exceptionMessage = JSONObject.toJSONString(generateThroughImageTextDTO) +
+                    " Exception message ： " + e.getMsg();
             HashMap<String, String> exceptionInfo = new HashMap<>();
             exceptionInfo.put(String.valueOf(uniqueId), exceptionMessage);
             // 存redis
             redisUtil.addToMap(exceptionMapKey, exceptionInfo);
         }
-
-//        log.info(JSONObject.parseObject(msg.getBody(), GenerateThroughImageTextDTO.class).toString());
-//        try {
-//            Thread.sleep(10000);
-//        } catch (InterruptedException e) {
-//            throw new RuntimeException(e);
-//        }
         log.info("============end listening==========");
     }
 

src/main/java/com/ai/da/common/utils/AsyncCallerUtil.java

@@ -14,7 +14,7 @@ import java.util.concurrent.*;
 @Component
 public class AsyncCallerUtil {
 
-    public static Map<Long, Boolean> waitingStatus = new HashMap<>();
+    public static Map<String, Boolean> waitingStatus = new HashMap<>();
 
     private static PythonService pythonService;
 
@@ -27,9 +27,10 @@ public class AsyncCallerUtil {
         return CompletableFuture.supplyAsync(() -> pythonService.generateSketchOrPrint(generateToPython));
     }
 
-    public List<String> generate(GenerateToPythonDTO generateToPython, Long requestId) {
+    public List<String> generate(GenerateToPythonDTO generateToPython) {
-        ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);
+        ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(5);
-        waitingStatus.put(requestId, true);
+        String taskId = generateToPython.getTasks_id();
+        waitingStatus.put(taskId, true);
         ScheduledFuture<?> timeoutTask = null;
 
         try {
@@ -37,10 +38,10 @@ public class AsyncCallerUtil {
             // 10秒后第一次确认，之后每隔10秒确认一次用户选择结果
             timeoutTask = scheduledExecutorService.scheduleAtFixedRate(() -> {
                 // 调用另一个接口获取用户的选择
-                if (!waitingStatus.get(requestId)) {
+                if (!waitingStatus.get(taskId)) {
                     // 如果用户选择取消，则取消对generate的调用，cancel判断是否成功取消
                     generateResult.cancel(true);
-                    waitingStatus.remove(requestId);
+                    waitingStatus.remove(taskId);
                 }
                 System.out.println("持续等待...... : " + DateUtil.getByTimeZone("Asia/Shanghai"));
             }, 10, 10, TimeUnit.SECONDS);
@@ -54,16 +55,15 @@ public class AsyncCallerUtil {
             // 处理结果
             System.out.println("generate 响应: " + result);
             System.out.println("schedule finish time : " + DateUtil.getByTimeZone("Asia/Shanghai"));
-            waitingStatus.remove(requestId);
+            waitingStatus.remove(taskId);
             return result;
         } catch (InterruptedException | ExecutionException | BusinessException e) {
             // 处理异常
             log.error("发生错误 ： " + e);
-            e.printStackTrace();
             // 取消定时任务
             assert timeoutTask != null;
             timeoutTask.cancel(true);
-            throw new BusinessException("generate.interface.error");
+            throw new BusinessException(e.getMessage());
         } finally {
             // 关闭线程池
 //            executorService.shutdown();

src/main/java/com/ai/da/controller/GenerateController.java

@@ -56,21 +56,19 @@ public class GenerateController {
     @ApiOperation(value = "发起生成请求，异步获取结果")
     @PostMapping("/prepare")
     public Response<String> prepareForGenerate(@Valid @RequestBody GenerateThroughImageTextDTO generateThroughImageTextDTO) {
-        Long l = generateService.prepareForGenerate(generateThroughImageTextDTO);
+        return Response.success(generateService.prepareForGenerate(generateThroughImageTextDTO));
-        // 防止long精度丢失，这里转为String类型进行传输
-        return Response.success(String.valueOf(l));
     }
 
     @ApiOperation(value = "取消继续生成")
     @GetMapping("/stopWaiting")
-    public Response<String> stopWaiting(@RequestParam("uniqueId") Long uniqueId){
+    public Response<String> stopWaiting(@RequestParam("uniqueId") String uniqueId) {
         generateService.cancelGenerate(uniqueId);
         return Response.success("stop waiting successfully");
     }
 
     @ApiOperation(value = "获取生成结果")
     @GetMapping("/result")
-    public Response<GenerateCollectionVO> getGenerateResult(@RequestParam("uniqueId") Long uniqueId){
+    public Response<GenerateCollectionVO> getGenerateResult(@RequestParam("uniqueId") String uniqueId) {
         GenerateCollectionVO generateResult = generateService.getGenerateResult(uniqueId);
         return Response.success(generateResult);
     }

src/main/java/com/ai/da/mapper/entity/Generate.java

@@ -28,9 +28,9 @@ public class Generate {
     private Long accountId;
 
     /**
-     * 唯一id，用于保持消息的唯一性
+     * 唯一id
      */
-    private Long uniqueId;
+    private String uniqueId;
 
     /**
      * Sketchboard  Printboard

src/main/java/com/ai/da/model/dto/GenerateThroughImageTextDTO.java

@@ -46,5 +46,5 @@ public class GenerateThroughImageTextDTO {
     String timeZone;
 
     @ApiModelProperty("唯一id，用于保持消息唯一性")
-    Long uniqueId;
+    String uniqueId;
 }

src/main/java/com/ai/da/model/dto/GenerateToPythonDTO.java

@@ -22,4 +22,6 @@ public class GenerateToPythonDTO {
     private String version;
 
     private String gender;
+
+    private String tasks_id;
 }

src/main/java/com/ai/da/python/PythonService.java

@@ -56,6 +56,9 @@ public class PythonService {
     private String accessPythonIp;
     @Value("${access.python.port:''}")
     private String accessPythonPort;
+    @Value("${access.generate.port:''}")
+    private String accessGeneratePort;
+
     @Resource
     private PythonTAllInfoService pythonTAllInfoService;
 
@@ -2251,9 +2254,10 @@ public class PythonService {
         Request request = new Request.Builder()
 //                .url("http://18.167.251.121:9992")
 //                .url("http://127.0.0.1:5000/api/diffusion")
-                .url(accessPythonIp + ":" + accessPythonPort + "/api/diffusion")
+//                .url(accessPythonIp + ":" + accessPythonPort + "/api/diffusion")
+                .url(accessPythonIp + ":" + accessPythonPort + "/api/generate_image")
                 .method("POST", body)
-                .addHeader("Authorization", "Basic YWlkbGFiOjEyMw==")
+//                .addHeader("Authorization", "Basic YWlkbGFiOjEyMw==")
                 .addHeader("Content-Type", "application/json")
                 .build();
         Response response = null;
@@ -2263,23 +2267,28 @@ public class PythonService {
             response = client.newCall(request).execute();
         } catch (IOException ioException) {
             log.error("PythonService##generateSketchOrPrint异常###{}", ExceptionUtil.getThrowableList(ioException));
-            throw new BusinessException("generate.interface.error");
+//            throw new BusinessException("generate.interface.error");
+            throw new BusinessException(ioException.getMessage());
         }
         //去除限流
         AccessLimitUtils.validateOut("generateSketchOrPrint");
 
         // 判断是否生成失败
-        if (Objects.isNull(response) || Objects.isNull(response.body())) {
+        if (Objects.isNull(response.body())) {
             log.error("PythonService##generateSketchOrPrint异常###{}", "response or body is empty!");
-            throw new BusinessException("generate.interface.error");
+//            throw new BusinessException("generate.interface.error");
+            throw new BusinessException("PythonService##generateSketchOrPrint异常###: response or body is empty!");
         } 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("generate.interface.error");
+            throw new BusinessException("PythonService##generateSketchOrPrint异常### Response error!Response code ## " + response.code() + " ##");
         } else {
             try {
                 bodyString = response.body().string();
             } catch (IOException e) {
-                throw new BusinessException("generate.interface.error");
+                log.error(e.getMessage());
+//                throw new BusinessException("generate.interface.error");
+                throw new BusinessException(e.getMessage());
             }
         }
         JSONObject jsonObject = JSON.parseObject(bodyString);
@@ -2417,4 +2426,28 @@ public class PythonService {
         //生成失败
         throw new BusinessException("cloth-classification.interface.exception");
     }
+
+    public void 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();
+        HttpUrl.Builder builder = HttpUrl.parse(accessPythonIp + ":" + accessGeneratePort + "/cancel_task").newBuilder();
+        builder.addQueryParameter("task_id",taskId);
+        Request request = new Request.Builder()
+                .url(builder.build().toString())
+                .addHeader("Authorization", "Basic YWlkbGFiOjEyMw==")
+                .addHeader("Content-Type", "application/json")
+                .build();
+        try {
+            log.info("getGenerateResult请求入参content###{}", taskId);
+            client.newCall(request).execute();
+        } catch (IOException ioException) {
+            log.error("PythonService##getGenerateResult异常###{}", ExceptionUtil.getThrowableList(ioException));
+            throw new BusinessException("generate.interface.error");
+        }
+    }
+
 }

src/main/java/com/ai/da/service/GenerateService.java

@@ -25,12 +25,12 @@ public interface GenerateService extends IService<Generate> {
 
     List<GenerateDetail> selectBatchByLibraryId(List<Long> libraryId);
 
-    GenerateCollectionVO getGenerateResult(Long uniqueId);
+    GenerateCollectionVO getGenerateResult(String uniqueId);
 
-    Long prepareForGenerate(GenerateThroughImageTextDTO generateThroughImageTextDTO);
+    String prepareForGenerate(GenerateThroughImageTextDTO generateThroughImageTextDTO);
 
-    Long getRankPosition(Long uniqueId);
+    Long getRankPosition(String uniqueId);
 
-    void cancelGenerate(Long uniqueId);
+    void cancelGenerate(String uniqueId);
 
 }

src/main/java/com/ai/da/service/impl/GenerateServiceImpl.java

@@ -24,6 +24,7 @@ import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
 import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
 import io.minio.errors.MinioException;
 import io.netty.util.internal.StringUtil;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
@@ -35,6 +36,7 @@ import java.util.*;
 
 import static com.ai.da.common.enums.CollectionLevel1TypeEnum.*;
 
+@Slf4j
 @Service
 public class GenerateServiceImpl extends ServiceImpl<GenerateMapper, Generate> implements GenerateService {
 
@@ -120,8 +122,8 @@ public class GenerateServiceImpl extends ServiceImpl<GenerateMapper, Generate> i
         String category = generateThroughImageTextDTO.getLevel1Type().equals(SKETCH_BOARD.getRealName()) ? "sketch" :
                 generateThroughImageTextDTO.getLevel1Type().equals(PRINT_BOARD.getRealName()) ? "print" : "moodboard";
         AsyncCallerUtil asyncCallerUtil = new AsyncCallerUtil();
-        List<String> generatedSketchUrl = asyncCallerUtil.generate(new GenerateToPythonDTO(accountId, Objects.isNull(collectionElement) ? null : collectionElement.getUrl(),
+        List<String> 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<String> generatedSketchUrl = pythonService.generateSketchOrPrint(new GenerateToPythonDTO(accountId, Objects.isNull(elementId) ? null : collectionElement.getUrl(),
 //                category, text, mode, "1", generateThroughImageTextDTO.getGender()));
         if (CollectionUtils.isEmpty(generatedSketchUrl)) {
@@ -307,7 +309,7 @@ public class GenerateServiceImpl extends ServiceImpl<GenerateMapper, Generate> i
     }
 
     @Override
-    public Long prepareForGenerate(GenerateThroughImageTextDTO generateThroughImageTextDTO) {
+    public String prepareForGenerate(GenerateThroughImageTextDTO generateThroughImageTextDTO) {
         // 1、参数检查，判断必须参数是否为空
         if (Objects.isNull(generateThroughImageTextDTO.getUserId())) {
             throw new BusinessException("userId cannot be empty");
@@ -320,58 +322,66 @@ public class GenerateServiceImpl extends ServiceImpl<GenerateMapper, Generate> i
         Long elementId = generateThroughImageTextDTO.getCollectionElementId();
         validateGeneraType(new Generate(), text, elementId, generateType);
 
-        // 2、确定当前排队人数总数，超过15个，暂停使用当前功能
+        // 2、生成唯一id 使用uuid
-        Long zSetTotal = redisUtil.getZSetTotal(consumptionOrderKey);
+        String uuid = UUID.randomUUID().toString();
-        if (zSetTotal.equals(15L)){
-            return null;
-        }
 
-        // 3、生成唯一id
+//        SnowflakeUtil idWorker = new SnowflakeUtil(0, 0);
-        SnowflakeUtil idWorker = new SnowflakeUtil(0, 0);
+//        long snowflakeId = idWorker.nextId();
-        long snowflakeId = idWorker.nextId();
 
-        if (AsyncCallerUtil.waitingStatus.containsKey(snowflakeId)){
+        int num = 1;
-            snowflakeId = idWorker.nextId();
+        // 判断与已经正常生成结果的uuid中有相同的id
+        while (redisUtil.isElementExistsInMap(resultMapKey, uuid) && num < 10) {
+            uuid = UUID.randomUUID().toString();
+            num++;
         }
-        generateThroughImageTextDTO.setUniqueId(snowflakeId);
+        // 无依据确定的数字
+        if (num > 10){
+            try {
+                Thread.sleep(1000);
+            } catch (InterruptedException e) {
+                throw new RuntimeException(e);
+            }
+            uuid = UUID.randomUUID().toString();
+        }
+        generateThroughImageTextDTO.setUniqueId(uuid);
         String jsonString = JSON.toJSONString(generateThroughImageTextDTO);
 
-        // 4、加入redis排队，便于获取实时排队信息
+        // 3、加入redis排队，便于获取实时排队信息
         Double maxScore = redisUtil.getMaxScore(consumptionOrderKey);
-        redisUtil.addToZSet(consumptionOrderKey, String.valueOf(snowflakeId),maxScore);
+        redisUtil.addToZSet(consumptionOrderKey, uuid, maxScore);
 
-        // 5、将消息发布到MQ消息队列
+        // 4、将消息发布到MQ消息队列
         rabbitMQService.publishMessage(jsonString);
 
-        // 6、返回唯一id
+        // 5、返回唯一id
-        return snowflakeId;
+        return uuid;
     }
 
     @Override
-    public Long getRankPosition(Long uniqueId) {
+    public Long getRankPosition(String uniqueId) {
-        return redisUtil.getRank(consumptionOrderKey, String.valueOf(uniqueId));
+        return redisUtil.getRank(consumptionOrderKey, uniqueId);
     }
 
     @Override
-    public GenerateCollectionVO getGenerateResult(Long uniqueId) {
+    public GenerateCollectionVO getGenerateResult(String uniqueId) {
         // 1、判断该请求是否已经异常
-        Boolean isMember = redisUtil.isElementExistsInMap(exceptionMapKey, String.valueOf(uniqueId));
+        Boolean isMember = redisUtil.isElementExistsInMap(exceptionMapKey, uniqueId);
         if (isMember) {
             throw new BusinessException("generate.interface.error");
         }
 
         // 2、判断该请求是否还在排队
-        Boolean existsInZSet = redisUtil.isElementExistsInZSet(consumptionOrderKey, String.valueOf(uniqueId));
+        Boolean existsInZSet = redisUtil.isElementExistsInZSet(consumptionOrderKey, uniqueId);
         if (existsInZSet) {
             // 排队中，给出当前排序位置
             return new GenerateCollectionVO(getRankPosition(uniqueId) + 1L);
         }
 
         // 3、判断redis中有没有
-        boolean hasHashKey = redisUtil.isElementExistsInMap(resultMapKey, String.valueOf(uniqueId));
+        boolean hasHashKey = redisUtil.isElementExistsInMap(resultMapKey, uniqueId);
         if (hasHashKey) {
             // 3.1 有直接从redis中拿
-            String resultString = redisUtil.getMapValue(resultMapKey, String.valueOf(uniqueId));
+            String resultString = redisUtil.getMapValue(resultMapKey, uniqueId);
             return JSONObject.parseObject(resultString, GenerateCollectionVO.class);
         }
 
@@ -401,7 +411,7 @@ public class GenerateServiceImpl extends ServiceImpl<GenerateMapper, Generate> i
         return new GenerateCollectionVO(generateId, null, generatedCollectionItems);
     }
 
-    public Generate selectByUniqueId(Long uniqueId){
+    public Generate selectByUniqueId(String uniqueId){
         QueryWrapper<Generate> qw = new QueryWrapper<>();
         qw.eq("unique_id",uniqueId);
 
@@ -409,23 +419,24 @@ public class GenerateServiceImpl extends ServiceImpl<GenerateMapper, Generate> i
     }
 
     @Override
-    public void cancelGenerate(Long uniqueId) {
+    public void cancelGenerate(String uniqueId) {
         // 1、确认当前消息是否还在排队中
-        Boolean exists = redisUtil.isElementExistsInZSet(consumptionOrderKey, String.valueOf(uniqueId));
+        Boolean exists = redisUtil.isElementExistsInZSet(consumptionOrderKey, uniqueId);
         if (exists) {
             // 1.1、将需要取消的唯一id加入redis，以便及时取消生成
-            redisUtil.addToSet(cancelSetKey, String.valueOf(uniqueId));
+            redisUtil.addToSet(cancelSetKey, uniqueId);
             // 1.2 将需要取消的id从redis的ConsumptionOrder中删除
-            redisUtil.removeFromZSet(consumptionOrderKey, String.valueOf(uniqueId));
+            redisUtil.removeFromZSet(consumptionOrderKey, uniqueId);
         }else {
             // 2、判断该消息是否异常
-            boolean hasKey = redisUtil.isElementExistsInMap(exceptionMapKey, String.valueOf(uniqueId));
+            boolean hasKey = redisUtil.isElementExistsInMap(exceptionMapKey, uniqueId);
             // 3、判断该消息是否已经消费结束
-            Boolean existsInResult = redisUtil.isElementExistsInMap(resultMapKey, String.valueOf(uniqueId));
+            Boolean existsInResult = redisUtil.isElementExistsInMap(resultMapKey, uniqueId);
             if (!hasKey && !existsInResult){
                 // 设置取等待状态为false
                 AsyncCallerUtil.waitingStatus.put(uniqueId,false);
                 // 3、直接发送取消请求到python端
+                pythonService.cancelGenerateTask(uniqueId);
             }
         }
     }

src/main/resources/application-test.properties

@@ -50,8 +50,8 @@ spring.servlet.multipart.max-request-size= 5MB
 #access.python.ip=http://43.198.80.117
 access.python.ip=http://18.167.251.121
 #access.python.ip=http://18.167.251.121:9991/
-#access.python.port=9992
+access.python.port=9992
-access.python.port=9991
+#access.python.port=9991
 
 # minIO服务配置之信息
 minio.endpoint=https://www.minio.aida.com.hk:9000
@@ -70,7 +70,8 @@ spring.rabbitmq.username=rabbit
 spring.rabbitmq.password=123456
 spring.rabbitmq.virtual-host=/
 
-spring.redis.host=172.31.11.32
+#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