重构事件融合引擎并添加新事件回调功能

- 将 EvtFusionEngine 抽象类改为实现 IFusionEngine 接口- 移除 EvtFusionEngine 中的抽象方法，改为默认实现
- 在 EvtFusionEngine 中添加 newEvtCallback 方法用于处理新事件回调
- 更新 FenceBreakInEngine 和 PointFusionEngine 实现类
- 新增 IFusionEngine 接口定义基本的融合引擎行为

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

@@ -9,13 +9,12 @@ import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 
-import javax.annotation.PostConstruct;
 import javax.annotation.Resource;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.LinkedBlockingQueue;
 
-public abstract class EvtFusionEngine {
+public abstract class EvtFusionEngine implements IFusionEngine {
     private final static Logger logger = LoggerFactory.getLogger(EvtFusionEngine.class);
 
     @Resource
@@ -33,7 +32,6 @@ public abstract class EvtFusionEngine {
     @Resource(name = "threadPoolTaskExecutor")
     private ThreadPoolTaskExecutor threadPoolTaskExecutor;
 
-    @PostConstruct
     public void init() {
         for (int i = 0; i < threadPoolSize; i++) {
             messageQueueList.add(new LinkedBlockingQueue<>());
@@ -55,14 +53,6 @@ public abstract class EvtFusionEngine {
         messageQueue.offer(msg);
     }
 
-    public abstract Boolean check(LocationInfo older, LocationInfo newer);
-
-    public abstract String getKey(LocationInfo msg);
-
-    public abstract void getBizId(LocationInfo msg);
-
-    public abstract void processOlderData(LocationInfo msg);
-
     public void start() {
         for (int i = 0; i < threadPoolSize; i++) {
             int finalI = i;
@@ -76,6 +66,7 @@ public abstract class EvtFusionEngine {
                         if (!redisCache.hasKey(key)) {
                             getBizId(locationInfoNew);
                             redisCache.setCacheObject(key, locationInfoNew);
+                            newEvtCallback(locationInfoNew);
                             continue;
                         }
                         LocationInfo locationInfo = redisCache.getCacheObject(key);
@@ -90,6 +81,7 @@ public abstract class EvtFusionEngine {
                             processOlderData(locationInfo);
                             getBizId(locationInfoNew);
                             redisCache.setCacheObject(key, locationInfoNew);
+                            newEvtCallback(locationInfoNew);
                         }
                     } catch (InterruptedException e) {
                         logger.error("{} error", this.engineName, e);

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

@@ -0,0 +1,20 @@
+package com.ruoyi.bd.service.engine;
+
+public interface IFusionEngine {
+    default Boolean check(LocationInfo older, LocationInfo newer) {
+        return false;
+    }
+
+    default String getKey(LocationInfo msg) {
+        return "not implement";
+    }
+
+    default void getBizId(LocationInfo msg) {
+    }
+
+    default void processOlderData(LocationInfo msg) {
+    }
+
+    default void newEvtCallback(LocationInfo msg) {
+    }
+}

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

@@ -101,6 +101,10 @@ public class FenceBreakInEngine extends EvtFusionEngine {
         });
     }
 
+    @Override
+    public void newEvtCallback(LocationInfo msg) {
+    }
+
     public void generateEvt(JSONObject msg, byte[] payload) {
         List<BdFenceInfo> cacheList = redisCache.getCacheList(BDConst.REDIS_KEY.FENCE);
         Polygon polygon;

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

@@ -71,16 +71,4 @@ public class PointFusionEngine extends EvtFusionEngine {
         bdDevcTrailUwbService.insertBdDevcTrailUwb(bdDevcTrailUwb);
         msg.setBizId(bdDevcTrailUwb.getId().toString());
     }
-
-    /**
-     * Process older data.
-     * 可以做事件消散等事务
-     *
-     * @param msg the msg
-     * @author chen.cheng
-     */
-    @Override
-    public void processOlderData(LocationInfo msg) {
-
-    }
 }