BJ_ZTEITS-Developers / parking-cloud-yudao

Commit 672247dbe4f068543018bc17e26f1b2b7d0ce04f

Authored by 芋道源码
Committed by Gitee
2 parents fa40ae1d 4be18af2

!1248 【代码优化】IoT: 基于 guava 对 producer 做 cache 

Merge pull request !1248 from puhui999/iot
Inline Side-by-side
Showing 6 changed files with 263 additions and 154 deletions
yudao-framework/yudao-spring-boot-starter-mq/pom.xml
  View file @672247d
@@ -36,6 +36,7 @@ @@ -36,6 +36,7 @@
36 <dependency> 36 <dependency>
37 <groupId>org.apache.rocketmq</groupId> 37 <groupId>org.apache.rocketmq</groupId>
38 <artifactId>rocketmq-spring-boot-starter</artifactId> 38 <artifactId>rocketmq-spring-boot-starter</artifactId>
  39 + <optional>true</optional>
39 </dependency> 40 </dependency>
40 </dependencies> 41 </dependencies>
41 42
yudao-module-iot/yudao-module-iot-biz/pom.xml
  View file @672247d
@@ -75,10 +75,11 @@ @@ -75,10 +75,11 @@
75 <!-- <groupId>org.eclipse.paho</groupId> &lt;!&ndash; MQTT &ndash;&gt;--> 75 <!-- <groupId>org.eclipse.paho</groupId> &lt;!&ndash; MQTT &ndash;&gt;-->
76 <!-- <artifactId>org.eclipse.paho.client.mqttv3</artifactId>--> 76 <!-- <artifactId>org.eclipse.paho.client.mqttv3</artifactId>-->
77 <!-- </dependency>--> 77 <!-- </dependency>-->
78 - <!-- 工具类相关 --> 78 + <!-- 消息队列相关 -->
79 <dependency> 79 <dependency>
80 - <groupId>cn.iocoder.boot</groupId>  
81 - <artifactId>yudao-spring-boot-starter-mq</artifactId> 80 + <groupId>org.apache.rocketmq</groupId>
  81 + <artifactId>rocketmq-spring-boot-starter</artifactId>
  82 + <optional>true</optional>
82 </dependency> 83 </dependency>
83 84
84 <dependency> 85 <dependency>
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/action/IotRuleSceneDataBridgeAction.java
  View file @672247d
1 package cn.iocoder.yudao.module.iot.service.rule.action; 1 package cn.iocoder.yudao.module.iot.service.rule.action;
2 2
3 -import cn.hutool.core.collection.CollUtil;  
4 import cn.hutool.core.lang.Assert; 3 import cn.hutool.core.lang.Assert;
5 import cn.iocoder.yudao.framework.common.enums.CommonStatusEnum; 4 import cn.iocoder.yudao.framework.common.enums.CommonStatusEnum;
6 -import cn.iocoder.yudao.framework.common.util.http.HttpUtils;  
7 -import cn.iocoder.yudao.framework.common.util.json.JsonUtils;  
8 import cn.iocoder.yudao.module.iot.dal.dataobject.rule.IotDataBridgeDO; 5 import cn.iocoder.yudao.module.iot.dal.dataobject.rule.IotDataBridgeDO;
9 import cn.iocoder.yudao.module.iot.dal.dataobject.rule.IotRuleSceneDO; 6 import cn.iocoder.yudao.module.iot.dal.dataobject.rule.IotRuleSceneDO;
10 -import cn.iocoder.yudao.module.iot.enums.rule.IotDataBridgTypeEnum;  
11 import cn.iocoder.yudao.module.iot.enums.rule.IotRuleSceneActionTypeEnum; 7 import cn.iocoder.yudao.module.iot.enums.rule.IotRuleSceneActionTypeEnum;
12 import cn.iocoder.yudao.module.iot.mq.message.IotDeviceMessage; 8 import cn.iocoder.yudao.module.iot.mq.message.IotDeviceMessage;
13 import cn.iocoder.yudao.module.iot.service.rule.IotDataBridgeService; 9 import cn.iocoder.yudao.module.iot.service.rule.IotDataBridgeService;
  10 +import cn.iocoder.yudao.module.iot.service.rule.execute.IotDataBridgeExecute;
14 import jakarta.annotation.Resource; 11 import jakarta.annotation.Resource;
15 import lombok.extern.slf4j.Slf4j; 12 import lombok.extern.slf4j.Slf4j;
16 -import org.apache.rocketmq.client.producer.DefaultMQProducer;  
17 -import org.apache.rocketmq.client.producer.SendResult;  
18 -import org.apache.rocketmq.client.producer.SendStatus;  
19 -import org.apache.rocketmq.common.message.Message;  
20 -import org.apache.rocketmq.remoting.common.RemotingHelper;  
21 -import org.springframework.http.*;  
22 import org.springframework.stereotype.Component; 13 import org.springframework.stereotype.Component;
23 -import org.springframework.web.client.RestTemplate;  
24 -import org.springframework.web.util.UriComponentsBuilder;  
25 14
26 -import java.time.LocalDateTime;  
27 -import java.util.HashMap;  
28 -import java.util.Map;  
29 -  
30 -import static cn.iocoder.yudao.framework.web.core.util.WebFrameworkUtils.HEADER_TENANT_ID; 15 +import java.util.List;
31 16
32 /** 17 /**
33 * IoT 数据桥梁的 {@link IotRuleSceneAction} 实现类 18 * IoT 数据桥梁的 {@link IotRuleSceneAction} 实现类
@@ -42,10 +27,9 @@ import static cn.iocoder.yudao.framework.web.core.util.WebFrameworkUtils.HEADER_ @@ -42,10 +27,9 @@ import static cn.iocoder.yudao.framework.web.core.util.WebFrameworkUtils.HEADER_
42 public class IotRuleSceneDataBridgeAction implements IotRuleSceneAction { 27 public class IotRuleSceneDataBridgeAction implements IotRuleSceneAction {
43 28
44 @Resource 29 @Resource
45 - private RestTemplate restTemplate;  
46 -  
47 - @Resource  
48 private IotDataBridgeService dataBridgeService; 30 private IotDataBridgeService dataBridgeService;
  31 + @Resource
  32 + private List<IotDataBridgeExecute> dataBridgeExecutes;
49 33
50 @Override 34 @Override
51 public void execute(IotDeviceMessage message, IotRuleSceneDO.ActionConfig config) { 35 public void execute(IotDeviceMessage message, IotRuleSceneDO.ActionConfig config) {
@@ -65,26 +49,8 @@ public class IotRuleSceneDataBridgeAction implements IotRuleSceneAction { @@ -65,26 +49,8 @@ public class IotRuleSceneDataBridgeAction implements IotRuleSceneAction {
65 return; 49 return;
66 } 50 }
67 51
68 - // 2.1 执行 HTTP 请求  
69 - if (IotDataBridgTypeEnum.HTTP.getType().equals(dataBridge.getType())) {  
70 - executeHttp(message, (IotDataBridgeDO.HttpConfig) dataBridge.getConfig());  
71 - return;  
72 - }  
73 - // 2.2 执行 RocketMQ 发送消息  
74 - if (IotDataBridgTypeEnum.ROCKETMQ.getType().equals(dataBridge.getType())) {  
75 - executeRocketMQ(message, (IotDataBridgeDO.RocketMQConfig) dataBridge.getConfig());  
76 - return;  
77 - }  
78 -  
79 - // TODO @芋艿：因为下面的，都是有状态的，所以通过 guava 缓存连接，然后通过 RemovalNotification 实现关闭。例如说，一次新建有效期是 10 分钟；  
80 - // TODO @芋艿：mq-redis  
81 - // TODO @芋艿：mq-数据库  
82 - // TODO @芋艿：kafka  
83 - // TODO @芋艿：rocketmq  
84 - // TODO @芋艿：rabbitmq  
85 - // TODO @芋艿：mqtt  
86 - // TODO @芋艿：tcp  
87 - // TODO @芋艿：websocket 52 + // 2.1 执行数据桥接操作
  53 + dataBridgeExecutes.forEach(execute -> execute.execute(message, dataBridge));
88 } 54 }
89 55
90 @Override 56 @Override
@@ -92,115 +58,4 @@ public class IotRuleSceneDataBridgeAction implements IotRuleSceneAction { @@ -92,115 +58,4 @@ public class IotRuleSceneDataBridgeAction implements IotRuleSceneAction {
92 return IotRuleSceneActionTypeEnum.DATA_BRIDGE; 58 return IotRuleSceneActionTypeEnum.DATA_BRIDGE;
93 } 59 }
94 60
95 - @SuppressWarnings({"unchecked", "deprecation"})  
96 - private void executeHttp(IotDeviceMessage message, IotDataBridgeDO.HttpConfig config) {  
97 - String url = null;  
98 - HttpMethod method = HttpMethod.valueOf(config.getMethod().toUpperCase());  
99 - HttpEntity<String> requestEntity = null;  
100 - ResponseEntity<String> responseEntity = null;  
101 - try {  
102 - // 1.1 构建 Header  
103 - HttpHeaders headers = new HttpHeaders();  
104 - if (CollUtil.isNotEmpty(config.getHeaders())) {  
105 - config.getHeaders().putAll(config.getHeaders());  
106 - }  
107 - headers.add(HEADER_TENANT_ID, message.getTenantId().toString());  
108 - // 1.2 构建 URL  
109 - UriComponentsBuilder uriBuilder = UriComponentsBuilder.fromUriString(config.getUrl());  
110 - if (CollUtil.isNotEmpty(config.getQuery())) {  
111 - config.getQuery().forEach(uriBuilder::queryParam);  
112 - }  
113 - // 1.3 构建请求体  
114 - if (method == HttpMethod.GET) {  
115 - uriBuilder.queryParam("message", HttpUtils.encodeUtf8(JsonUtils.toJsonString(message)));  
116 - url = uriBuilder.build().toUriString();  
117 - requestEntity = new HttpEntity<>(headers);  
118 - } else {  
119 - url = uriBuilder.build().toUriString();  
120 - Map<String, Object> requestBody = JsonUtils.parseObject(config.getBody(), Map.class);  
121 - if (requestBody == null) {  
122 - requestBody = new HashMap<>();  
123 - }  
124 - requestBody.put("message", message);  
125 - headers.add(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_UTF8_VALUE);  
126 - requestEntity = new HttpEntity<>(JsonUtils.toJsonString(requestBody), headers);  
127 - }  
128 -  
129 - // 2.1 发送请求  
130 - responseEntity = restTemplate.exchange(url, method, requestEntity, String.class);  
131 - // 2.2 记录日志  
132 - if (responseEntity.getStatusCode().is2xxSuccessful()) {  
133 - log.info("[executeHttp][message({}) config({}) url({}) method({}) requestEntity({}) 请求成功({})]",  
134 - message, config, url, method, requestEntity, responseEntity);  
135 - } else {  
136 - log.error("[executeHttp][message({}) config({}) url({}) method({}) requestEntity({}) 请求失败({})]",  
137 - message, config, url, method, requestEntity, responseEntity);  
138 - }  
139 - } catch (Exception e) {  
140 - log.error("[executeHttp][message({}) config({}) url({}) method({}) requestEntity({}) 请求异常({})]",  
141 - message, config, url, method, requestEntity, responseEntity, e);  
142 - }  
143 - }  
144 -  
145 - private void executeRocketMQ(IotDeviceMessage message, IotDataBridgeDO.RocketMQConfig config) {  
146 - // 1.1 创建生产者实例，指定生产者组名  
147 - DefaultMQProducer producer = new DefaultMQProducer(config.getGroup());  
148 - // TODO @puhui999：可以考虑，基于 guava 做 cache，使用 config 作为 key，然后假个 listener 超时，销毁 producer  
149 - try {  
150 - // 1.2 设置 NameServer 地址  
151 - producer.setNamesrvAddr(config.getNameServer());  
152 - // 1.3 启动生产者  
153 - producer.start();  
154 -  
155 - // 2.1 创建消息对象，指定Topic、Tag和消息体  
156 - Message msg = new Message(  
157 - config.getTopic(),  
158 - config.getTags(),  
159 - message.toString().getBytes(RemotingHelper.DEFAULT_CHARSET)  
160 - );  
161 - // 2.2 发送同步消息并处理结果  
162 - SendResult sendResult = producer.send(msg);  
163 - // 2.3 处理发送结果  
164 - if (SendStatus.SEND_OK.equals(sendResult.getSendStatus())) {  
165 - log.info("[executeRocketMQ][message({}) config({}) 发送成功，结果({})]", message, config, sendResult);  
166 - } else {  
167 - log.error("[executeRocketMQ][message({}) config({}) 发送失败，结果({})]", message, config, sendResult);  
168 - }  
169 - } catch (Exception e) {  
170 - log.error("[executeRocketMQ][message({}) config({}) 发送异常]", message, config, e);  
171 - } finally {  
172 - // 3. 关闭生产者  
173 - producer.shutdown();  
174 - }  
175 - }  
176 -  
177 - // TODO @芋艿：测试代码，后续清理  
178 - public static void main(String[] args) {  
179 - // 1. 创建 IotRuleSceneDataBridgeAction 实例  
180 - IotRuleSceneDataBridgeAction action = new IotRuleSceneDataBridgeAction();  
181 -  
182 - // 2. 创建测试消息  
183 - IotDeviceMessage message = IotDeviceMessage.builder()  
184 - .requestId("TEST-001")  
185 - .productKey("testProduct")  
186 - .deviceName("testDevice")  
187 - .deviceKey("testDeviceKey")  
188 - .type("property")  
189 - .identifier("temperature")  
190 - .data("{\"value\": 60}")  
191 - .reportTime(LocalDateTime.now())  
192 - .tenantId(1L)  
193 - .build();  
194 -  
195 - // 3. 创建 RocketMQ 配置  
196 - IotDataBridgeDO.RocketMQConfig config = new IotDataBridgeDO.RocketMQConfig();  
197 - config.setNameServer("127.0.0.1:9876");  
198 - config.setGroup("test-group");  
199 - config.setTopic("test-topic");  
200 - config.setTags("test-tag");  
201 -  
202 - // 4. 执行测试  
203 - action.executeRocketMQ(message, config);  
204 - }  
205 -  
206 } 61 }
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/execute/IotDataBridgeExecute.java 0 → 100644
  View file @672247d
  1 +package cn.iocoder.yudao.module.iot.service.rule.execute;
  2 +
  3 +import cn.iocoder.yudao.module.iot.dal.dataobject.rule.IotDataBridgeDO;
  4 +import cn.iocoder.yudao.module.iot.mq.message.IotDeviceMessage;
  5 +
  6 +
  7 +/**
  8 + * IoT 数据桥梁的执行器 execute 接口
  9 + *
  10 + * @author HUIHUI
  11 + */
  12 +public interface IotDataBridgeExecute {
  13 +
  14 + /**
  15 + * 执行数据桥接操作
  16 + *
  17 + * @param message 设备消息
  18 + * @param dataBridge 数据桥梁
  19 + */
  20 + void execute(IotDeviceMessage message, IotDataBridgeDO dataBridge);
  21 +
  22 + // TODO @芋艿：因为下面的，都是有状态的，所以通过 guava 缓存连接，然后通过 RemovalNotification 实现关闭。例如说，一次新建有效期是 10 分钟；
  23 + // TODO @芋艿：mq-redis
  24 + // TODO @芋艿：mq-数据库
  25 + // TODO @芋艿：kafka
  26 + // TODO @芋艿：rocketmq
  27 + // TODO @芋艿：rabbitmq
  28 + // TODO @芋艿：mqtt
  29 + // TODO @芋艿：tcp
  30 + // TODO @芋艿：websocket
  31 +
  32 +}
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/execute/IotHttpDataBridgeExecute.java 0 → 100644
  View file @672247d
  1 +package cn.iocoder.yudao.module.iot.service.rule.execute;
  2 +
  3 +import cn.hutool.core.collection.CollUtil;
  4 +import cn.iocoder.yudao.framework.common.util.http.HttpUtils;
  5 +import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
  6 +import cn.iocoder.yudao.module.iot.dal.dataobject.rule.IotDataBridgeDO;
  7 +import cn.iocoder.yudao.module.iot.enums.rule.IotDataBridgTypeEnum;
  8 +import cn.iocoder.yudao.module.iot.mq.message.IotDeviceMessage;
  9 +import jakarta.annotation.Resource;
  10 +import lombok.extern.slf4j.Slf4j;
  11 +import org.springframework.http.*;
  12 +import org.springframework.stereotype.Component;
  13 +import org.springframework.web.client.RestTemplate;
  14 +import org.springframework.web.util.UriComponentsBuilder;
  15 +
  16 +import java.util.HashMap;
  17 +import java.util.Map;
  18 +
  19 +import static cn.iocoder.yudao.framework.web.core.util.WebFrameworkUtils.HEADER_TENANT_ID;
  20 +
  21 +/**
  22 + * Http 的 {@link IotDataBridgeExecute} 实现类
  23 + *
  24 + * @author HUIHUI
  25 + */
  26 +@Component
  27 +@Slf4j
  28 +public class IotHttpDataBridgeExecute implements IotDataBridgeExecute {
  29 +
  30 + @Resource
  31 + private RestTemplate restTemplate;
  32 +
  33 + @Override
  34 + public void execute(IotDeviceMessage message, IotDataBridgeDO dataBridge) {
  35 + // 1.1 校验数据桥接的类型 == HTTP
  36 + if (!IotDataBridgTypeEnum.HTTP.getType().equals(dataBridge.getType())) {
  37 + return;
  38 + }
  39 + // 1.2 执行 HTTP 请求
  40 + executeHttp(message, (IotDataBridgeDO.HttpConfig) dataBridge.getConfig());
  41 + }
  42 +
  43 + @SuppressWarnings({"unchecked", "deprecation"})
  44 + private void executeHttp(IotDeviceMessage message, IotDataBridgeDO.HttpConfig config) {
  45 + String url = null;
  46 + HttpMethod method = HttpMethod.valueOf(config.getMethod().toUpperCase());
  47 + HttpEntity<String> requestEntity = null;
  48 + ResponseEntity<String> responseEntity = null;
  49 + try {
  50 + // 1.1 构建 Header
  51 + HttpHeaders headers = new HttpHeaders();
  52 + if (CollUtil.isNotEmpty(config.getHeaders())) {
  53 + config.getHeaders().putAll(config.getHeaders());
  54 + }
  55 + headers.add(HEADER_TENANT_ID, message.getTenantId().toString());
  56 + // 1.2 构建 URL
  57 + UriComponentsBuilder uriBuilder = UriComponentsBuilder.fromUriString(config.getUrl());
  58 + if (CollUtil.isNotEmpty(config.getQuery())) {
  59 + config.getQuery().forEach(uriBuilder::queryParam);
  60 + }
  61 + // 1.3 构建请求体
  62 + if (method == HttpMethod.GET) {
  63 + uriBuilder.queryParam("message", HttpUtils.encodeUtf8(JsonUtils.toJsonString(message)));
  64 + url = uriBuilder.build().toUriString();
  65 + requestEntity = new HttpEntity<>(headers);
  66 + } else {
  67 + url = uriBuilder.build().toUriString();
  68 + Map<String, Object> requestBody = JsonUtils.parseObject(config.getBody(), Map.class);
  69 + if (requestBody == null) {
  70 + requestBody = new HashMap<>();
  71 + }
  72 + requestBody.put("message", message);
  73 + headers.add(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_UTF8_VALUE);
  74 + requestEntity = new HttpEntity<>(JsonUtils.toJsonString(requestBody), headers);
  75 + }
  76 +
  77 + // 2.1 发送请求
  78 + responseEntity = restTemplate.exchange(url, method, requestEntity, String.class);
  79 + // 2.2 记录日志
  80 + if (responseEntity.getStatusCode().is2xxSuccessful()) {
  81 + log.info("[executeHttp][message({}) config({}) url({}) method({}) requestEntity({}) 请求成功({})]",
  82 + message, config, url, method, requestEntity, responseEntity);
  83 + } else {
  84 + log.error("[executeHttp][message({}) config({}) url({}) method({}) requestEntity({}) 请求失败({})]",
  85 + message, config, url, method, requestEntity, responseEntity);
  86 + }
  87 + } catch (Exception e) {
  88 + log.error("[executeHttp][message({}) config({}) url({}) method({}) requestEntity({}) 请求异常({})]",
  89 + message, config, url, method, requestEntity, responseEntity, e);
  90 + }
  91 + }
  92 +
  93 +}
0 \ No newline at end of file 94 \ No newline at end of file
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/execute/IotRocketMQDataBridgeExecute.java 0 → 100644
  View file @672247d
  1 +package cn.iocoder.yudao.module.iot.service.rule.execute;
  2 +
  3 +import cn.iocoder.yudao.module.iot.dal.dataobject.rule.IotDataBridgeDO;
  4 +import cn.iocoder.yudao.module.iot.enums.rule.IotDataBridgTypeEnum;
  5 +import cn.iocoder.yudao.module.iot.mq.message.IotDeviceMessage;
  6 +import com.google.common.cache.CacheBuilder;
  7 +import com.google.common.cache.CacheLoader;
  8 +import com.google.common.cache.LoadingCache;
  9 +import lombok.extern.slf4j.Slf4j;
  10 +import org.apache.rocketmq.client.producer.DefaultMQProducer;
  11 +import org.apache.rocketmq.client.producer.SendResult;
  12 +import org.apache.rocketmq.client.producer.SendStatus;
  13 +import org.apache.rocketmq.common.message.Message;
  14 +import org.apache.rocketmq.remoting.common.RemotingHelper;
  15 +import org.springframework.stereotype.Component;
  16 +
  17 +import java.time.Duration;
  18 +import java.time.LocalDateTime;
  19 +import java.util.concurrent.Executors;
  20 +
  21 +/**
  22 + * RocketMQ 的 {@link IotDataBridgeExecute} 实现类
  23 + *
  24 + * @author HUIHUI
  25 + */
  26 +@Component
  27 +@Slf4j
  28 +public class IotRocketMQDataBridgeExecute implements IotDataBridgeExecute {
  29 +
  30 + /**
  31 + * 针对 {@link IotDataBridgeDO.RocketMQConfig} 的 DefaultMQProducer 缓存
  32 + */
  33 + private final LoadingCache<IotDataBridgeDO.RocketMQConfig, DefaultMQProducer> PRODUCER_CACHE = CacheBuilder.newBuilder()
  34 + // 只阻塞当前数据加载线程，其他线程返回旧值
  35 + .refreshAfterWrite(Duration.ofMinutes(10))
  36 + // 增加移除监听器，自动关闭 producer
  37 + .removalListener(notification -> {
  38 + DefaultMQProducer producer = (DefaultMQProducer) notification.getValue();
  39 + if (producer != null) {
  40 + try {
  41 + producer.shutdown();
  42 + log.info("[PRODUCER_CACHE][配置({}) 对应的 producer 已关闭]", notification.getKey());
  43 + } catch (Exception e) {
  44 + log.error("[PRODUCER_CACHE][配置({}) 对应的 producer 关闭失败]", notification.getKey(), e);
  45 + }
  46 + }
  47 + })
  48 + // 通过 asyncReloading 实现全异步加载，包括 refreshAfterWrite 被阻塞的加载线程
  49 + .build(CacheLoader.asyncReloading(new CacheLoader<IotDataBridgeDO.RocketMQConfig, DefaultMQProducer>() {
  50 + @Override
  51 + public DefaultMQProducer load(IotDataBridgeDO.RocketMQConfig config) throws Exception {
  52 + DefaultMQProducer producer = new DefaultMQProducer(config.getGroup());
  53 + producer.setNamesrvAddr(config.getNameServer());
  54 + producer.start();
  55 + log.info("[PRODUCER_CACHE][配置({}) 对应的 producer 已创建并启动]", config);
  56 + return producer;
  57 + }
  58 + }, Executors.newCachedThreadPool()));
  59 +
  60 + @Override
  61 + public void execute(IotDeviceMessage message, IotDataBridgeDO dataBridge) {
  62 + // 1.1 校验数据桥接的类型 == ROCKETMQ
  63 + if (!IotDataBridgTypeEnum.ROCKETMQ.getType().equals(dataBridge.getType())) {
  64 + return;
  65 + }
  66 + // 1.2 执行 RocketMQ 发送消息
  67 + executeRocketMQ(message, (IotDataBridgeDO.RocketMQConfig) dataBridge.getConfig());
  68 + }
  69 +
  70 + private void executeRocketMQ(IotDeviceMessage message, IotDataBridgeDO.RocketMQConfig config) {
  71 + try {
  72 + // 1. 获取或创建 Producer
  73 + DefaultMQProducer producer = PRODUCER_CACHE.get(config);
  74 +
  75 + // 2.1 创建消息对象，指定Topic、Tag和消息体
  76 + Message msg = new Message(
  77 + config.getTopic(),
  78 + config.getTags(),
  79 + message.toString().getBytes(RemotingHelper.DEFAULT_CHARSET)
  80 + );
  81 + // 2.2 发送同步消息并处理结果
  82 + SendResult sendResult = producer.send(msg);
  83 + // 2.3 处理发送结果
  84 + if (SendStatus.SEND_OK.equals(sendResult.getSendStatus())) {
  85 + log.info("[executeRocketMQ][message({}) config({}) 发送成功，结果({})]", message, config, sendResult);
  86 + } else {
  87 + log.error("[executeRocketMQ][message({}) config({}) 发送失败，结果({})]", message, config, sendResult);
  88 + }
  89 + } catch (Exception e) {
  90 + log.error("[executeRocketMQ][message({}) config({}) 发送异常]", message, config, e);
  91 + }
  92 + }
  93 +
  94 + // TODO @芋艿：测试代码，后续清理
  95 + public static void main(String[] args) {
  96 + // 1. 创建一个共享的实例
  97 + IotRocketMQDataBridgeExecute action = new IotRocketMQDataBridgeExecute();
  98 +
  99 + // 2. 创建共享的配置
  100 + IotDataBridgeDO.RocketMQConfig config = new IotDataBridgeDO.RocketMQConfig();
  101 + config.setNameServer("127.0.0.1:9876");
  102 + config.setGroup("test-group");
  103 + config.setTopic("test-topic");
  104 + config.setTags("test-tag");
  105 +
  106 + // 3. 创建共享的消息
  107 + IotDeviceMessage message = IotDeviceMessage.builder()
  108 + .requestId("TEST-001")
  109 + .productKey("testProduct")
  110 + .deviceName("testDevice")
  111 + .deviceKey("testDeviceKey")
  112 + .type("property")
  113 + .identifier("temperature")
  114 + .data("{\"value\": 60}")
  115 + .reportTime(LocalDateTime.now())
  116 + .tenantId(1L)
  117 + .build();
  118 +
  119 + // 4. 执行两次测试，验证缓存
  120 + log.info("[main][第一次执行，应该会创建新的 producer]");
  121 + action.executeRocketMQ(message, config);
  122 +
  123 + log.info("[main][第二次执行，应该会复用缓存的 producer]");
  124 + action.executeRocketMQ(message, config);
  125 + }
  126 +
  127 +}