1

src/main/java/com/fdkankan/manage/mq/common/MqQueueUtil.java

@@ -6,5 +6,6 @@ public class MqQueueUtil {
     public static String updateUserCommand = "updateUser";
 
     public static String ucenterScenePayStatusQueue = "scene-pay-status";
+    public static String sceneRestoreQueue = "scene-rest-store";
     public static String laserMoveQueue = "laser-migrate-scene";
 }

src/main/java/com/fdkankan/manage/mq/consumer/SceneRestStoreConsumer.java

@@ -0,0 +1,114 @@
+package com.fdkankan.manage.mq.consumer;
+
+import com.alibaba.fastjson.JSONObject;
+import com.fdkankan.fyun.face.FYunFileServiceInterface;
+import com.fdkankan.manage.common.RedisKeyUtil;
+import com.fdkankan.manage.common.ResultCode;
+import com.fdkankan.manage.exception.BusinessException;
+import com.fdkankan.manage.mq.param.SceneRestStoreVo;
+import com.fdkankan.manage.service.ISceneColdStorageLogService;
+import com.fdkankan.manage.service.ISceneColdStorageService;
+import com.fdkankan.manage.thread.ThreadService;
+import com.fdkankan.redis.util.RedisUtil;
+import com.rabbitmq.client.Channel;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.rabbit.annotation.Queue;
+import org.springframework.amqp.rabbit.annotation.RabbitListener;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * 场景解冻 mq
+ */
+@Slf4j
+@Component
+public class SceneRestStoreConsumer {
+
+    @Autowired
+    FYunFileServiceInterface fYunFileServiceInterface;
+    @Autowired
+    RedisUtil redisUtil;
+    @Autowired
+    ISceneColdStorageService sceneColdStorageService;
+    @Autowired
+    ISceneColdStorageLogService sceneColdStorageLogService;
+
+    @RabbitListener(
+            queuesToDeclare = @Queue("${queue.scene.rest-store:scene-rest-store}"),
+            concurrency = "10"
+    )
+    public void consumerQueue(Channel channel, Message message)  {
+        try {
+            String messageId = message.getMessageProperties().getMessageId();
+            String msg = new String(message.getBody(), StandardCharsets.UTF_8);
+            log.info("scene-restStore-mq--messageId:{},msg:{}",messageId,msg);
+
+            SceneRestStoreVo sceneRestStoreVo = JSONObject.parseObject(msg, SceneRestStoreVo.class);
+            String num = sceneRestStoreVo.getNum();
+            String cloudBucket= sceneRestStoreVo.getCloudBucket();
+            String bucket= sceneRestStoreVo.getBucket();
+            String folderName= sceneRestStoreVo.getFolderName();
+            String redisKey = String.format(RedisKeyUtil.restStoreKey, num);
+            if(redisUtil.hasKey(redisKey)){
+                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
+                return;
+            }
+            redisUtil.set(redisKey,folderName);
+
+            try {
+                fYunFileServiceInterface.restoreFolder(cloudBucket,folderName,1);
+            }catch (Exception e){
+                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
+                return;
+            }
+
+            AtomicReference<Integer> state = new AtomicReference<>(1);
+            AtomicReference<String> reason = new AtomicReference<>("");
+            ScheduledThreadPoolExecutor scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(1);
+            //等待任务执行结束，在间隔2秒执行。
+            scheduledThreadPoolExecutor.scheduleWithFixedDelay(()->{
+                try {
+                    Integer process = fYunFileServiceInterface.getRestoreFolderProcess(cloudBucket, folderName);
+                    System.out.println(process);
+                    log.info("ThreadService-cloudBucket:{},bucket:{},folderName:{},process:{}",cloudBucket,bucket,folderName,process);
+                    if(process == 100){
+                        fYunFileServiceInterface.copyFileBetweenBucket(cloudBucket,folderName,bucket,folderName);
+                        log.info("ThreadService-copy-bucket");
+                        sceneColdStorageService.updateStatus(num,2);
+                        scheduledThreadPoolExecutor.shutdown();
+                    }
+                }catch (Exception e){
+                    log.info("ThreadService-error:",e);
+                    state.set(-1);
+                    reason.set(e.getMessage());
+                    scheduledThreadPoolExecutor.shutdown();
+                    throw new BusinessException(ResultCode.SCENE_NOT_STORE);
+                }finally {
+                    redisUtil.del(redisKey);
+                    sceneColdStorageLogService.saveLog(num,folderName,state,reason);
+                    try {
+                        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
+                    } catch (IOException e) {
+                        e.printStackTrace();
+                    }
+                }
+                //任务启动多久之后   ，周期 每10s执行一次，时间单位
+            },1000,60*1000, TimeUnit.MILLISECONDS);
+
+        }catch (Exception e){
+            log.info("scene-restStore----消费失败",e);
+        }finally {
+
+        }
+
+    }
+
+}

src/main/java/com/fdkankan/manage/mq/param/SceneRestStoreVo.java

@@ -0,0 +1,13 @@
+package com.fdkankan.manage.mq.param;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+
+@Data
+@AllArgsConstructor
+public class SceneRestStoreVo {
+    private String num ;
+    private String cloudBucket;
+    private String bucket;
+    private String folderName;
+}

src/main/java/com/fdkankan/manage/service/ISceneColdStorageLogService.java

@@ -3,6 +3,8 @@ package com.fdkankan.manage.service;
 import com.fdkankan.manage.entity.SceneColdStorageLog;
 import com.baomidou.mybatisplus.extension.service.IService;
 
+import java.util.concurrent.atomic.AtomicReference;
+
 /**
  * <p>
  * 场景冷归档日志表 服务类
@@ -13,4 +15,5 @@ import com.baomidou.mybatisplus.extension.service.IService;
  */
 public interface ISceneColdStorageLogService extends IService<SceneColdStorageLog> {
 
+    void saveLog(String num, String folderName, AtomicReference<Integer> state,AtomicReference<String> reason);
 }

src/main/java/com/fdkankan/manage/service/impl/SceneColdStorageLogServiceImpl.java

@@ -1,5 +1,8 @@
 package com.fdkankan.manage.service.impl;
+import java.util.Date;
+import java.util.concurrent.atomic.AtomicReference;
 
+import com.fdkankan.manage.entity.SceneColdStorage;
 import com.fdkankan.manage.entity.SceneColdStorageLog;
 import com.fdkankan.manage.mapper.ISceneColdStorageLogMapper;
 import com.fdkankan.manage.service.ISceneColdStorageLogService;
@@ -17,4 +20,14 @@ import org.springframework.stereotype.Service;
 @Service
 public class SceneColdStorageLogServiceImpl extends ServiceImpl<ISceneColdStorageLogMapper, SceneColdStorageLog> implements ISceneColdStorageLogService {
 
+    @Override
+    public void saveLog(String num, String folderName, AtomicReference<Integer> state,AtomicReference<String> reason) {
+        SceneColdStorageLog sceneColdStorageLog = new SceneColdStorageLog();
+        sceneColdStorageLog.setNum(num);
+        sceneColdStorageLog.setPath(folderName);
+        sceneColdStorageLog.setType(2);
+        sceneColdStorageLog.setState(state.get());
+        sceneColdStorageLog.setReason(reason.get());
+        this.save(sceneColdStorageLog);
+    }
 }

src/main/java/com/fdkankan/manage/service/impl/SceneProServiceImpl.java

@@ -24,7 +24,9 @@ import com.fdkankan.manage.entity.*;
 import com.fdkankan.manage.mapper.ISceneProMapper;
 import com.fdkankan.manage.mq.common.MqQueueUtil;
 import com.fdkankan.manage.mq.param.ScenePayStatusVo;
+import com.fdkankan.manage.mq.param.SceneRestStoreVo;
 import com.fdkankan.manage.service.*;
+import com.fdkankan.manage.thread.ThreadService;
 import com.fdkankan.manage.util.Dateutils;
 import com.fdkankan.manage.util.ProvinceUtils;
 import com.fdkankan.manage.util.SceneStatusUtil;
@@ -101,6 +103,8 @@ public class SceneProServiceImpl extends ServiceImpl<ISceneProMapper, ScenePro>
     RedisUtil redisUtil;
     @Autowired
     ISceneColdStorageService sceneColdStorageService;
+    @Autowired
+    ThreadService threadService;
 
 
     @Override
@@ -706,6 +710,10 @@ public class SceneProServiceImpl extends ServiceImpl<ISceneProMapper, ScenePro>
         if(StringUtils.isBlank(num)){
             throw new BusinessException(ResultCode.MISSING_REQUIRED_PARAMETERS);
         }
+        String redisKey = String.format(RedisKeyUtil.restStoreKey, num);
+        if(redisUtil.hasKey(redisKey)){
+            throw new BusinessException(ResultCode.SCENE_STORE_ING);
+        }
         SceneColdStorage sceneColdStorage = sceneColdStorageService.getByNum(num);
         if(sceneColdStorage == null){
             throw new BusinessException(ResultCode.SCENE_NOT_STORE);
@@ -715,7 +723,9 @@ public class SceneProServiceImpl extends ServiceImpl<ISceneProMapper, ScenePro>
             throw new BusinessException(ResultCode.SCENE_NOT_EXIST);
         }
         String homePath = dataSource.replace("/mnt/data","home");
-        fYunFileServiceInterface.restoreFolder(sceneColdStorage.getColdBucket(),homePath);
+        SceneRestStoreVo sceneRestStoreVo = new SceneRestStoreVo(num,sceneColdStorage.getColdBucket(),sceneColdStorage.getBucket(),homePath);
+        rabbitMqProducer.sendByWorkQueue(MqQueueUtil.sceneRestoreQueue,sceneRestStoreVo);
+        //threadService.checkRestore(num,sceneColdStorage.getColdBucket(),sceneColdStorage.getBucket(),homePath);
     }
 
     @Override

src/main/java/com/fdkankan/manage/thread/ThreadService.java

@@ -4,16 +4,16 @@ import com.fdkankan.fyun.face.FYunFileServiceInterface;
 import com.fdkankan.manage.common.RedisKeyUtil;
 import com.fdkankan.manage.common.ResultCode;
 import com.fdkankan.manage.exception.BusinessException;
+import com.fdkankan.manage.service.ISceneColdStorageLogService;
+import com.fdkankan.manage.service.ISceneColdStorageService;
 import com.fdkankan.redis.util.RedisUtil;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Service;
 
-import java.util.ArrayList;
-import java.util.List;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
 
 @Service
 @Slf4j
@@ -22,6 +22,10 @@ public class ThreadService {
     FYunFileServiceInterface fYunFileServiceInterface;
     @Autowired
     RedisUtil redisUtil;
+    @Autowired
+    ISceneColdStorageService sceneColdStorageService;
+    @Autowired
+    ISceneColdStorageLogService sceneColdStorageLogService;
 
 
     public void checkRestore(String num,String cloudBucket,String bucket,String folderName){
@@ -30,6 +34,8 @@ public class ThreadService {
             return;
         }
         redisUtil.set(redisKey,folderName);
+        AtomicReference<Integer> state = new AtomicReference<>(1);
+        AtomicReference<String> reason = new AtomicReference<>("");
         ScheduledThreadPoolExecutor scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(1);
         //等待任务执行结束，在间隔2秒执行。
         scheduledThreadPoolExecutor.scheduleWithFixedDelay(()->{
@@ -40,14 +46,18 @@ public class ThreadService {
                 if(process == 100){
                     fYunFileServiceInterface.copyFileBetweenBucket(cloudBucket,folderName,bucket,folderName);
                     log.info("ThreadService-copy-bucket");
+                    sceneColdStorageService.updateStatus(num,2);
                     scheduledThreadPoolExecutor.shutdown();
                 }
             }catch (Exception e){
                 log.info("ThreadService-error:",e);
+                state.set(-1);
+                reason.set(e.getMessage());
                 scheduledThreadPoolExecutor.shutdown();
                 throw new BusinessException(ResultCode.SCENE_NOT_STORE);
             }finally {
                 redisUtil.del(redisKey);
+                sceneColdStorageLogService.saveLog(num,folderName,state,reason);
             }
 
             //任务启动多久之后   ，周期 每10s执行一次，时间单位