From f76843573ee25546b829b5a1209bba9b2ffcb12f Mon Sep 17 00:00:00 2001 From: YunaiV Date: Sat, 22 Feb 2025 18:31:09 +0800 Subject: [PATCH] 【代码评审】IoT：数据桥梁的执行器抽离 --- yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/action/IotRuleSceneDataBridgeAction.java | 7 ++----- yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/action/databridge/IotDataBridgeExecute.java | 32 ++++++++++++++++++++++++++++++++ yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/action/databridge/IotHttpDataBridgeExecute.java | 93 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/action/databridge/IotRocketMQDataBridgeExecute.java | 131 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/execute/IotDataBridgeExecute.java | 32 -------------------------------- yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/execute/IotHttpDataBridgeExecute.java | 93 --------------------------------------------------------------------------------------------- yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/execute/IotRocketMQDataBridgeExecute.java | 127 ------------------------------------------------------------------------------------------------------------------------------- 7 files changed, 258 insertions(+), 257 deletions(-) create mode 100644 yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/action/databridge/IotDataBridgeExecute.java create mode 100644 yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/action/databridge/IotHttpDataBridgeExecute.java create mode 100644 yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/action/databridge/IotRocketMQDataBridgeExecute.java delete mode 100644 yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/execute/IotDataBridgeExecute.java delete mode 100644 yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/execute/IotHttpDataBridgeExecute.java delete mode 100644 yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/execute/IotRocketMQDataBridgeExecute.java diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/action/IotRuleSceneDataBridgeAction.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/action/IotRuleSceneDataBridgeAction.java index f87cd60..6733331 100644 --- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/action/IotRuleSceneDataBridgeAction.java +++ b/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)); } diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/action/databridge/IotDataBridgeExecute.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/action/databridge/IotDataBridgeExecute.java new file mode 100644 index 0000000..cd00f4f --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/action/databridge/IotDataBridgeExecute.java @@ -0,0 +1,32 @@ +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; + + +/** + * IoT 数据桥梁的执行器 execute 接口 + * + * @author HUIHUI + */ +public interface IotDataBridgeExecute { + + /** + * 执行数据桥接操作 + * + * @param message 设备消息 + * @param dataBridge 数据桥梁 + */ + void execute(IotDeviceMessage message, IotDataBridgeDO dataBridge); + + // TODO @芋艿：因为下面的，都是有状态的，所以通过 guava 缓存连接，然后通过 RemovalNotification 实现关闭。例如说，一次新建有效期是 10 分钟； + // TODO @芋艿：mq-redis + // TODO @芋艿：mq-数据库 + // TODO @芋艿：kafka + // TODO @芋艿：rocketmq + // TODO @芋艿：rabbitmq + // TODO @芋艿：mqtt + // TODO @芋艿：tcp + // TODO @芋艿：websocket + +} diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/action/databridge/IotHttpDataBridgeExecute.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/action/databridge/IotHttpDataBridgeExecute.java new file mode 100644 index 0000000..76f1b79 --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/action/databridge/IotHttpDataBridgeExecute.java @@ -0,0 +1,93 @@ +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; +import cn.iocoder.yudao.framework.common.util.json.JsonUtils; +import cn.iocoder.yudao.module.iot.dal.dataobject.rule.IotDataBridgeDO; +import cn.iocoder.yudao.module.iot.enums.rule.IotDataBridgTypeEnum; +import cn.iocoder.yudao.module.iot.mq.message.IotDeviceMessage; +import jakarta.annotation.Resource; +import lombok.extern.slf4j.Slf4j; +import org.springframework.http.*; +import org.springframework.stereotype.Component; +import org.springframework.web.client.RestTemplate; +import org.springframework.web.util.UriComponentsBuilder; + +import java.util.HashMap; +import java.util.Map; + +import static cn.iocoder.yudao.framework.web.core.util.WebFrameworkUtils.HEADER_TENANT_ID; + +/** + * Http 的 {@link IotDataBridgeExecute} 实现类 + * + * @author HUIHUI + */ +@Component +@Slf4j +public class IotHttpDataBridgeExecute implements IotDataBridgeExecute { + + @Resource + private RestTemplate restTemplate; + + @Override + public void execute(IotDeviceMessage message, IotDataBridgeDO dataBridge) { + // 1.1 校验数据桥接的类型 == HTTP + if (!IotDataBridgTypeEnum.HTTP.getType().equals(dataBridge.getType())) { + return; + } + // 1.2 执行 HTTP 请求 + executeHttp(message, (IotDataBridgeDO.HttpConfig) dataBridge.getConfig()); + } + + @SuppressWarnings({"unchecked", "deprecation"}) + private void executeHttp(IotDeviceMessage message, IotDataBridgeDO.HttpConfig config) { + String url = null; + HttpMethod method = HttpMethod.valueOf(config.getMethod().toUpperCase()); + HttpEntity requestEntity = null; + ResponseEntity responseEntity = null; + try { + // 1.1 构建 Header + HttpHeaders headers = new HttpHeaders(); + if (CollUtil.isNotEmpty(config.getHeaders())) { + config.getHeaders().putAll(config.getHeaders()); + } + headers.add(HEADER_TENANT_ID, message.getTenantId().toString()); + // 1.2 构建 URL + UriComponentsBuilder uriBuilder = UriComponentsBuilder.fromUriString(config.getUrl()); + if (CollUtil.isNotEmpty(config.getQuery())) { + config.getQuery().forEach(uriBuilder::queryParam); + } + // 1.3 构建请求体 + if (method == HttpMethod.GET) { + uriBuilder.queryParam("message", HttpUtils.encodeUtf8(JsonUtils.toJsonString(message))); + url = uriBuilder.build().toUriString(); + requestEntity = new HttpEntity<>(headers); + } else { + url = uriBuilder.build().toUriString(); + Map requestBody = JsonUtils.parseObject(config.getBody(), Map.class); + if (requestBody == null) { + requestBody = new HashMap<>(); + } + requestBody.put("message", message); + headers.add(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_UTF8_VALUE); + requestEntity = new HttpEntity<>(JsonUtils.toJsonString(requestBody), headers); + } + + // 2.1 发送请求 + responseEntity = restTemplate.exchange(url, method, requestEntity, String.class); + // 2.2 记录日志 + if (responseEntity.getStatusCode().is2xxSuccessful()) { + log.info("[executeHttp][message({}) config({}) url({}) method({}) requestEntity({}) 请求成功({})]", + message, config, url, method, requestEntity, responseEntity); + } else { + log.error("[executeHttp][message({}) config({}) url({}) method({}) requestEntity({}) 请求失败({})]", + message, config, url, method, requestEntity, responseEntity); + } + } catch (Exception e) { + log.error("[executeHttp][message({}) config({}) url({}) method({}) requestEntity({}) 请求异常({})]", + message, config, url, method, requestEntity, responseEntity, e); + } + } + +} \ No newline at end of file diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/action/databridge/IotRocketMQDataBridgeExecute.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/action/databridge/IotRocketMQDataBridgeExecute.java new file mode 100644 index 0000000..626208e --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/action/databridge/IotRocketMQDataBridgeExecute.java @@ -0,0 +1,131 @@ +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; +import cn.iocoder.yudao.module.iot.mq.message.IotDeviceMessage; +import com.google.common.cache.CacheBuilder; +import com.google.common.cache.CacheLoader; +import com.google.common.cache.LoadingCache; +import lombok.extern.slf4j.Slf4j; +import org.apache.rocketmq.client.producer.DefaultMQProducer; +import org.apache.rocketmq.client.producer.SendResult; +import org.apache.rocketmq.client.producer.SendStatus; +import org.apache.rocketmq.common.message.Message; +import org.apache.rocketmq.remoting.common.RemotingHelper; +import org.springframework.stereotype.Component; + +import java.time.Duration; +import java.time.LocalDateTime; +import java.util.concurrent.Executors; + +/** + * RocketMQ 的 {@link IotDataBridgeExecute} 实现类 + * + * @author HUIHUI + */ +@Component +@Slf4j +public class IotRocketMQDataBridgeExecute implements IotDataBridgeExecute { + + /** + * 针对 {@link IotDataBridgeDO.RocketMQConfig} 的 DefaultMQProducer 缓存 + */ + // TODO @puhui999：因为 kafka 之类也存在这个情况，是不是得搞个抽象类。提供一个 initProducer，和 closeProducer 方法 + private final LoadingCache PRODUCER_CACHE = CacheBuilder.newBuilder() + .refreshAfterWrite(Duration.ofMinutes(10)) // TODO puhui999：应该是 read 30 分钟哈 + // 增加移除监听器，自动关闭 producer + .removalListener(notification -> { + DefaultMQProducer producer = (DefaultMQProducer) notification.getValue(); + // TODO puhui999：if return，更简短哈 + if (producer != null) { + try { + producer.shutdown(); + log.info("[PRODUCER_CACHE][配置({}) 对应的 producer 已关闭]", notification.getKey()); + } catch (Exception e) { + log.error("[PRODUCER_CACHE][配置({}) 对应的 producer 关闭失败]", notification.getKey(), e); + } + } + }) + // TODO @puhui999：就同步哈，不用异步处理。 + // 通过 asyncReloading 实现全异步加载，包括 refreshAfterWrite 被阻塞的加载线程 + .build(CacheLoader.asyncReloading(new CacheLoader() { + + @Override + public DefaultMQProducer load(IotDataBridgeDO.RocketMQConfig config) throws Exception { + DefaultMQProducer producer = new DefaultMQProducer(config.getGroup()); + producer.setNamesrvAddr(config.getNameServer()); + producer.start(); + log.info("[PRODUCER_CACHE][配置({}) 对应的 producer 已创建并启动]", config); + return producer; + } + + }, Executors.newCachedThreadPool())); + + @Override + public void execute(IotDeviceMessage message, IotDataBridgeDO dataBridge) { + // 1.1 校验数据桥接的类型 == ROCKETMQ + if (!IotDataBridgTypeEnum.ROCKETMQ.getType().equals(dataBridge.getType())) { + return; + } + // 1.2 执行 RocketMQ 发送消息 + executeRocketMQ(message, (IotDataBridgeDO.RocketMQConfig) dataBridge.getConfig()); + } + + private void executeRocketMQ(IotDeviceMessage message, IotDataBridgeDO.RocketMQConfig config) { + try { + // 1. 获取或创建 Producer + DefaultMQProducer producer = PRODUCER_CACHE.get(config); + + // 2.1 创建消息对象，指定Topic、Tag和消息体 + Message msg = new Message( + config.getTopic(), + config.getTags(), + message.toString().getBytes(RemotingHelper.DEFAULT_CHARSET) + ); + // 2.2 发送同步消息并处理结果 + SendResult sendResult = producer.send(msg); + // 2.3 处理发送结果 + if (SendStatus.SEND_OK.equals(sendResult.getSendStatus())) { + log.info("[executeRocketMQ][message({}) config({}) 发送成功，结果({})]", message, config, sendResult); + } else { + log.error("[executeRocketMQ][message({}) config({}) 发送失败，结果({})]", message, config, sendResult); + } + } catch (Exception e) { + log.error("[executeRocketMQ][message({}) config({}) 发送异常]", message, config, e); + } + } + + // TODO @芋艿：测试代码，后续清理 + public static void main(String[] args) { + // 1. 创建一个共享的实例 + IotRocketMQDataBridgeExecute action = new IotRocketMQDataBridgeExecute(); + + // 2. 创建共享的配置 + IotDataBridgeDO.RocketMQConfig config = new IotDataBridgeDO.RocketMQConfig(); + config.setNameServer("127.0.0.1:9876"); + config.setGroup("test-group"); + config.setTopic("test-topic"); + config.setTags("test-tag"); + + // 3. 创建共享的消息 + IotDeviceMessage message = IotDeviceMessage.builder() + .requestId("TEST-001") + .productKey("testProduct") + .deviceName("testDevice") + .deviceKey("testDeviceKey") + .type("property") + .identifier("temperature") + .data("{\"value\": 60}") + .reportTime(LocalDateTime.now()) + .tenantId(1L) + .build(); + + // 4. 执行两次测试，验证缓存 + log.info("[main][第一次执行，应该会创建新的 producer]"); + action.executeRocketMQ(message, config); + + log.info("[main][第二次执行，应该会复用缓存的 producer]"); + action.executeRocketMQ(message, config); + } + +} diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/execute/IotDataBridgeExecute.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/execute/IotDataBridgeExecute.java deleted file mode 100644 index 8979d21..0000000 --- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/execute/IotDataBridgeExecute.java +++ /dev/null @@ -1,32 +0,0 @@ -package cn.iocoder.yudao.module.iot.service.rule.execute; - -import cn.iocoder.yudao.module.iot.dal.dataobject.rule.IotDataBridgeDO; -import cn.iocoder.yudao.module.iot.mq.message.IotDeviceMessage; - - -/** - * IoT 数据桥梁的执行器 execute 接口 - * - * @author HUIHUI - */ -public interface IotDataBridgeExecute { - - /** - * 执行数据桥接操作 - * - * @param message 设备消息 - * @param dataBridge 数据桥梁 - */ - void execute(IotDeviceMessage message, IotDataBridgeDO dataBridge); - - // TODO @芋艿：因为下面的，都是有状态的，所以通过 guava 缓存连接，然后通过 RemovalNotification 实现关闭。例如说，一次新建有效期是 10 分钟； - // TODO @芋艿：mq-redis - // TODO @芋艿：mq-数据库 - // TODO @芋艿：kafka - // TODO @芋艿：rocketmq - // TODO @芋艿：rabbitmq - // TODO @芋艿：mqtt - // TODO @芋艿：tcp - // TODO @芋艿：websocket - -} diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/execute/IotHttpDataBridgeExecute.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/execute/IotHttpDataBridgeExecute.java deleted file mode 100644 index 861fadd..0000000 --- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/execute/IotHttpDataBridgeExecute.java +++ /dev/null @@ -1,93 +0,0 @@ -package cn.iocoder.yudao.module.iot.service.rule.execute; - -import cn.hutool.core.collection.CollUtil; -import cn.iocoder.yudao.framework.common.util.http.HttpUtils; -import cn.iocoder.yudao.framework.common.util.json.JsonUtils; -import cn.iocoder.yudao.module.iot.dal.dataobject.rule.IotDataBridgeDO; -import cn.iocoder.yudao.module.iot.enums.rule.IotDataBridgTypeEnum; -import cn.iocoder.yudao.module.iot.mq.message.IotDeviceMessage; -import jakarta.annotation.Resource; -import lombok.extern.slf4j.Slf4j; -import org.springframework.http.*; -import org.springframework.stereotype.Component; -import org.springframework.web.client.RestTemplate; -import org.springframework.web.util.UriComponentsBuilder; - -import java.util.HashMap; -import java.util.Map; - -import static cn.iocoder.yudao.framework.web.core.util.WebFrameworkUtils.HEADER_TENANT_ID; - -/** - * Http 的 {@link IotDataBridgeExecute} 实现类 - * - * @author HUIHUI - */ -@Component -@Slf4j -public class IotHttpDataBridgeExecute implements IotDataBridgeExecute { - - @Resource - private RestTemplate restTemplate; - - @Override - public void execute(IotDeviceMessage message, IotDataBridgeDO dataBridge) { - // 1.1 校验数据桥接的类型 == HTTP - if (!IotDataBridgTypeEnum.HTTP.getType().equals(dataBridge.getType())) { - return; - } - // 1.2 执行 HTTP 请求 - executeHttp(message, (IotDataBridgeDO.HttpConfig) dataBridge.getConfig()); - } - - @SuppressWarnings({"unchecked", "deprecation"}) - private void executeHttp(IotDeviceMessage message, IotDataBridgeDO.HttpConfig config) { - String url = null; - HttpMethod method = HttpMethod.valueOf(config.getMethod().toUpperCase()); - HttpEntity requestEntity = null; - ResponseEntity responseEntity = null; - try { - // 1.1 构建 Header - HttpHeaders headers = new HttpHeaders(); - if (CollUtil.isNotEmpty(config.getHeaders())) { - config.getHeaders().putAll(config.getHeaders()); - } - headers.add(HEADER_TENANT_ID, message.getTenantId().toString()); - // 1.2 构建 URL - UriComponentsBuilder uriBuilder = UriComponentsBuilder.fromUriString(config.getUrl()); - if (CollUtil.isNotEmpty(config.getQuery())) { - config.getQuery().forEach(uriBuilder::queryParam); - } - // 1.3 构建请求体 - if (method == HttpMethod.GET) { - uriBuilder.queryParam("message", HttpUtils.encodeUtf8(JsonUtils.toJsonString(message))); - url = uriBuilder.build().toUriString(); - requestEntity = new HttpEntity<>(headers); - } else { - url = uriBuilder.build().toUriString(); - Map requestBody = JsonUtils.parseObject(config.getBody(), Map.class); - if (requestBody == null) { - requestBody = new HashMap<>(); - } - requestBody.put("message", message); - headers.add(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_UTF8_VALUE); - requestEntity = new HttpEntity<>(JsonUtils.toJsonString(requestBody), headers); - } - - // 2.1 发送请求 - responseEntity = restTemplate.exchange(url, method, requestEntity, String.class); - // 2.2 记录日志 - if (responseEntity.getStatusCode().is2xxSuccessful()) { - log.info("[executeHttp][message({}) config({}) url({}) method({}) requestEntity({}) 请求成功({})]", - message, config, url, method, requestEntity, responseEntity); - } else { - log.error("[executeHttp][message({}) config({}) url({}) method({}) requestEntity({}) 请求失败({})]", - message, config, url, method, requestEntity, responseEntity); - } - } catch (Exception e) { - log.error("[executeHttp][message({}) config({}) url({}) method({}) requestEntity({}) 请求异常({})]", - message, config, url, method, requestEntity, responseEntity, e); - } - } - -} \ No newline at end of file diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/execute/IotRocketMQDataBridgeExecute.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/execute/IotRocketMQDataBridgeExecute.java deleted file mode 100644 index 11ca233..0000000 --- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/execute/IotRocketMQDataBridgeExecute.java +++ /dev/null @@ -1,127 +0,0 @@ -package cn.iocoder.yudao.module.iot.service.rule.execute; - -import cn.iocoder.yudao.module.iot.dal.dataobject.rule.IotDataBridgeDO; -import cn.iocoder.yudao.module.iot.enums.rule.IotDataBridgTypeEnum; -import cn.iocoder.yudao.module.iot.mq.message.IotDeviceMessage; -import com.google.common.cache.CacheBuilder; -import com.google.common.cache.CacheLoader; -import com.google.common.cache.LoadingCache; -import lombok.extern.slf4j.Slf4j; -import org.apache.rocketmq.client.producer.DefaultMQProducer; -import org.apache.rocketmq.client.producer.SendResult; -import org.apache.rocketmq.client.producer.SendStatus; -import org.apache.rocketmq.common.message.Message; -import org.apache.rocketmq.remoting.common.RemotingHelper; -import org.springframework.stereotype.Component; - -import java.time.Duration; -import java.time.LocalDateTime; -import java.util.concurrent.Executors; - -/** - * RocketMQ 的 {@link IotDataBridgeExecute} 实现类 - * - * @author HUIHUI - */ -@Component -@Slf4j -public class IotRocketMQDataBridgeExecute implements IotDataBridgeExecute { - - /** - * 针对 {@link IotDataBridgeDO.RocketMQConfig} 的 DefaultMQProducer 缓存 - */ - private final LoadingCache PRODUCER_CACHE = CacheBuilder.newBuilder() - // 只阻塞当前数据加载线程，其他线程返回旧值 - .refreshAfterWrite(Duration.ofMinutes(10)) - // 增加移除监听器，自动关闭 producer - .removalListener(notification -> { - DefaultMQProducer producer = (DefaultMQProducer) notification.getValue(); - if (producer != null) { - try { - producer.shutdown(); - log.info("[PRODUCER_CACHE][配置({}) 对应的 producer 已关闭]", notification.getKey()); - } catch (Exception e) { - log.error("[PRODUCER_CACHE][配置({}) 对应的 producer 关闭失败]", notification.getKey(), e); - } - } - }) - // 通过 asyncReloading 实现全异步加载，包括 refreshAfterWrite 被阻塞的加载线程 - .build(CacheLoader.asyncReloading(new CacheLoader() { - @Override - public DefaultMQProducer load(IotDataBridgeDO.RocketMQConfig config) throws Exception { - DefaultMQProducer producer = new DefaultMQProducer(config.getGroup()); - producer.setNamesrvAddr(config.getNameServer()); - producer.start(); - log.info("[PRODUCER_CACHE][配置({}) 对应的 producer 已创建并启动]", config); - return producer; - } - }, Executors.newCachedThreadPool())); - - @Override - public void execute(IotDeviceMessage message, IotDataBridgeDO dataBridge) { - // 1.1 校验数据桥接的类型 == ROCKETMQ - if (!IotDataBridgTypeEnum.ROCKETMQ.getType().equals(dataBridge.getType())) { - return; - } - // 1.2 执行 RocketMQ 发送消息 - executeRocketMQ(message, (IotDataBridgeDO.RocketMQConfig) dataBridge.getConfig()); - } - - private void executeRocketMQ(IotDeviceMessage message, IotDataBridgeDO.RocketMQConfig config) { - try { - // 1. 获取或创建 Producer - DefaultMQProducer producer = PRODUCER_CACHE.get(config); - - // 2.1 创建消息对象，指定Topic、Tag和消息体 - Message msg = new Message( - config.getTopic(), - config.getTags(), - message.toString().getBytes(RemotingHelper.DEFAULT_CHARSET) - ); - // 2.2 发送同步消息并处理结果 - SendResult sendResult = producer.send(msg); - // 2.3 处理发送结果 - if (SendStatus.SEND_OK.equals(sendResult.getSendStatus())) { - log.info("[executeRocketMQ][message({}) config({}) 发送成功，结果({})]", message, config, sendResult); - } else { - log.error("[executeRocketMQ][message({}) config({}) 发送失败，结果({})]", message, config, sendResult); - } - } catch (Exception e) { - log.error("[executeRocketMQ][message({}) config({}) 发送异常]", message, config, e); - } - } - - // TODO @芋艿：测试代码，后续清理 - public static void main(String[] args) { - // 1. 创建一个共享的实例 - IotRocketMQDataBridgeExecute action = new IotRocketMQDataBridgeExecute(); - - // 2. 创建共享的配置 - IotDataBridgeDO.RocketMQConfig config = new IotDataBridgeDO.RocketMQConfig(); - config.setNameServer("127.0.0.1:9876"); - config.setGroup("test-group"); - config.setTopic("test-topic"); - config.setTags("test-tag"); - - // 3. 创建共享的消息 - IotDeviceMessage message = IotDeviceMessage.builder() - .requestId("TEST-001") - .productKey("testProduct") - .deviceName("testDevice") - .deviceKey("testDeviceKey") - .type("property") - .identifier("temperature") - .data("{\"value\": 60}") - .reportTime(LocalDateTime.now()) - .tenantId(1L) - .build(); - - // 4. 执行两次测试，验证缓存 - log.info("[main][第一次执行，应该会创建新的 producer]"); - action.executeRocketMQ(message, config); - - log.info("[main][第二次执行，应该会复用缓存的 producer]"); - action.executeRocketMQ(message, config); - } - -} -- libgit2 0.21.4