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

【代码优化】IoT: Redis Stream 数据桥梁执行器

This commit is contained in:
puhui999 2025-03-24 16:13:19 +08:00
parent cd656fad4b
commit 3266bf0f98
5 changed files with 19 additions and 35 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/controller/admin/rule/vo/databridge/config/IotDataBridgeAbstractConfig.java
View File

@ -18,7 +18,7 @@ import lombok.Data;
@JsonSubTypes({
@JsonSubTypes.Type(value = IotDataBridgeHttpConfig.class, name = "1"),
@JsonSubTypes.Type(value = IotDataBridgeMqttConfig.class, name = "10"),
@JsonSubTypes.Type(value = IotDataBridgeRedisStreamMQConfig.class, name = "21"),
@JsonSubTypes.Type(value = IotDataBridgeRedisStreamConfig.class, name = "21"),
@JsonSubTypes.Type(value = IotDataBridgeRocketMQConfig.class, name = "30"),
@JsonSubTypes.Type(value = IotDataBridgeRabbitMQConfig.class, name = "31"),
@JsonSubTypes.Type(value = IotDataBridgeKafkaMQConfig.class, name = "32"),

3
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/rule/vo/databridge/config/IotDataBridgeRedisStreamMQConfig.java → yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/rule/vo/databridge/config/IotDataBridgeRedisStreamConfig.java
View File

@ -2,14 +2,13 @@ package cn.iocoder.yudao.module.iot.controller.admin.rule.vo.databridge.config;
import lombok.Data;
// TODO @puhui999：MQ 可以去掉哈。stream 更精准
/**
* IoT Redis Stream 配置 {@link IotDataBridgeAbstractConfig} 实现类
*
* @author HUIHUI
*/
@Data
public class IotDataBridgeRedisStreamMQConfig extends IotDataBridgeAbstractConfig {
public class IotDataBridgeRedisStreamConfig extends IotDataBridgeAbstractConfig {
/**
* Redis 服务器地址

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

@ -101,7 +101,7 @@ public abstract class AbstractCacheableDataBridgeExecute<Config, Producer> imple
@Override
@SuppressWarnings({"unchecked"})
public void execute(IotDeviceMessage message, IotDataBridgeDO dataBridge) {
if (ObjUtil.notEqual(message.getType(), getType())) {
if (ObjUtil.notEqual(dataBridge.getType(), getType())) {
return;
}
try {

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

@ -1,12 +1,9 @@
package cn.iocoder.yudao.module.iot.service.rule.action.databridge;
import cn.hutool.core.util.ReflectUtil;
import cn.hutool.core.util.StrUtil;
import cn.iocoder.yudao.module.iot.controller.admin.rule.vo.databridge.config.IotDataBridgeRedisStreamMQConfig;
import cn.iocoder.yudao.module.iot.controller.admin.rule.vo.databridge.config.IotDataBridgeRedisStreamConfig;
import cn.iocoder.yudao.module.iot.enums.rule.IotDataBridgeTypeEnum;
import cn.iocoder.yudao.module.iot.mq.message.IotDeviceMessage;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import lombok.extern.slf4j.Slf4j;
import org.redisson.Redisson;
import org.redisson.api.RedissonClient;
@ -21,14 +18,14 @@ import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.stereotype.Component;
/**
* Redis Stream MQ 的 {@link IotDataBridgeExecute} 实现类
* Redis Stream 的 {@link IotDataBridgeExecute} 实现类
*
* @author HUIHUI
*/
@Component
@Slf4j
public class IotRedisStreamMQDataBridgeExecute extends
AbstractCacheableDataBridgeExecute<IotDataBridgeRedisStreamMQConfig, RedisTemplate<String, Object>> {
public class IotRedisStreamDataBridgeExecute extends
AbstractCacheableDataBridgeExecute<IotDataBridgeRedisStreamConfig, RedisTemplate<String, Object>> {
@Override
public Integer getType() {
@ -36,7 +33,7 @@ public class IotRedisStreamMQDataBridgeExecute extends
}
@Override
public void execute0(IotDeviceMessage message, IotDataBridgeRedisStreamMQConfig config) throws Exception {
public void execute0(IotDeviceMessage message, IotDataBridgeRedisStreamConfig config) throws Exception {
// 1. 获取 RedisTemplate
RedisTemplate<String, Object> redisTemplate = getProducer(config);
@ -48,7 +45,7 @@ public class IotRedisStreamMQDataBridgeExecute extends
}
@Override
protected RedisTemplate<String, Object> initProducer(IotDataBridgeRedisStreamMQConfig config) {
protected RedisTemplate<String, Object> initProducer(IotDataBridgeRedisStreamConfig config) {
// 1.1 创建 Redisson 配置
Config redissonConfig = new Config();
SingleServerConfig serverConfig = redissonConfig.useSingleServer()
@ -59,20 +56,17 @@ public class IotRedisStreamMQDataBridgeExecute extends
serverConfig.setPassword(config.getPassword());
}
// TODO @huihui：看看能不能简化一些。按道理说，不用这么多的哈。
// 2.1 创建 RedissonClient
// TODO @芋艿：看看怎么优化
// 创建 RedisTemplate 并配置
RedissonClient redisson = Redisson.create(redissonConfig);
// 2.2 创建并配置 RedisTemplate
RedisTemplate<String, Object> template = new RedisTemplate<>();
// 设置 RedisConnection 工厂。😈 它就是实现多种 Java Redis 客户端接入的秘密工厂。感兴趣的胖友，可以自己去撸下。
template.setConnectionFactory(new RedissonConnectionFactory(redisson));
// 使用 String 序列化方式，序列化 KEY 。
// 设置序列化器
template.setKeySerializer(RedisSerializer.string());
template.setHashKeySerializer(RedisSerializer.string());
// 使用 JSON 序列化方式（库是 Jackson ），序列化 VALUE 。
template.setValueSerializer(buildRedisSerializer());
template.setHashValueSerializer(buildRedisSerializer());
template.afterPropertiesSet();// 初始化
template.setValueSerializer(RedisSerializer.json());
template.setHashValueSerializer(RedisSerializer.json());
template.afterPropertiesSet();
return template;
}
@ -84,13 +78,4 @@ public class IotRedisStreamMQDataBridgeExecute extends
}
}
// TODO @huihui：看看能不能简化一些。按道理说，不用这么多的哈。
public static RedisSerializer<?> buildRedisSerializer() {
RedisSerializer<Object> json = RedisSerializer.json();
// 解决 LocalDateTime 的序列化
ObjectMapper objectMapper = (ObjectMapper) ReflectUtil.getFieldValue(json, "mapper");
objectMapper.registerModules(new JavaTimeModule());
return json;
}
}

8
yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/rule/action/databridge/IotDataBridgeExecuteTest.java
View File

@ -92,12 +92,12 @@ public class IotDataBridgeExecuteTest extends BaseMockitoUnitTest {
}
@Test
public void testRedisStreamMQDataBridge() throws Exception {
public void testRedisStreamDataBridge() throws Exception {
// 1. 创建执行器实例
IotRedisStreamMQDataBridgeExecute action = new IotRedisStreamMQDataBridgeExecute();
IotRedisStreamDataBridgeExecute action = new IotRedisStreamDataBridgeExecute();
// 2. 创建配置
IotDataBridgeRedisMQConfig config = new IotDataBridgeRedisMQConfig()
IotDataBridgeRedisStreamConfig config = new IotDataBridgeRedisStreamConfig()
.setHost("127.0.0.1")
.setPort(6379)
.setDatabase(0)
@ -105,7 +105,7 @@ public class IotDataBridgeExecuteTest extends BaseMockitoUnitTest {
.setTopic("test-stream");
// 3. 执行测试并验证缓存
executeAndVerifyCache(action, config, "RedisStreamMQ");
executeAndVerifyCache(action, config, "RedisStream");
}
@Test