szz
/
ShiShiYiBan
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

【代码优化】IoT: 优化数据桥梁的执行器减少子类代码冗余

This commit is contained in:
puhui999 2025-03-03 12:48:07 +08:00
parent 61ea09488e
commit 3b54deb989
7 changed files with 56 additions and 55 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/action/databridge/AbstractCacheableDataBridgeExecute.java
View File

@ -33,7 +33,7 @@ import java.time.Duration;
* @author HUIHUI
*/
@Slf4j
public abstract class AbstractCacheableDataBridgeExecute<Config, Producer> implements IotDataBridgeExecute {
public abstract class AbstractCacheableDataBridgeExecute<Config, Producer> implements IotDataBridgeExecute<Config> {
/**
* Producer 缓存

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

@ -9,9 +9,14 @@ import cn.iocoder.yudao.module.iot.mq.message.IotDeviceMessage;
*
* @author HUIHUI
*/
public interface IotDataBridgeExecute {
public interface IotDataBridgeExecute<Config> {
// TODO @huihui：要不搞个 getType？然后 execute0 由子类实现。这样，子类的 executeRedisStream ，其实就是 execute0 了。
/**
* 获取数据桥梁类型
*
* @return 数据桥梁类型
*/
Integer getType();
/**
* 执行数据桥梁操作
@ -19,6 +24,23 @@ public interface IotDataBridgeExecute {
* @param message 设备消息
* @param dataBridge 数据桥梁
*/
void execute(IotDeviceMessage message, IotDataBridgeDO dataBridge);
@SuppressWarnings({"unchecked"})
default void execute(IotDeviceMessage message, IotDataBridgeDO dataBridge) {
// 1.1 校验数据桥梁类型
if (!getType().equals(dataBridge.getType())) {
return;
}
// 1.2 执行对应的数据桥梁发送消息
execute0(message, (Config) dataBridge.getConfig());
}
/**
* 【真正】执行数据桥梁操作
*
* @param message 设备消息
* @param config 桥梁配置
*/
void execute0(IotDeviceMessage message, Config config);
}

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

@ -25,23 +25,19 @@ import static cn.iocoder.yudao.framework.web.core.util.WebFrameworkUtils.HEADER_
*/
@Component
@Slf4j
public class IotHttpDataBridgeExecute implements IotDataBridgeExecute {
public class IotHttpDataBridgeExecute implements IotDataBridgeExecute<IotDataBridgeDO.HttpConfig> {
@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());
public Integer getType() {
return IotDataBridgTypeEnum.HTTP.getType();
}
@Override
@SuppressWarnings({"unchecked", "deprecation"})
private void executeHttp(IotDeviceMessage message, IotDataBridgeDO.HttpConfig config) {
public void execute0(IotDeviceMessage message, IotDataBridgeDO.HttpConfig config) {
String url = null;
HttpMethod method = HttpMethod.valueOf(config.getMethod().toUpperCase());
HttpEntity<String> requestEntity = null;

16
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/action/databridge/IotKafkaMQDataBridgeExecute.java
View File

@ -30,16 +30,12 @@ public class IotKafkaMQDataBridgeExecute extends
private static final Duration SEND_TIMEOUT = Duration.ofMillis(10);
@Override
public void execute(IotDeviceMessage message, IotDataBridgeDO dataBridge) {
// 1. 校验数据桥梁的类型 == KAFKA
if (!IotDataBridgTypeEnum.KAFKA.getType().equals(dataBridge.getType())) {
return;
}
// 2. 执行 Kafka 发送消息
executeKafka(message, (IotDataBridgeDO.KafkaMQConfig) dataBridge.getConfig());
public Integer getType() {
return IotDataBridgTypeEnum.KAFKA.getType();
}
private void executeKafka(IotDeviceMessage message, IotDataBridgeDO.KafkaMQConfig config) {
@Override
public void execute0(IotDeviceMessage message, IotDataBridgeDO.KafkaMQConfig config) {
try {
// 1. 获取或创建 KafkaTemplate
KafkaTemplate<String, String> kafkaTemplate = getProducer(config);
@ -113,10 +109,10 @@ public class IotKafkaMQDataBridgeExecute extends
// 4. 执行两次测试，验证缓存
log.info("[main][第一次执行，应该会创建新的 producer]");
action.executeKafka(message, config);
action.execute0(message, config);
log.info("[main][第二次执行，应该会复用缓存的 producer]");
action.executeKafka(message, config);
action.execute0(message, config);
}
}

17
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/action/databridge/IotRabbitMQDataBridgeExecute.java
View File

@ -22,17 +22,14 @@ import java.time.LocalDateTime;
public class IotRabbitMQDataBridgeExecute extends
AbstractCacheableDataBridgeExecute<IotDataBridgeDO.RabbitMQConfig, Channel> {
@Override
public void execute(IotDeviceMessage message, IotDataBridgeDO dataBridge) {
// 1.1 校验数据桥梁的类型 == RABBITMQ
if (!IotDataBridgTypeEnum.RABBITMQ.getType().equals(dataBridge.getType())) {
return;
}
// 1.2 执行 RabbitMQ 发送消息
executeRabbitMQ(message, (IotDataBridgeDO.RabbitMQConfig) dataBridge.getConfig());
public Integer getType() {
return IotDataBridgTypeEnum.RABBITMQ.getType();
}
private void executeRabbitMQ(IotDeviceMessage message, IotDataBridgeDO.RabbitMQConfig config) {
@Override
public void execute0(IotDeviceMessage message, IotDataBridgeDO.RabbitMQConfig config) {
try {
// 1. 获取或创建 Channel
Channel channel = getProducer(config);
@ -111,10 +108,10 @@ public class IotRabbitMQDataBridgeExecute extends
// 4. 执行两次测试，验证缓存
log.info("[main][第一次执行，应该会创建新的 channel]");
action.executeRabbitMQ(message, config);
action.execute0(message, config);
log.info("[main][第二次执行，应该会复用缓存的 channel]");
action.executeRabbitMQ(message, config);
action.execute0(message, config);
}
}

18
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/action/databridge/IotRedisStreamMQDataBridgeExecute.java
View File

@ -33,18 +33,12 @@ public class IotRedisStreamMQDataBridgeExecute extends
AbstractCacheableDataBridgeExecute<IotDataBridgeDO.RedisStreamMQConfig, RedisTemplate<String, Object>> {
@Override
public void execute(IotDeviceMessage message, IotDataBridgeDO dataBridge) {
// 1.1 校验数据桥梁类型
if (!IotDataBridgTypeEnum.REDIS_STREAM.getType().equals(dataBridge.getType())) {
return;
}
// 1.2 执行消息发送
executeRedisStream(message, (IotDataBridgeDO.RedisStreamMQConfig) dataBridge.getConfig());
public Integer getType() {
return IotDataBridgTypeEnum.REDIS_STREAM.getType();
}
@SuppressWarnings("unchecked")
// TODO @huihui：try catch 交给父类来做，子类不处理异常
private void executeRedisStream(IotDeviceMessage message, IotDataBridgeDO.RedisStreamMQConfig config) {
@Override
public void execute0(IotDeviceMessage message, IotDataBridgeDO.RedisStreamMQConfig config) {
try {
// 1. 获取 RedisTemplate
RedisTemplate<String, Object> redisTemplate = getProducer(config);
@ -133,10 +127,10 @@ public class IotRedisStreamMQDataBridgeExecute extends
// 4. 执行两次测试，验证缓存
log.info("[main][第一次执行，应该会创建新的 RedisTemplate]");
action.executeRedisStream(message, config);
action.execute0(message, config);
log.info("[main][第二次执行，应该会复用缓存的 RedisTemplate]");
action.executeRedisStream(message, config);
action.execute0(message, config);
}
}

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

@ -24,16 +24,12 @@ public class IotRocketMQDataBridgeExecute extends
AbstractCacheableDataBridgeExecute<IotDataBridgeDO.RocketMQConfig, DefaultMQProducer> {
@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());
public Integer getType() {
return IotDataBridgTypeEnum.ROCKETMQ.getType();
}
private void executeRocketMQ(IotDeviceMessage message, IotDataBridgeDO.RocketMQConfig config) {
@Override
public void execute0(IotDeviceMessage message, IotDataBridgeDO.RocketMQConfig config) {
try {
// 1. 获取或创建 Producer
DefaultMQProducer producer = getProducer(config);
@ -97,10 +93,10 @@ public class IotRocketMQDataBridgeExecute extends
// 4. 执行两次测试，验证缓存
log.info("[main][第一次执行，应该会创建新的 producer]");
action.executeRocketMQ(message, config);
action.execute0(message, config);
log.info("[main][第二次执行，应该会复用缓存的 producer]");
action.executeRocketMQ(message, config);
action.execute0(message, config);
}
}