重构bd-location模块并优化位置处理逻辑

- 修改MQTT客户端配置，更新IP和端口
- 引入Redis缓存，优化数据存储和查询效率
-重构FenceBreakInEngine，实现围栏闯禁事件的生成和处理
- 优化PointFusionEngine，改进坐标点位抽稀逻辑
- 添加GeoUtils工具类方法，支持地理围栏相关计算- 调整UWBLocationSubscribeListener，实现消息订阅和分发

bd-location/src/main/java/com/ruoyi/bd/domain/BdFenceInfo.java

@@ -4,6 +4,7 @@ import com.ruoyi.common.annotation.Excel;
 import com.ruoyi.common.core.domain.BaseEntity;
 import org.apache.commons.lang3.builder.ToStringBuilder;
 import org.apache.commons.lang3.builder.ToStringStyle;
+import org.locationtech.jts.geom.Polygon;
 
 /**
  * 围栏基础信息对象 bd_fence_info
@@ -33,6 +34,8 @@ public class BdFenceInfo extends BaseEntity
     @Excel(name = "中心点")
     private Double centerLat;
 
+    private Polygon polygon;
+
     public void setId(Long id)
     {
         this.id = id;
@@ -79,6 +82,14 @@ public class BdFenceInfo extends BaseEntity
         return centerLat;
     }
 
+    public Polygon getPolygon() {
+        return polygon;
+    }
+
+    public void setPolygon(Polygon polygon) {
+        this.polygon = polygon;
+    }
+
     @Override
     public String toString() {
         return new ToStringBuilder(this,ToStringStyle.MULTI_LINE_STYLE)

bd-location/src/main/java/com/ruoyi/bd/service/engine/EvtFusionEngine.java

@@ -1,9 +1,9 @@
 package com.ruoyi.bd.service.engine;
 
 import com.alibaba.fastjson2.JSONObject;
+import com.ruoyi.common.core.redis.RedisCache;
 import com.ruoyi.common.utils.DateTimeUtil;
 import com.ruoyi.common.utils.StringUtils;
-import org.apache.commons.lang3.ObjectUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Value;
@@ -13,17 +13,20 @@ import javax.annotation.PostConstruct;
 import javax.annotation.Resource;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.LinkedBlockingQueue;
 
 public abstract class EvtFusionEngine {
     private final static Logger logger = LoggerFactory.getLogger(EvtFusionEngine.class);
-    private final ConcurrentHashMap<String, LocationInfo> evtBucket = new ConcurrentHashMap<>();
 
-    private final List<LinkedBlockingQueue<JSONObject>> messageQueueList = new ArrayList<>(5);
+    @Resource
+    private RedisCache redisCache;
+
 
     private String engineName;
 
+    private static final List<LinkedBlockingQueue<JSONObject>> messageQueueList = new ArrayList<>(5);
+
+
     @Value("${evt-fusion.thread-pool-size:5}")
     private int threadPoolSize;
 
@@ -37,6 +40,7 @@ public abstract class EvtFusionEngine {
         }
     }
 
+
     public void setEngineName(String name) {
         this.engineName = name;
     }
@@ -67,22 +71,35 @@ public abstract class EvtFusionEngine {
                     try {
                         JSONObject msg = messageQueueList.get(finalI).take();
                         LocationInfo locationInfoNew = new LocationInfo(msg.getDouble("latitude"), msg.getDouble("longitude"), msg.getLong("srcTimestamp"));
+                        locationInfoNew.setMsg(msg);
                         String key = getKey(locationInfoNew);
-                        LocationInfo locationInfo = evtBucket.get(key);
-                        locationInfoNew.setEvtTimestamp(DateTimeUtil.timestampMillis());
-                        if (ObjectUtils.isEmpty(locationInfo)) {
+                        if (!redisCache.hasKey(key)) {
                             getBizId(locationInfoNew);
-                            evtBucket.put(key, locationInfoNew);
+                            redisCache.setCacheObject(key, locationInfoNew);
                             continue;
                         }
+                        LocationInfo locationInfo = redisCache.getCacheObject(key);
+                        locationInfoNew.setEvtTimestamp(DateTimeUtil.timestampMillis());
                         // 判断消息是否需要融合
-                        if (!check(locationInfo, locationInfoNew)) {
+                        if (check(locationInfo, locationInfoNew)) {
+                            locationInfo.setSrcTimestamp(locationInfoNew.getSrcTimestamp());
+                            locationInfo.setHandleTimestamp(DateTimeUtil.timestampMillis());
+                            redisCache.setCacheObject(key, locationInfo);
+                        } else {
                             // 老数据释放
                             processOlderData(locationInfo);
+                            getBizId(locationInfoNew);
+                            redisCache.setCacheObject(key, locationInfoNew);
+                        }
+                    } catch (InterruptedException e) {
+                        logger.error("{} error", this.engineName, e);
+                        // 重置中断状态
+                        Thread.currentThread().interrupt();
+                        // 根据业务逻辑决定是否继续执行
+                        if (Thread.currentThread().isInterrupted()) {
+                            logger.error("任务已中断，不再继续执行");
+                            break;
                         }
-                        getBizId(locationInfoNew);
-                        locationInfo.setSrcTimestamp(locationInfoNew.getSrcTimestamp());
-                        evtBucket.put(key, locationInfoNew);
                     } catch (Exception e) {
                         logger.error("{} error", this.engineName, e);
                     }

bd-location/src/main/java/com/ruoyi/bd/service/engine/impl/FenceBreakInEngine.java

@@ -1,12 +1,49 @@
 package com.ruoyi.bd.service.engine.impl;
 
+import com.alibaba.fastjson2.JSONObject;
+import com.ruoyi.bd.domain.BdFenceInfo;
+import com.ruoyi.bd.service.IBdFenceInfoService;
 import com.ruoyi.bd.service.engine.EvtFusionEngine;
 import com.ruoyi.bd.service.engine.LocationInfo;
+import com.ruoyi.common.BDConst;
+import com.ruoyi.common.core.redis.RedisCache;
+import com.ruoyi.common.utils.geo.GeoUtils;
+import net.dreamlu.iot.mqtt.spring.client.MqttClientTemplate;
+import org.apache.commons.lang3.ObjectUtils;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
+import org.springframework.util.CollectionUtils;
+
+import javax.annotation.PostConstruct;
+import javax.annotation.Resource;
+import java.util.ArrayList;
+import java.util.List;
 
 @Service
 public class FenceBreakInEngine extends EvtFusionEngine {
-    // 冲缓存中读取围栏信息
+
+    @Autowired
+    private IBdFenceInfoService fenceInfoService;
+
+    @Resource
+    private RedisCache redisCache;
+
+    @Autowired
+    private MqttClientTemplate client;
+
+    @PostConstruct
+    public void init() {
+        this.setEngineName("围栏闯禁事件融合");
+        if (!redisCache.hasKey(BDConst.REDIS_KEY.FENCE)) {
+            List<BdFenceInfo> bdFenceInfos = fenceInfoService.selectBdFenceInfoList(new BdFenceInfo());
+            if (CollectionUtils.isEmpty(bdFenceInfos)) {
+                bdFenceInfos = new ArrayList<>();
+            }
+            bdFenceInfos.removeIf(item -> ObjectUtils.isEmpty(item.getPoly()) || ObjectUtils.isEmpty(GeoUtils.getPolygon(item.getPoly())));
+            redisCache.setCacheList(BDConst.REDIS_KEY.FENCE, bdFenceInfos);
+        }
+    }
+
     @Override
     public Boolean check(LocationInfo locationInfo, LocationInfo msg) {
         return null;
@@ -26,4 +63,13 @@ public class FenceBreakInEngine extends EvtFusionEngine {
     public void processOlderData(LocationInfo msg) {
 
     }
+
+    public void generateEvt(JSONObject msg, byte[] payload) {
+        List<BdFenceInfo> cacheList = redisCache.getCacheList(BDConst.REDIS_KEY.FENCE);
+        for (BdFenceInfo fenceInfo : cacheList) {
+            if (GeoUtils.isPointInGeoFence(fenceInfo.getPolygon(), msg.getString("longitude"), msg.getString("latitude"))) {
+                client.publish(BDConst.MQTT_TOPIC.EVT_LOCATION_TOPIC, payload);
+            }
+        }
+    }
 }

bd-location/src/main/java/com/ruoyi/bd/service/engine/impl/PointFusionEngine.java

@@ -4,6 +4,7 @@ import com.ruoyi.bd.domain.BdDevcTrailUwb;
 import com.ruoyi.bd.service.IBdDevcTrailUwbService;
 import com.ruoyi.bd.service.engine.EvtFusionEngine;
 import com.ruoyi.bd.service.engine.LocationInfo;
+import com.ruoyi.common.BDConst;
 import com.ruoyi.common.utils.DateTimeUtil;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
@@ -37,6 +38,7 @@ public class PointFusionEngine extends EvtFusionEngine {
      */
     private static final int MAX_POINT_TIME_GAP = 2;
 
+
     @PostConstruct
     public void init() {
         this.setEngineName("坐标点位抽稀");
@@ -47,12 +49,12 @@ public class PointFusionEngine extends EvtFusionEngine {
         long gap = newer.getSrcTimestamp() - older.getSrcTimestamp();
         // 将gap 转换为秒
         gap = gap / 1000;
-        return gap >= MAX_POINT_TIME_GAP;
+        return gap <= MAX_POINT_TIME_GAP;
     }
 
     @Override
     public String getKey(LocationInfo msg) {
-        return msg.getMsg().getString("deviceId");
+        return BDConst.REDIS_KEY.BD_POINT_KEY + msg.getMsg().getString("deviceId");
     }
 
     @Override

bd-location/src/main/java/com/ruoyi/common/BDConst.java

@@ -0,0 +1,16 @@
+package com.ruoyi.common;
+
+public interface BDConst {
+    public interface REDIS_KEY {
+        String BD_POINT_KEY = "bd:point:";
+        String FENCE_BREAK_IN_KEY = "fence:break:in:";
+        String FENCE = "fence";
+    }
+
+    public interface MQTT_TOPIC {
+        String POINT_LOCATION_TOPIC = "/uwb/point/#";
+        String EVT_LOCATION_TOPIC = "/uwb/evt/#";
+
+
+    }
+}

bd-location/src/main/java/com/ruoyi/mqtt/UWBLocationSubscribeListener.java

@@ -1,34 +1,48 @@
 package com.ruoyi.mqtt;
 
+import com.alibaba.fastjson2.JSON;
+import com.alibaba.fastjson2.JSONObject;
+import com.ruoyi.bd.service.engine.impl.FenceBreakInEngine;
+import com.ruoyi.bd.service.engine.impl.PointFusionEngine;
+import com.ruoyi.common.BDConst;
 import net.dreamlu.iot.mqtt.codec.MqttQoS;
 import net.dreamlu.iot.mqtt.spring.client.MqttClientSubscribe;
-import net.dreamlu.iot.mqtt.spring.client.MqttClientTemplate;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
+import javax.annotation.PostConstruct;
 import java.nio.charset.StandardCharsets;
 
 @Service
 public class UWBLocationSubscribeListener {
     private static final Logger logger = LoggerFactory.getLogger(UWBLocationSubscribeListener.class);
+
+    @Autowired
+    private PointFusionEngine pointFusionEngine;
+
     @Autowired
-    private MqttClientTemplate client;
+    private FenceBreakInEngine fenceBreakInEngine;
 
-    public boolean publish() {
-        client.publish("/test/client", "mica最牛皮".getBytes(StandardCharsets.UTF_8));
-        return true;
+    @PostConstruct
+    public void init() {
+        pointFusionEngine.start();
+        fenceBreakInEngine.start();
     }
 
-    @MqttClientSubscribe("/uwb/2/#")
+    @MqttClientSubscribe(value = BDConst.MQTT_TOPIC.POINT_LOCATION_TOPIC)
     public void subQos0(String topic, byte[] payload) {
         logger.info("topic:{} payload:{}", topic, new String(payload, StandardCharsets.UTF_8));
+        JSONObject jsonObject = JSON.parseObject(payload);
+        pointFusionEngine.push(jsonObject);
+        fenceBreakInEngine.generateEvt(JSON.parseObject(payload), payload);
     }
 
-    @MqttClientSubscribe(value = "/uwb/1/#", qos = MqttQoS.QOS0)
+    @MqttClientSubscribe(value = BDConst.MQTT_TOPIC.EVT_LOCATION_TOPIC, qos = MqttQoS.QOS0)
     public void subQos1(String topic, byte[] payload) {
         logger.info("topic:{} payload:{}", topic, new String(payload, StandardCharsets.UTF_8));
+        fenceBreakInEngine.push(JSON.parseObject(payload));
     }
 
     @MqttClientSubscribe("/sys/${productKey}/${deviceName}/thing/sub/register")

bd-location/src/main/resources/application-druid.yml

@@ -84,8 +84,8 @@ spring:
 mqtt:
   client:
     enabled: true
-    ip: xt.wenhq.top
-    port: 8581
+    ip: 200.200.19.121
+    port: 31005
     name: uwb-location-client
     client-id: uwb-000001
     global-subscribe:

ruoyi-common/src/main/java/com/ruoyi/common/utils/geo/GeoUtils.java

@@ -58,6 +58,28 @@ public class GeoUtils {
     }
 
     /**
+     * getPolygon
+     * POLYGON((0 0, 0 5, 5 5, 5 0, 0 0))
+     *
+     * @param wktPolygon the wkt polygon
+     * @return the poly center
+     * @author chen.cheng
+     */
+    public static Polygon getPolygon(String wktPolygon) {
+        // 创建WKTReader对象
+        WKTReader reader = new WKTReader(geometryFactory);
+        // 从WKT字符串中读取几何对象
+        Polygon geometry = null;
+        try {
+            geometry = (Polygon) reader.read(wktPolygon);
+        } catch (ParseException e) {
+            log.info("Invalid WKT string: " + e.getMessage());
+        }
+        // 获取中心点
+        return geometry;
+    }
+
+    /**
      * 判断指定的GPS点是否在电子围栏内
      *
      * @param fencePointsList 包含电子围栏的经纬度数据的列表，格式为 "经度,纬度"
@@ -100,6 +122,17 @@ public class GeoUtils {
         return geoFencePolygon.contains(point);
     }
 
+    public static boolean isPointInGeoFence(Polygon geoFencePolygon, String lng, String lat) {
+        // 将指定的GPS点转换为坐标
+        Coordinate testPoint = parseCoordinate(lng, lat);
+
+        // 创建指定的GPS点
+        Point point = geometryFactory.createPoint(testPoint);
+
+        // 检查指定的GPS点是否在电子围栏内
+        return geoFencePolygon.contains(point);
+    }
+
     /**
      * 判断指定的GPS点是否在电子围栏内
      *
@@ -157,4 +190,8 @@ public class GeoUtils {
         double lat = Double.parseDouble(parts[1]);
         return new Coordinate(lon, lat);
     }
+
+    private static Coordinate parseCoordinate(String lng, String lat) {
+        return new Coordinate(Double.parseDouble(lng), Double.parseDouble(lat));
+    }
 }