解冻mq

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

@@ -21,6 +21,7 @@ import org.springframework.stereotype.Component;
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
 import java.util.Arrays;
+import java.util.Date;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
@@ -45,7 +46,7 @@ public class SceneRestStoreConsumer {
             queuesToDeclare = @Queue("${queue.scene.rest-store:scene-rest-store}"),
             concurrency = "1"
     )
-    public void consumerQueue(Channel channel, Message message)  {
+    public void consumerQueue(Channel channel, Message message) throws IOException {
         try {
             String messageId = message.getMessageProperties().getMessageId();
             String msg = new String(message.getBody(), StandardCharsets.UTF_8);
@@ -58,55 +59,34 @@ public class SceneRestStoreConsumer {
             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();
-                    }
+            Long startTime = new Date().getTime();
+            Integer process = fYunFileServiceInterface.getRestoreFolderProcess(cloudBucket, folderName);
+            log.info("scene-restStore-mq-cloudBucket:{},bucket:{},folderName:{},process:{}",cloudBucket,bucket,folderName,process);
+            while (process != 100 ){
+                Long time = new Date().getTime();
+                if(time - startTime >1000 * 60){
+                    process = fYunFileServiceInterface.getRestoreFolderProcess(cloudBucket, folderName);
+                    log.info("scene-restStore-mq-cloudBucket:{},bucket:{},folderName:{},process:{}",cloudBucket,bucket,folderName,process);
                 }
-                //任务启动多久之后   ，周期 每10s执行一次，时间单位
-            },1000,60*1000, TimeUnit.MILLISECONDS);
+            }
+            fYunFileServiceInterface.copyFileBetweenBucket(cloudBucket,folderName,bucket,folderName);
+            log.info("Tscene-restStore-mq-copy-bucket");
+            sceneColdStorageService.updateStatus(num,2);
+            sceneColdStorageLogService.saveLog(num,folderName);
+            redisUtil.del(redisKey);
 
         }catch (Exception e){
             log.info("scene-restStore----消费失败",e);
         }finally {
-
+            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
         }
 
     }

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

@@ -15,5 +15,5 @@ import java.util.concurrent.atomic.AtomicReference;
  */
 public interface ISceneColdStorageLogService extends IService<SceneColdStorageLog> {
 
-    void saveLog(String num, String folderName, AtomicReference<Integer> state,AtomicReference<String> reason);
+    void saveLog(String num, String folderName);
 }

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

@@ -21,13 +21,13 @@ import org.springframework.stereotype.Service;
 public class SceneColdStorageLogServiceImpl extends ServiceImpl<ISceneColdStorageLogMapper, SceneColdStorageLog> implements ISceneColdStorageLogService {
 
     @Override
-    public void saveLog(String num, String folderName, AtomicReference<Integer> state,AtomicReference<String> reason) {
+    public void saveLog(String num, String folderName) {
         SceneColdStorageLog sceneColdStorageLog = new SceneColdStorageLog();
         sceneColdStorageLog.setNum(num);
         sceneColdStorageLog.setPath(folderName);
         sceneColdStorageLog.setType(2);
-        sceneColdStorageLog.setState(state.get());
-        sceneColdStorageLog.setReason(reason.get());
+        sceneColdStorageLog.setState(1);
+        sceneColdStorageLog.setReason("");
         this.save(sceneColdStorageLog);
     }
 }