Merge remote-tracking branch 'origin/test'

张志伟
2 parents 3fa994dd 5e8c709c
Showing 12 changed files with 175 additions and 27 deletions
fw-hermes-common/pom.xml
fw-hermes-common/src/main/java/cn/fw/hermes/common/utils/RobotHelper.java
fw-hermes-common/src/main/java/cn/fw/hermes/common/utils/ThreadPoolUtil.java
fw-hermes-server/src/main/java/cn/fw/hermes/config/ScheduleConfig.java
fw-hermes-server/src/main/java/cn/fw/hermes/controller/erp/SysDebugController.java
fw-hermes-server/src/main/java/cn/fw/hermes/task/MessageSendTask.java
fw-hermes-server/src/main/resources/application-prd.yml
fw-hermes-server/src/main/resources/application.yml
fw-hermes-service/pom.xml
fw-hermes-service/src/main/java/cn/fw/hermes/service/biz/CommonBizService.java
fw-hermes-service/src/main/java/cn/fw/hermes/service/emqx/MqttService.java
fw-hermes-service/src/main/java/cn/fw/hermes/service/property/MqttProperty.java
@@ -25,6 +25,10 @@
             <groupId>cn.fw</groupId>
             <artifactId>fw-data-base</artifactId>
         </dependency>
+        <dependency>
+            <groupId>commons-codec</groupId>
+            <artifactId>commons-codec</artifactId>
+        </dependency>
     </dependencies>
  
     <build>
@@ -38,4 +42,4 @@
             </plugin>
         </plugins>
     </build>
-</project>
 \ No newline at end of file
+</project>
+package cn.fw.hermes.common.utils;
+
+import cn.hutool.http.HttpRequest;
+import cn.hutool.http.HttpUtil;
+import cn.hutool.json.JSONUtil;
+import org.apache.commons.codec.binary.Base64;
+
+import javax.crypto.Mac;
+import javax.crypto.spec.SecretKeySpec;
+import java.net.URLEncoder;
+import java.nio.charset.StandardCharsets;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * 机器人工具
+ *
+ * @author : kurisu
+ * @version : 2.0
+ * @desc : 机器人工具
+ * @date : 2023-09-08 14:40
+ */
+public class RobotHelper {
+    private static final String DING_TALK_ROBOT_API = "https://oapi.dingtalk.com/robot/send";
+
+    public static String sendEarlyWarning(String accessToken, String secret) throws Exception {
+        Long timestamp = System.currentTimeMillis();
+        String sign = createSign(timestamp, secret);
+        String url = String.format("%s?access_token=%s&timestamp=%s&sign=%s", DING_TALK_ROBOT_API, accessToken, timestamp, sign);
+
+        Map<String, String> markdown = new HashMap<>();
+        markdown.put("title", "预警信息");
+        markdown.put("text", "### MQTT服务断开连接 \n" +
+            " ##### 重连次数超过10次\n" +
+            " ###### " + DateUtil.getFormatString(new Date(), "MM月dd日 HH点mm分") +
+            " 发布  \n");
+
+        Map<String, Object> params = new HashMap<>();
+        params.put("msgtype", "markdown");
+        params.put("markdown", markdown);
+        HttpRequest post = HttpUtil.createPost(url);
+        post.body(JSONUtil.toJsonStr(params), "application/json");
+
+        return post.setConnectionTimeout(-1).execute().body();
+    }
+
+    public static void main(String[] args) throws Exception {
+        String res = sendEarlyWarning("31a16df234bc4b78a29c5a4cc35a7421fc90a87cc339dbb545117b6966d2ca32", "SEC509e8cd6d0c3969563acf9e33c177c49cf2a365cd60ddc841d4eb728afcd6940");
+        System.out.println(res);
+    }
+
+    private static String createSign(Long timestamp, String secret) throws Exception {
+        String stringToSign = timestamp + "\n" + secret;
+        Mac mac = Mac.getInstance("HmacSHA256");
+        mac.init(new SecretKeySpec(secret.getBytes(StandardCharsets.UTF_8), "HmacSHA256"));
+        byte[] signData = mac.doFinal(stringToSign.getBytes(StandardCharsets.UTF_8));
+        return URLEncoder.encode(new String(Base64.encodeBase64(signData)), "UTF-8");
+    }
+}
@@ -4,7 +4,6 @@ import cn.hutool.core.thread.ExecutorBuilder;
 import cn.hutool.core.thread.NamedThreadFactory;
  
 import java.util.concurrent.ExecutorService;
-import java.util.concurrent.LinkedBlockingQueue;
  
 /**
  * ThreadPoolUtil
@@ -21,12 +20,12 @@ public class ThreadPoolUtil {
  
     private ThreadPoolUtil() {
         this.executor = ExecutorBuilder.create()
-                .setCorePoolSize(20)
-                .setMaxPoolSize(200)
-                .setThreadFactory(new NamedThreadFactory("Hermes-Custom-", false))
-                .setAllowCoreThreadTimeOut(true)
-                .setWorkQueue(new LinkedBlockingQueue<>(4096))
-                .build();
+            .setCorePoolSize(2 << 5)
+            .setMaxPoolSize(2 << 6)
+            .setThreadFactory(new NamedThreadFactory("Hermes-Custom-", false))
+            .setAllowCoreThreadTimeOut(true)
+            .useArrayBlockingQueue(2 << 10)
+            .build();
     }
  
     public static ThreadPoolUtil getInstance() {
@@ -26,11 +26,11 @@ public class ScheduleConfig {
     /**
      * 核心线程数（默认线程数）
      */
-    private static final int corePoolSize = 25;
+    private static final int corePoolSize = 2 << 4;
     /**
      * 最大线程数
      */
-    private static final int maxPoolSize = 50;
+    private static final int maxPoolSize = 2 << 7;
     /**
      * 允许线程空闲时间（单位：默认为秒）
      */
@@ -38,7 +38,7 @@ public class ScheduleConfig {
     /**
      * 缓冲队列大小
      */
-    private static final int queueCapacity = 20480;
+    private static final int queueCapacity = 2 << 10;
     /**
      * 线程池名前缀
      */
@@ -74,4 +74,17 @@ public class SysDebugController {
         commonBizService.saveMsgBizType(type, typeDesc);
         return success();
     }
+
+    @ControllerMethod("重连mqtt服务器")
+    @PostMapping("/client/reconnect")
+    public Message<Boolean> reconnect() {
+        return success(commonBizService.reconnect());
+    }
+
+    @ControllerMethod("模拟发送预警消息")
+    @PostMapping("/send/warning/msg")
+    public Message<Void> sendWarningMsg() {
+        commonBizService.debugSendWarningMsg();
+        return success();
+    }
 }
@@ -66,7 +66,7 @@ public class MessageSendTask {
      * 推送消息
      */
     @Async("taskExecutor")
-    @Scheduled(cron = "0/5 * * * * *")
+    @Scheduled(cron = "0/2 * * * * *")
     public void sendMsg() {
         List<String> failList = new ArrayList<>();
         String jsonStr;
@@ -24,3 +24,7 @@ rocketmq:
 mqtt:
   host: tcp://192.168.0.55
   port: 1883
+  robot:
+    enabled: true
+    access-token: 31a16df234bc4b78a29c5a4cc35a7421fc90a87cc339dbb545117b6966d2ca32
+    secret: SEC509e8cd6d0c3969563acf9e33c177c49cf2a365cd60ddc841d4eb728afcd6940
@@ -148,6 +148,10 @@ mqtt:
   clientId: hermes
   timeout: 10
   keepalive: 30
+  robot:
+    enabled: false
+    access-token:
+    secret:
  
 disturb:
   start: '22:00'
@@ -27,7 +27,6 @@
             <groupId>cn.fw</groupId>
             <artifactId>fw-hermes-rpc</artifactId>
         </dependency>
-
         <dependency>
             <groupId>org.apache.rocketmq</groupId>
             <artifactId>rocketmq-spring-boot-starter</artifactId>
@@ -74,4 +73,4 @@
             </plugin>
         </plugins>
     </build>
-</project>
 \ No newline at end of file
+</project>
@@ -14,6 +14,7 @@ import cn.fw.hermes.domain.vo.TopicVO;
 import cn.fw.hermes.service.data.MessageBusinessTypeService;
 import cn.fw.hermes.service.data.SysPlatformService;
 import cn.fw.hermes.service.data.SysTopicService;
+import cn.fw.hermes.service.emqx.MqttService;
 import com.baomidou.mybatisplus.core.toolkit.Wrappers;
 import lombok.Getter;
 import lombok.RequiredArgsConstructor;
@@ -44,6 +45,7 @@ public class CommonBizService {
     private final SysTopicService sysTopicService;
     private final MessageBusinessTypeService messageBusinessTypeService;
     private final CacheExtender cache;
+    private final MqttService mqttService;
  
     @Value("${spring.cache.locker.key-prefix}:common")
     @Getter
@@ -121,7 +123,7 @@ public class CommonBizService {
     @DisLock(prefix = "#this.getKeyPrefix()", key = "'msg_biz_type'", message = "请勿重复操作")
     public void saveMsgBizType(String type, String desc) {
         boolean exist = messageBusinessTypeService.count(Wrappers.<MessageBusinessType>lambdaQuery()
-                .eq(MessageBusinessType::getType, type)
+            .eq(MessageBusinessType::getType, type)
         ) > 0;
         BV.isFalse(exist, () -> "已存在该类型");
         MessageBusinessType bizType = new MessageBusinessType();
@@ -129,4 +131,24 @@ public class CommonBizService {
         bizType.setTypeDesc(desc);
         messageBusinessTypeService.save(bizType);
     }
+
+    /**
+     * 重连mqtt服务器
+     *
+     * @return
+     */
+    @DisLock(prefix = "#this.getKeyPrefix()", key = "'msg_client_reconnect'", message = "请勿重复操作")
+    public boolean reconnect() {
+        return mqttService.manualReConnect();
+    }
+
+    /**
+     * 模拟发送测试预警钉钉机器人消息
+     *
+     * @return
+     */
+    @DisLock(prefix = "#this.getKeyPrefix()", key = "'debug_send_warning_msg'", message = "请勿重复操作")
+    public void debugSendWarningMsg() {
+        mqttService.sendWarningMsg();
+    }
 }
 package cn.fw.hermes.service.emqx;
  
 import cn.fw.hermes.common.constant.Constant;
+import cn.fw.hermes.common.utils.RobotHelper;
 import cn.fw.hermes.common.utils.ThreadPoolUtil;
 import cn.fw.hermes.domain.mqtt.Message;
 import cn.fw.hermes.domain.mqtt.UpdateMessage;
@@ -16,6 +17,7 @@ import org.springframework.stereotype.Service;
  
 import java.nio.charset.StandardCharsets;
 import java.util.List;
+import java.util.Objects;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicBoolean;
  
@@ -54,6 +56,24 @@ public class MqttService implements MqttCallbackExtended {
         }
     }
  
+    /**
+     * 手动重连
+     *
+     * @return
+     */
+    public boolean manualReConnect() {
+        if (mqttAsyncClient.isConnected()) {
+            return true;
+        }
+        try {
+            mqttAsyncClient.reconnect();
+            return true;
+        } catch (MqttException e) {
+            log.info("[{}]手动重连失败", TAG, e);
+        }
+        return false;
+    }
+
     public void disconnect() {
         try {
             if (mqttAsyncClient.isConnected()) {
@@ -161,7 +181,7 @@ public class MqttService implements MqttCallbackExtended {
         if (!mqttAsyncClient.isConnected()) {
             this.reconnecting.set(false);
             log.info("[{}]重连次数耗尽", TAG);
-            // TODO: 2023/7/21 钉钉消息通知
+            sendWarningMsg();
         }
     }
  
@@ -213,4 +233,24 @@ public class MqttService implements MqttCallbackExtended {
             }, ThreadPoolUtil.getInstance().getExecutor());
         }
     }
+
+    /**
+     * 手动发送预警消息
+     *
+     * @return
+     */
+    public void sendWarningMsg() {
+        MqttProperty.RobotConfig robot = mqttProperty.getRobot();
+        log.info("机器人配置信息：[{}]", robot);
+        if (Objects.nonNull(robot) && robot.isEnabled()) {
+            try {
+                String res = RobotHelper.sendEarlyWarning(robot.getAccessToken(), robot.getSecret());
+                log.info("发送预警消息：[{}]", res);
+            } catch (Exception e) {
+                log.info("发送预警失败", e);
+            }
+        } else {
+            log.info("机器人预警通知未启用：[{}]", robot);
+        }
+    }
 }
 package cn.fw.hermes.service.property;
  
+import lombok.Data;
 import lombok.Getter;
+import lombok.ToString;
 import org.springframework.boot.context.properties.ConfigurationProperties;
  
 /**
@@ -12,6 +14,7 @@ import org.springframework.boot.context.properties.ConfigurationProperties;
  * @date : 2023-07-20 18:03
  */
 @Getter
+@ToString
 @ConfigurationProperties(prefix = "mqtt")
 public class MqttProperty {
     private String host = "localhost";
@@ -21,6 +24,7 @@ public class MqttProperty {
     private int timeout = 10;
     private int keepalive = 20;
     private String clientId;
+    private RobotConfig robot;
  
     public void setHost(String host) {
         this.host = host;
@@ -50,16 +54,15 @@ public class MqttProperty {
         this.clientId = clientId;
     }
  
-    @Override
-    public String toString() {
-        return "MqttProperty{" +
-                "host='" + host + '\'' +
-                ", userName='" + userName + '\'' +
-                ", password='" + password + '\'' +
-                ", port=" + port +
-                ", timeout=" + timeout +
-                ", keepalive=" + keepalive +
-                ", clientId='" + clientId + '\'' +
-                '}';
+    public void setRobot(RobotConfig robot) {
+        this.robot = robot;
+    }
+
+
+    @Data
+    public static class RobotConfig {
+        private String accessToken;
+        private String secret;
+        private boolean enabled;
     }
 }