【代码评审】IoT：数据桥梁的执行器抽离

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/action/IotRuleSceneDataBridgeAction.java

@@ -7,7 +7,7 @@ import cn.iocoder.yudao.module.iot.dal.dataobject.rule.IotRuleSceneDO;
 import cn.iocoder.yudao.module.iot.enums.rule.IotRuleSceneActionTypeEnum;
 import cn.iocoder.yudao.module.iot.mq.message.IotDeviceMessage;
 import cn.iocoder.yudao.module.iot.service.rule.IotDataBridgeService;
-import cn.iocoder.yudao.module.iot.service.rule.execute.IotDataBridgeExecute;
+import cn.iocoder.yudao.module.iot.service.rule.action.databridge.IotDataBridgeExecute;
 import jakarta.annotation.Resource;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Component;
@@ -19,9 +19,6 @@ import java.util.List;
  *
  * @author 芋道源码
  */
-// TODO @芋艿：【优化】因为 bridge 会比较多，所以可以考虑在 rule 下，新建一个 bridge 的 package，然后定义一个 bridgehandler，它有：
-//    1. input 方法、output 方法
-//    2. build 方法，用于有状态的连接，例如说 mq、tcp、websocket
 @Component
 @Slf4j
 public class IotRuleSceneDataBridgeAction implements IotRuleSceneAction {
@@ -49,7 +46,7 @@ public class IotRuleSceneDataBridgeAction implements IotRuleSceneAction {
             return;
         }
 
-        // 2.1 执行数据桥接操作
+        // 2. 执行数据桥接操作
         dataBridgeExecutes.forEach(execute -> execute.execute(message, dataBridge));
     }
 

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/action/databridge/IotDataBridgeExecute.java

@@ -1,4 +1,4 @@
-package cn.iocoder.yudao.module.iot.service.rule.execute;
+package cn.iocoder.yudao.module.iot.service.rule.action.databridge;
 
 import cn.iocoder.yudao.module.iot.dal.dataobject.rule.IotDataBridgeDO;
 import cn.iocoder.yudao.module.iot.mq.message.IotDeviceMessage;

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/action/databridge/IotHttpDataBridgeExecute.java

@@ -1,4 +1,4 @@
-package cn.iocoder.yudao.module.iot.service.rule.execute;
+package cn.iocoder.yudao.module.iot.service.rule.action.databridge;
 
 import cn.hutool.core.collection.CollUtil;
 import cn.iocoder.yudao.framework.common.util.http.HttpUtils;

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/action/databridge/IotRocketMQDataBridgeExecute.java

@@ -1,4 +1,4 @@
-package cn.iocoder.yudao.module.iot.service.rule.execute;
+package cn.iocoder.yudao.module.iot.service.rule.action.databridge;
 
 import cn.iocoder.yudao.module.iot.dal.dataobject.rule.IotDataBridgeDO;
 import cn.iocoder.yudao.module.iot.enums.rule.IotDataBridgTypeEnum;
@@ -30,12 +30,13 @@ public class IotRocketMQDataBridgeExecute implements IotDataBridgeExecute {
     /**
      * 针对 {@link IotDataBridgeDO.RocketMQConfig} 的 DefaultMQProducer 缓存
      */
+    // TODO @puhui999：因为 kafka 之类也存在这个情况，是不是得搞个抽象类。提供一个 initProducer，和 closeProducer 方法
     private final LoadingCache<IotDataBridgeDO.RocketMQConfig, DefaultMQProducer> PRODUCER_CACHE = CacheBuilder.newBuilder()
-            // 只阻塞当前数据加载线程，其他线程返回旧值
+            .refreshAfterWrite(Duration.ofMinutes(10)) // TODO puhui999：应该是 read 30 分钟哈
-            .refreshAfterWrite(Duration.ofMinutes(10))
             // 增加移除监听器，自动关闭 producer
             .removalListener(notification -> {
                 DefaultMQProducer producer = (DefaultMQProducer) notification.getValue();
+                // TODO puhui999：if return，更简短哈
                 if (producer != null) {
                     try {
                         producer.shutdown();
@@ -45,8 +46,10 @@ public class IotRocketMQDataBridgeExecute implements IotDataBridgeExecute {
                     }
                 }
             })
+            // TODO @puhui999：就同步哈，不用异步处理。
             // 通过 asyncReloading 实现全异步加载，包括 refreshAfterWrite 被阻塞的加载线程
             .build(CacheLoader.asyncReloading(new CacheLoader<IotDataBridgeDO.RocketMQConfig, DefaultMQProducer>() {
+
                 @Override
                 public DefaultMQProducer load(IotDataBridgeDO.RocketMQConfig config) throws Exception {
                     DefaultMQProducer producer = new DefaultMQProducer(config.getGroup());
@@ -55,6 +58,7 @@ public class IotRocketMQDataBridgeExecute implements IotDataBridgeExecute {
                     log.info("[PRODUCER_CACHE][配置({}) 对应的 producer 已创建并启动]", config);
                     return producer;
                 }
+
             }, Executors.newCachedThreadPool()));
 
     @Override
@@ -120,7 +124,7 @@ public class IotRocketMQDataBridgeExecute implements IotDataBridgeExecute {
         log.info("[main][第一次执行，应该会创建新的 producer]");
         action.executeRocketMQ(message, config);
 
-        log.info("[main][第二次执行，应该会复用缓存的 producer]"); 
+        log.info("[main][第二次执行，应该会复用缓存的 producer]");
         action.executeRocketMQ(message, config);
     }