1.数据固化成功后给业务侧发送通知。

amos-boot-data/amos-boot-data-equip/src/main/java/com/yeejoin/equip/service/HandleESMessage2TDService.java

 package com.yeejoin.equip.service;
 
+import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson.JSONObject;
 import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
 import com.google.common.collect.Lists;
@@ -11,8 +12,12 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Component;
 import org.springframework.transaction.annotation.Transactional;
+import org.typroject.tyboot.component.emq.EmqKeeper;
+
 import java.text.SimpleDateFormat;
+import java.util.HashMap;
 import java.util.List;
+
 /**
  * @author LiuLin
  * @date 2023/6/25
@@ -23,11 +28,13 @@ import java.util.List;
 public class HandleESMessage2TDService {
     public static final String ES_INDEX = "jxiop_equipments";
     public static final int SIZE = 1000;
+    private static final String datePattern = "yyyy-MM-dd HH:mm:ss.SSS";
     @Autowired
     private ESEquipmentsMapper esEquipmentsMapper;
     @Autowired
     private ElasticSearchUtil elasticSearchUtil;
-    private static final String datePattern = "yyyy-MM-dd HH:mm:ss.SSS";
+    @Autowired
+    private EmqKeeper emqKeeper;
 
     /**
      * 十分钟拉取ES数据存入TdEngine
@@ -41,9 +48,19 @@ public class HandleESMessage2TDService {
         String time = format.format(System.currentTimeMillis());
         for (List<ESEquipments> tempDataList : allDataList) {
             if (CollectionUtils.isNotEmpty(tempDataList)) {
-                esEquipmentsMapper.batchInsert(tempDataList,time);
+                esEquipmentsMapper.batchInsert(tempDataList, time);
             }
         }
-        log.info("同步ES数据至TDEngine成功!共同步{}条!",result.size());
+        log.info("同步ES数据至TDEngine成功!共同步{}条!", result.size());
+
+        try {
+            HashMap<String, String> syncFlag = new HashMap<>();
+            syncFlag.put("sync_flag", "success");
+            emqKeeper.getMqttClient().publish("sync_esdata_to_tdengine_notice", JSON.toJSONString(syncFlag).getBytes(), 0, false);
+            log.info("同步ES数据至发发送消息给业务发送通知成功！");
+        } catch (Exception exception) {
+            log.info("同步ES数据至发发送消息给业务发送通知失败！");
+        }
     }
+
 }