添加时间过长超时捕获不到的问题

data-service/pom.xml

@@ -85,6 +85,14 @@
             <groupId>org.redisson</groupId>
             <artifactId>redisson</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.springframework.amqp</groupId>
+            <artifactId>spring-rabbit</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.amqp</groupId>
+            <artifactId>spring-amqp</artifactId>
+        </dependency>
     </dependencies>
 
     <build>

data-service/src/main/java/com/zksy/data/annotation/XhLogAspect.java

@@ -5,6 +5,7 @@ import cn.hutool.core.util.StrUtil;
 import com.zksy.data.domain.po.XhDataLog;
 import com.zksy.data.mapper.XhDataLogMapper;
 import com.zksy.data.service.XhDataLogService;
+import com.zksy.data.utils.PublisherDelayMessage;
 import lombok.extern.slf4j.Slf4j;
 import org.aspectj.lang.ProceedingJoinPoint;
 import org.aspectj.lang.annotation.Around;
@@ -28,6 +29,8 @@ public class XhLogAspect {
     private XhDataLogMapper xhDataLogMapper;
     @Autowired
     private RedisTemplate<String,String> redisTemplate;
+    @Autowired
+    private PublisherDelayMessage publisherDelayMessage;
 
     public void saveBeforeLog(XhDataLog xhDataLog) {
         xhDataLogMapper.insert(xhDataLog);
@@ -59,6 +62,7 @@ public class XhLogAspect {
 
         CompletableFuture.runAsync(() -> {
             saveBeforeLog(xhDataLog);
+            publisherDelayMessage.emitDelayMessage(id,7200000);
         });
         try {
             joinPoint.proceed();
@@ -74,7 +78,7 @@ public class XhLogAspect {
             xhDataLog.setUseTime((int) (System.currentTimeMillis() - xhDataLog.getCreateTime().getTime()));
             xhDataLog.setUpdateTime(new Date());
             xhDataLogMapper.updateById(xhDataLog);
-            Thread.sleep(5000);
+            Thread.sleep(20000);
             redisTemplate.delete(firstParam);
         }
     }

data-service/src/main/java/com/zksy/data/controller/Test2Controller.java

@@ -1,9 +1,6 @@
 package com.zksy.data.controller;
 
-import com.zksy.data.service.XcrAbnormalBusinessOperationsService;
-import com.zksy.data.service.XcrAdministrativeLicensingService;
-import com.zksy.data.service.XcrIncomeTaxReturnService;
-import com.zksy.data.service.XcrMainFinancialStatementsService;
+import com.zksy.data.service.*;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.RequestMapping;
@@ -20,12 +17,15 @@ public class Test2Controller {
     private XcrIncomeTaxReturnService xcrIncomeTaxReturnService;
     @Autowired
     private XcrAdministrativeLicensingService xcrAdministrativeLicensingService;
+    @Autowired
+    private XcrAbnormalHouseholdIdentificationService xcrAbnormalHouseholdIdentificationService;
     @GetMapping("/a")
     public String hello() {
 //        xcrAbnormalBusinessOperationsService.saveDataByUniCode("gateway/api/1/getQyycmlxxxxByUniscid");
 //        xcrMainFinancialStatementsService.saveDataByUniCode("gateway/api/1/getCwbbzbByUniscid");
 //        xcrIncomeTaxReturnService.saveDataByUniCode("gateway/api/1/getQysdssbxxByUniscid1");
-        xcrAdministrativeLicensingService.saveDataByUniCode("interface/v1/doublePublic/findLicensingList");
+//        xcrAdministrativeLicensingService.saveDataByUniCode("interface/v1/doublePublic/findLicensingList");
+        xcrAbnormalHouseholdIdentificationService.saveDataByUniCode("gateway/api/1/getFzchrdxxByUniscid");
         return "hello";
     }
 }

data-service/src/main/java/com/zksy/data/controller/Test3Controller.java

@@ -0,0 +1,19 @@
+package com.zksy.data.controller;
+
+import com.zksy.data.utils.PublisherDelayMessage;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+@RestController
+@RequestMapping("/test3")
+public class Test3Controller {
+    @Autowired
+    private PublisherDelayMessage publisherDelayMessage;
+    @GetMapping("/a")
+    public String hello(String msg){
+        publisherDelayMessage.emitDelayMessage(msg,5000);
+        return "hello world";
+    }
+}

data-service/src/main/java/com/zksy/data/listener/LogListener.java

@@ -0,0 +1,40 @@
+package com.zksy.data.listener;
+
+import com.zksy.data.service.XhDataLogService;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.amqp.rabbit.annotation.Exchange;
+import org.springframework.amqp.rabbit.annotation.Queue;
+import org.springframework.amqp.rabbit.annotation.QueueBinding;
+import org.springframework.amqp.rabbit.annotation.RabbitListener;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Configuration;
+
+import java.util.Date;
+import java.util.Optional;
+
+@Configuration
+@Slf4j
+public class LogListener {
+    @Autowired
+    private XhDataLogService xhDataLogService;
+    @RabbitListener(bindings = @QueueBinding(
+            value = @Queue(name = "delay.queue", durable = "true"),
+            exchange = @Exchange(name = "delay.direct", delayed = "true"),
+            key = "delay"
+    ))
+    public void listenDelayIdForLog(String id){
+        Optional.ofNullable(id)
+                .flatMap(i -> Optional.ofNullable(xhDataLogService.getById(i)))
+                .filter(log -> System.currentTimeMillis() - log.getCreateTime().getTime() > 7200000L)
+                .filter(log -> log.getIsSuccess() == 0)
+                .filter(log -> log.getIsRunning() == 1)
+                .ifPresent(log -> {
+                    log.setIsRunning(0);
+                    log.setUseTime((int) (System.currentTimeMillis() - log.getCreateTime().getTime()));
+                    log.setUpdateTime(new Date());
+                    log.setException("超时异常");
+                    xhDataLogService.updateById(log);
+                });
+        log.info("设置id为：{}的日志为未完成", id);
+    }
+}

data-service/src/main/java/com/zksy/data/utils/PublisherDelayMessage.java

@@ -0,0 +1,17 @@
+package com.zksy.data.utils;
+
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+@Component
+public class PublisherDelayMessage {
+    @Autowired
+    private RabbitTemplate rabbitTemplate;
+    public void emitDelayMessage(String message,Integer duration) {
+        rabbitTemplate.convertAndSend("delay.direct", "delay", message, q -> {
+            q.getMessageProperties().setDelay(duration);
+            return q;
+        });
+    }
+}

data-service/src/main/java/com/zksy/data/utils/XhRequestUtil.java

@@ -97,7 +97,7 @@ public class XhRequestUtil {
         if(redisTemplate.opsForValue().get(address) !=null){
             throw new CommonException("停止并行流",5001);
         }
-        log.info("正在同步:{}下的:{}", address, uniCode);
+//        log.info("正在同步:{}下的:{}", address, uniCode);
         ArrayList<K> resList = new ArrayList<>();
         if (StrUtil.isEmpty(uniCode)) {
             log.warn("统一社会信用代码为空");

pom.xml

@@ -35,8 +35,6 @@
       <redis.version>3.0.5</redis.version>
       <sentinel.version>2021.0.4.0</sentinel.version>
       <minioutil.version>1.0.0</minioutil.version>
-      <mqtt.version>5.5.9</mqtt.version>
-      <netty.version>4.1.75.Final</netty.version>
       <redisson.version>3.13.6</redisson.version>
       <spring-aop.version>2.7.12</spring-aop.version>
    </properties>
@@ -109,6 +107,18 @@
             <artifactId>spring-boot-starter-aop</artifactId>
             <version>${spring-aop.version}</version>
          </dependency>
+         <!--AMQP依赖-->
+         <dependency>
+            <groupId>org.springframework.amqp</groupId>
+            <artifactId>spring-amqp</artifactId>
+            <version>2.4.12</version>
+         </dependency>
+         <!--Spring整合Rabbit依赖-->
+         <dependency>
+            <groupId>org.springframework.amqp</groupId>
+            <artifactId>spring-rabbit</artifactId>
+            <version>2.4.12</version>
+         </dependency>
       </dependencies>
    </dependencyManagement>