mq 上行

depot-intelligent/pom.xml

@@ -16,22 +16,24 @@
     <parent>
         <groupId>org.springframework.boot</groupId>
         <artifactId>spring-boot-starter-parent</artifactId>
-        <version>2.0.2.RELEASE</version>
+        <version>2.3.12.RELEASE</version>
         <relativePath/> <!-- lookup parent from repository -->
     </parent>
-    <distributionManagement>
+    <!--<distributionManagement>
         <repository>
             <id>nexus</id>
-            <!--名称自己定义-->
+            &lt;!&ndash;名称自己定义&ndash;&gt;
             <name>release</name>
             <url>http://10.10.1.25:8081/repository/host_repository/</url>
         </repository>
-    </distributionManagement>
+    </distributionManagement>-->
     <properties>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
         <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
+        <skipTests>true</skipTests>
         <java.version>1.8</java.version>
-        <spring-cloud.version>Finchley.RELEASE</spring-cloud.version>
+        <spring-cloud.version>Hoxton.SR12</spring-cloud.version>
+        <rocketmq-starter.version>2.2.1</rocketmq-starter.version>
     </properties>
 
     <dependencies>
@@ -141,12 +143,6 @@
             <artifactId>quartz</artifactId>
             <version>2.3.0</version>
         </dependency>
-
-        <dependency>
-            <groupId>org.springframework.boot</groupId>
-            <artifactId>spring-boot-starter-test</artifactId>
-        </dependency>
-
         <dependency>
             <groupId>org.smslib</groupId>
             <artifactId>smslib</artifactId>
@@ -176,6 +172,12 @@
             <groupId>org.springframework.session</groupId>
             <artifactId>spring-session-data-redis</artifactId>
         </dependency>
+        <!-- rocketmq -->
+        <dependency>
+            <groupId>org.apache.rocketmq</groupId>
+            <artifactId>rocketmq-spring-boot-starter</artifactId>
+            <version>${rocketmq-starter.version}</version>
+        </dependency>
     </dependencies>
 
     <dependencyManagement>

depot-intelligent/src/main/java/com/chinaitop/depot/intelligent/pushs/service/impl/GrainPushsServiceImpl.java

@@ -12,6 +12,8 @@ import com.chinaitop.depot.intelligent.grainsituation.model.TWarningThresholdHis
 import com.chinaitop.depot.intelligent.grainsituation.service.TWarningThresholdHistoryService;
 import com.chinaitop.depot.intelligent.grainsituation.service.TemperatureRecordService;
 import com.chinaitop.depot.intelligent.pushs.service.GrainPushsService;
+import com.chinaitop.depot.intelligent.rocketMQ.producerMQ.model.MqMsg;
+import com.chinaitop.depot.intelligent.rocketMQ.producerMQ.service.RocketMqService;
 import com.chinaitop.depot.intelligent.utils.JsonToObjectUtils;
 import com.chinaitop.depot.intelligent.utils.UuidUtils;
 import com.chinaitop.depot.intelligent.utils.VerificationSourceUtils;
@@ -38,27 +40,22 @@ public class GrainPushsServiceImpl implements GrainPushsService {
 
     @Resource
     private TemperatureRecordService temperatureRecordService;
-
     @Resource
     private TTestdataLayerMapper tTestdataLayerMapper;
-
     @Resource
     private WarningThresholdService warningThresholdService;
-
     @Resource
     private TWarningThresholdHistoryMapper tWarningThresholdHistoryMapper;
-
     @Resource
     private UuidUtils uuidUtils;
-
     @Resource
     private WarningMessage warningMessage;
-
     @Resource
     private ResolutionToSendAlarmService resolutionToSendAlarmService;
-
     @Resource
     private TWarningThresholdHistoryService tWarningThresholdHistoryService;
+    @Resource
+    private RocketMqService rocketMqService;
 
     @Transactional(rollbackFor = Exception.class)
     @Override
@@ -282,18 +279,23 @@ public class GrainPushsServiceImpl implements GrainPushsService {
 
                 if (tTestdataList.size() > 0) {
                     temperatureRecordService.insertBatch(tTestdataList);
+                    // 消息 CRUD 增加(Create)、* 不需要检索(Retrieve)、更新(Update)和删除(Delete) 小写区分增删改查
+                    rocketMqService.asyncSend(new MqMsg("smart_grp_yn", "t_testdata_c", null, tTestdataList));
                 }
                 if (tTestdataLayerArrayList.size() > 0) {
                     tTestdataLayerMapper.insertBatch(tTestdataLayerArrayList);
+                    rocketMqService.asyncSend(new MqMsg("smart_grp_yn", "t_testdata_layer_c", null, tTestdataLayerArrayList));
                 }
                 if (tWarningThresholdHistorieList.size() > 0) {
                     tWarningThresholdHistoryMapper.insertBatch(tWarningThresholdHistorieList);
+                    rocketMqService.asyncSend(new MqMsg("smart_grp_yn", "t_warning_threshold_history_c", null, tWarningThresholdHistorieList));
                     List<TWarningThresholdHistory> collect = tWarningThresholdHistorieList.stream().filter(TWarningThresholdHistory -> TWarningThresholdHistory.getwWarning().equals("0")).collect(Collectors.toList());
                     boolean alarmDetails = resolutionToSendAlarmService.getAlarmDetails(collect);
                     if (!alarmDetails) {
                         collect.forEach(obj -> {
                             obj.setIsSend("2");
                             tWarningThresholdHistoryService.updateById(obj);
+                            rocketMqService.asyncSend(new MqMsg("smart_grp_yn", "t_warning_threshold_history_u", null, tWarningThresholdHistorieList));
                         });
                     }
                 }

depot-intelligent/src/main/java/com/chinaitop/depot/intelligent/rocketMQ/producerMQ/model/MqMsg.java

@@ -0,0 +1,42 @@
+package com.chinaitop.depot.intelligent.rocketMQ.producerMQ.model;
+
+import lombok.Data;
+import javax.validation.constraints.NotNull;
+import java.io.Serializable;
+
+/**
+ * @author qingsong.han
+ * @description:
+ * @create 2022-03-23 13:59
+ */
+@Data
+public class MqMsg implements Serializable {
+    private static final long serialVersionUID = -2023812638050201776L;
+
+    public MqMsg(@NotNull String topic, String tags, String transactionName, Object content) {
+        this.topic = topic;
+        this.tags = tags;
+        this.transactionName = transactionName;
+        this.content = content;
+    }
+
+    /**
+     * 一级消息:消息topic(topic为消息的主题)
+     */
+    @NotNull
+
+    private String topic;
+    /**
+     * 二级消息:消息topic对应的tags
+     */
+    private String tags;
+    /**
+     * 事务名称 为空默认 rocketmq_transaction_default_global_name
+     */
+    private String transactionName;
+    /**
+     * 消息内容
+     */
+    private Object content;
+
+}

depot-intelligent/src/main/java/com/chinaitop/depot/intelligent/rocketMQ/producerMQ/service/RocketMqService.java

@@ -0,0 +1,21 @@
+package com.chinaitop.depot.intelligent.rocketMQ.producerMQ.service;
+
+import com.chinaitop.depot.intelligent.rocketMQ.producerMQ.model.MqMsg;
+
+/**
+ * @author qingsong.han
+ * @description: send Rocket MQ
+ * @create 2022-03-23 13:45
+ */
+public interface RocketMqService {
+
+    /**
+     * 异步发送消息，异步返回消息结果 smart_grp_sx
+     *
+     * 当发送的消息很重要，且对响应时间非常敏感的时候采用async方式
+     *
+     * @param mqMsg 发送消息实体类
+     */
+    void asyncSend(MqMsg mqMsg);
+
+}

depot-intelligent/src/main/java/com/chinaitop/depot/intelligent/rocketMQ/producerMQ/service/impl/RocketMqServiceImpl.java

@@ -0,0 +1,53 @@
+package com.chinaitop.depot.intelligent.rocketMQ.producerMQ.service.impl;
+
+import com.chinaitop.depot.intelligent.rocketMQ.producerMQ.model.MqMsg;
+import com.chinaitop.depot.intelligent.rocketMQ.producerMQ.service.RocketMqService;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.rocketmq.client.producer.SendCallback;
+import org.apache.rocketmq.client.producer.SendResult;
+import org.apache.rocketmq.spring.core.RocketMQTemplate;
+import org.springframework.stereotype.Service;
+import org.springframework.util.StringUtils;
+
+import javax.annotation.Resource;
+
+/**
+ * @author qingsong.han
+ * @description:
+ * @create 2022-03-23 14:09
+ */
+@Slf4j
+@Service
+public class RocketMqServiceImpl implements RocketMqService {
+    @Resource
+    private RocketMQTemplate rocketMQTemplate;
+
+    /**
+     * topic: smart_grp_yn
+     * tags:
+     *  测温: yn_cw
+     *
+     * @param mqMsg 发送消息实体类
+     */
+    @Override
+    public void asyncSend(MqMsg mqMsg) {
+        log.info("asyncSend发送消息到mqMsg: {}", mqMsg);
+        rocketMQTemplate.asyncSend(
+                StringUtils.isEmpty(mqMsg.getTags())
+                        ? mqMsg.getTopic()
+                        : String.format("%s:%s", mqMsg.getTopic(), mqMsg.getTags()),
+                mqMsg.getContent(),
+                new SendCallback() {
+                    @Override
+                    public void onSuccess(SendResult sendResult) {
+                        // 成功不做日志记录或处理
+                    }
+
+                    @Override
+                    public void onException(Throwable throwable) {
+                        log.info("mqMsg: {}消息发送失败", mqMsg);
+                    }
+                }
+        );
+    }
+}

depot-intelligent/src/main/resources/bootstrap.yml

@@ -1,5 +1,5 @@
 server:
-  port: 9028
+  port: 9029
   tomcat:
     uri-encoding: utf-8
 
@@ -20,7 +20,7 @@ spring:
   # 数据库配置
   datasource:
     driver-class-name: com.mysql.jdbc.Driver
-    url: jdbc:mysql://localhost:3306/depot?useUnicode=true&characterEncoding=utf-8&useSSL=false&serverTimezone=UTC
+    url: jdbc:mysql://localhost:3306/depot_yunnan?useUnicode=true&characterEncoding=utf-8&useSSL=false&serverTimezone=UTC
     username: root
     password: 123456
   # 缓存配置
@@ -65,3 +65,15 @@ gas-org-ids: 386,177,273,280,300,298,294
 # 数量检测版本区分version3
 num_version3: :20013
 
+rocketmq:
+  # master
+  name-server: 172.16.0.8:9876,172.16.0.10:9876
+  producer:
+    group: depot_sxp_group
+    send-message-timeout: 2000
+    retry-times-when-send-async-failed: 5
+    retry-times-when-send-failed: 5
+# sql 打印
+#logging:
+#  level:
+#   com.chinaitop.depot.intelligent.grainsituation.mapper: debug