增加线下版本队列处理v3

sxz-application/src/main/resources/application-dev.properties

@@ -34,6 +34,7 @@ spring.datasource.hikari.connection-timeout=30000
 #rabbitmq
 spring.rabbitmq.host=192.168.0.47
 spring.rabbitmq.port=5672
+spring.rabbitmq.apiport=15672
 spring.rabbitmq.username=guest
 spring.rabbitmq.password=guest
 # 开启重试

sxz-application/src/main/resources/application-prod.properties

@@ -34,12 +34,17 @@ spring.datasource.hikari.connection-timeout=30000
 #rabbitmq
 spring.rabbitmq.host=127.0.0.1
 spring.rabbitmq.port=5672
+spring.rabbitmq.apiport=15672
 spring.rabbitmq.username=admin
 spring.rabbitmq.password=sxz12341
-# 开启重试
-spring.rabbitmq.listener.simple.retry.enabled=false
-# 重试次数,默认为3次
-spring.rabbitmq.listener.simple.retry.max-attempts=1
+##??????
+spring.rabbitmq.listener.simple.retry.max-attempts=3
+##???????????false????????????????????????????
+spring.rabbitmq.listener.simple.retry.enabled=true
+## ????????????
+spring.rabbitmq.listener.simple.retry.initial-interval=5000
+## ??????????????????false??????????????????????
+spring.rabbitmq.listener.simple.default-requeue-rejected=false
 ##mybatis-plus配置 ##
 mybatis-plus.configuration.log-impl=org.apache.ibatis.logging.slf4j.Slf4jImpl
 mybatis-plus.configuration.map-underscore-to-camel-case=true

sxz-application/src/main/resources/application-uat.properties

@@ -34,12 +34,17 @@ spring.datasource.hikari.connection-timeout=30000
 #rabbitmq
 spring.rabbitmq.host=192.168.0.47
 spring.rabbitmq.port=5672
+spring.rabbitmq.apiport=15672
 spring.rabbitmq.username=guest
 spring.rabbitmq.password=guest
-# 开启重试
-spring.rabbitmq.listener.simple.retry.enabled=false
-# 重试次数,默认为3次
-spring.rabbitmq.listener.simple.retry.max-attempts=1
+##??????
+spring.rabbitmq.listener.simple.retry.max-attempts=3
+##???????????false????????????????????????????
+spring.rabbitmq.listener.simple.retry.enabled=true
+## ????????????
+spring.rabbitmq.listener.simple.retry.initial-interval=5000
+## ??????????????????false??????????????????????
+spring.rabbitmq.listener.simple.default-requeue-rejected=false
 ##mybatis-plus配置 ##
 mybatis-plus.configuration.log-impl=org.apache.ibatis.logging.slf4j.Slf4jImpl
 mybatis-plus.configuration.map-underscore-to-camel-case=true

sxz-base/src/main/java/com/fdkk/sxz/util/RabbitMqUtil.java

@@ -0,0 +1,120 @@
+package com.fdkk.sxz.util;
+
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONObject;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Component;
+import sun.misc.BASE64Encoder;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.net.HttpURLConnection;
+import java.net.URL;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * @author Xiewj
+ * @date 2022/3/14
+ */
+@Slf4j
+@Component
+public class RabbitMqUtil {
+
+    @Value("${spring.rabbitmq.host}")
+    private String host;
+    @Value("${spring.rabbitmq.apiport}")
+    private String port;
+    @Value("${spring.rabbitmq.username}")
+    private String username;
+    @Value("${spring.rabbitmq.password}")
+    private String password;
+
+    /**
+     * 队列任务总数 * * @param queueName * @return
+     */
+    public int getMessageCount(String queueName) throws IOException {
+        String apiMessage = getApiMessage(queueName);
+        if (Objects.equals(apiMessage, "")) {
+            RabbitMqUtil.log.error("请求RabbitMQ API时出错！！");
+            return 0;
+        }
+        JSONObject jsonObject = JSON.parseObject(apiMessage);
+        return Integer.parseInt(jsonObject.get("messages").toString());
+    }
+
+    /**
+     * 队列ready任务数 * * @param queueName * @return
+     */
+    public int getMessageReadyCount(String queueName) throws IOException {
+        String apiMessage = getApiMessage(queueName);
+        if (Objects.equals(apiMessage, "")) {
+            RabbitMqUtil.log.error("请求RabbitMQ API时出错！！");
+            return 0;
+        }
+        JSONObject jsonObject = JSON.parseObject(apiMessage);
+        return Integer.parseInt(jsonObject.get("messages_ready").toString());
+    }
+
+    /**
+     * 队列unack数MQ * * @param queueName * @return
+     */
+    public int getMessagesUnacknowledgedCount(String queueName) throws IOException {
+        String apiMessage = getApiMessage(queueName);
+        if (Objects.equals(apiMessage, "")) {
+            RabbitMqUtil.log.error("请求RabbitMQ API时出错！！");
+            return 0;
+        }
+        JSONObject jsonObject = JSON.parseObject(apiMessage);
+        return Integer.parseInt(jsonObject.get("messages_unacknowledged").toString());
+    }
+
+    /**
+     * 获取队列消息总数、ready消息数、unack消息数 * * @param queueName * @return Map<String,Integer>
+     */
+    public Map<String, Integer> getMQCountMap(String queueName) throws IOException {
+        String apiMessage = getApiMessage(queueName);
+        JSONObject jsonObject = JSON.parseObject(apiMessage);
+        Map<String, Integer> map = new HashMap<>();
+        map.put("messages", Integer.parseInt(jsonObject.get("messages").toString()));
+        map.put("messages_ready", Integer.parseInt(jsonObject.get("messages_ready").toString()));
+        map.put("messages_unacknowledged", Integer.parseInt(jsonObject.get("messages_unacknowledged").toString()));
+        return map;
+    }
+
+    public String getApiMessage(String queueName) throws IOException {
+        //发送一个GET请求
+        HttpURLConnection httpConn = null;
+        BufferedReader in = null;
+
+        String urlString = "http://" + host + ":" + port + "/api/queues/%2f/" + queueName;
+        URL url = new URL(urlString);
+        httpConn = (HttpURLConnection) url.openConnection();
+        //设置用户名密码
+        String auth = username + ":" + password;
+        BASE64Encoder enc = new BASE64Encoder();
+        String encoding = enc.encode(auth.getBytes());
+        httpConn.setDoOutput(true);
+        httpConn.setRequestProperty("Authorization", "Basic " + encoding);
+        // 创建实际的链接
+        httpConn.connect();
+        //读取响应
+        if (httpConn.getResponseCode() == HttpURLConnection.HTTP_OK) {
+            StringBuilder content = new StringBuilder();
+            String tempStr = "";
+            in = new BufferedReader(new InputStreamReader(httpConn.getInputStream()));
+            while ((tempStr = in.readLine()) != null) {
+                content.append(tempStr);
+            }
+            in.close();
+            httpConn.disconnect();
+            return content.toString();
+        } else {
+            httpConn.disconnect();
+            return "";
+        }
+    }
+}

sxz-core/src/main/java/com/fdkk/sxz/other/listener/Model3dBuild.java

@@ -47,6 +47,8 @@ public class Model3dBuild {
 
     @Autowired
     private ICustomComponentService customComponentService;
+    @Autowired
+    RedisUtil redisUtil;
 
 
     @RabbitHandler
@@ -65,7 +67,6 @@ public class Model3dBuild {
                 RedisUtil.unlock("modelBiz3d:" + message.getPayload().getName(), token);
             }
         }
-
     }
 
     public void modelBiz3dHandler(RequestQueue str) throws InterruptedException {

sxz-core/src/main/java/com/fdkk/sxz/webApi/controller/QueueController.java

@@ -7,6 +7,8 @@ import com.fdkk.sxz.annotation.log.AroundLog;
 import com.fdkk.sxz.base.Result;
 import com.fdkk.sxz.other.mq.TopicRabbitConfig;
 import com.fdkk.sxz.util.OkHttpUtils;
+import com.fdkk.sxz.util.RabbitMqUtil;
+import com.fdkk.sxz.util.RedisUtil;
 import com.fdkk.sxz.vo.request.RequestQueue;
 import io.swagger.annotations.Api;
 import io.swagger.annotations.ApiOperation;
@@ -21,6 +23,8 @@ import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RequestMethod;
 import org.springframework.web.bind.annotation.RestController;
 
+import java.io.IOException;
+
 /**
  * @author Xiewj
  * @date 2021/10/19
@@ -35,6 +39,10 @@ public class QueueController {
     private RabbitTemplate rabbitTemplate;  //使用RabbitTemplate,这提供了接收/发送等等方
     @Value("${build.url}")
     private String buildUrl;
+    @Autowired
+    RedisUtil redisUtil;
+    @Autowired
+    RabbitMqUtil rabbitMqUtil;
 
     /**
      * 数据上传
@@ -45,10 +53,26 @@ public class QueueController {
     @RequestMapping(value = "/dataHandle", method = RequestMethod.POST)
     @NoAuthentication
     @AroundLog(name = "数据处理任务接收")
-    public Result dataHandle(@RequestBody RequestQueue params) {
+    public Result dataHandle(@RequestBody RequestQueue params) throws IOException {
         QueueController.log.info("数据处理任务接收:{}", params);
         Message message = MessageBuilder.withPayload(params).build();
         rabbitTemplate.convertAndSend(TopicRabbitConfig.TOPICE, TopicRabbitConfig.MODEL_BIZ_3D, message);
+        return Result.success("处理结果", rabbitMqUtil.getMessageCount(TopicRabbitConfig.MODEL_BIZ_3D));
+    }
+
+    /**
+     * 数据上传
+     *
+     * @return
+     */
+    @ApiOperation("test")
+    @RequestMapping(value = "/test", method = RequestMethod.POST)
+    @NoAuthentication
+    @AroundLog(name = "test")
+    public Result test() throws IOException {
+        System.out.println(rabbitMqUtil.getMessageCount(TopicRabbitConfig.MODEL_BIZ_3D));
+        System.out.println(rabbitMqUtil.getMessageReadyCount(TopicRabbitConfig.MODEL_BIZ_3D));
+        System.out.println(rabbitMqUtil.getMessagesUnacknowledgedCount(TopicRabbitConfig.MODEL_BIZ_3D));
         return Result.success("处理结果");
     }