From 3266bf0f9829effe1eb0c7774ef2b0ff4265c7a2 Mon Sep 17 00:00:00 2001 From: puhui999 Date: Mon, 24 Mar 2025 16:13:19 +0800 Subject: [PATCH] 【代码优化】IoT: Redis Stream 数据桥梁执行器 --- yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/rule/vo/databridge/config/IotDataBridgeAbstractConfig.java | 2 +- yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/rule/vo/databridge/config/IotDataBridgeRedisStreamConfig.java | 34 ++++++++++++++++++++++++++++++++++ yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/rule/vo/databridge/config/IotDataBridgeRedisStreamMQConfig.java | 35 ----------------------------------- yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/action/databridge/AbstractCacheableDataBridgeExecute.java | 2 +- yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/action/databridge/IotRedisStreamDataBridgeExecute.java | 81 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/action/databridge/IotRedisStreamMQDataBridgeExecute.java | 96 ------------------------------------------------------------------------------------------------ yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/rule/action/databridge/IotDataBridgeExecuteTest.java | 8 ++++---- 7 files changed, 121 insertions(+), 137 deletions(-) create mode 100644 yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/rule/vo/databridge/config/IotDataBridgeRedisStreamConfig.java delete mode 100644 yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/rule/vo/databridge/config/IotDataBridgeRedisStreamMQConfig.java create mode 100644 yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/action/databridge/IotRedisStreamDataBridgeExecute.java delete mode 100644 yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/action/databridge/IotRedisStreamMQDataBridgeExecute.java diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/rule/vo/databridge/config/IotDataBridgeAbstractConfig.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/rule/vo/databridge/config/IotDataBridgeAbstractConfig.java index 527e79b..7bf714f 100644 --- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/rule/vo/databridge/config/IotDataBridgeAbstractConfig.java +++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/rule/vo/databridge/config/IotDataBridgeAbstractConfig.java @@ -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"), diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/rule/vo/databridge/config/IotDataBridgeRedisStreamConfig.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/rule/vo/databridge/config/IotDataBridgeRedisStreamConfig.java new file mode 100644 index 0000000..fc7a4c3 --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/rule/vo/databridge/config/IotDataBridgeRedisStreamConfig.java @@ -0,0 +1,34 @@ +package cn.iocoder.yudao.module.iot.controller.admin.rule.vo.databridge.config; + +import lombok.Data; + +/** + * IoT Redis Stream 配置 {@link IotDataBridgeAbstractConfig} 实现类 + * + * @author HUIHUI + */ +@Data +public class IotDataBridgeRedisStreamConfig extends IotDataBridgeAbstractConfig { + + /** + * Redis 服务器地址 + */ + private String host; + /** + * 端口 + */ + private Integer port; + /** + * 密码 + */ + private String password; + /** + * 数据库索引 + */ + private Integer database; + + /** + * 主题 + */ + private String topic; +} \ No newline at end of file diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/rule/vo/databridge/config/IotDataBridgeRedisStreamMQConfig.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/rule/vo/databridge/config/IotDataBridgeRedisStreamMQConfig.java deleted file mode 100644 index 3c9bb33..0000000 --- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/rule/vo/databridge/config/IotDataBridgeRedisStreamMQConfig.java +++ /dev/null @@ -1,35 +0,0 @@ -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 { - - /** - * Redis 服务器地址 - */ - private String host; - /** - * 端口 - */ - private Integer port; - /** - * 密码 - */ - private String password; - /** - * 数据库索引 - */ - private Integer database; - - /** - * 主题 - */ - private String topic; -} \ 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/AbstractCacheableDataBridgeExecute.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/action/databridge/AbstractCacheableDataBridgeExecute.java index e7f84dd..f557e7b 100644 --- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/action/databridge/AbstractCacheableDataBridgeExecute.java +++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/action/databridge/AbstractCacheableDataBridgeExecute.java @@ -101,7 +101,7 @@ public abstract class AbstractCacheableDataBridgeExecute 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 { diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/action/databridge/IotRedisStreamDataBridgeExecute.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/action/databridge/IotRedisStreamDataBridgeExecute.java new file mode 100644 index 0000000..a2d4200 --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/action/databridge/IotRedisStreamDataBridgeExecute.java @@ -0,0 +1,81 @@ +package cn.iocoder.yudao.module.iot.service.rule.action.databridge; + +import cn.hutool.core.util.StrUtil; +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 lombok.extern.slf4j.Slf4j; +import org.redisson.Redisson; +import org.redisson.api.RedissonClient; +import org.redisson.config.Config; +import org.redisson.config.SingleServerConfig; +import org.redisson.spring.data.connection.RedissonConnectionFactory; +import org.springframework.data.redis.connection.RedisConnectionFactory; +import org.springframework.data.redis.connection.stream.ObjectRecord; +import org.springframework.data.redis.connection.stream.StreamRecords; +import org.springframework.data.redis.core.RedisTemplate; +import org.springframework.data.redis.serializer.RedisSerializer; +import org.springframework.stereotype.Component; + +/** + * Redis Stream 的 {@link IotDataBridgeExecute} 实现类 + * + * @author HUIHUI + */ +@Component +@Slf4j +public class IotRedisStreamDataBridgeExecute extends + AbstractCacheableDataBridgeExecute> { + + @Override + public Integer getType() { + return IotDataBridgeTypeEnum.REDIS_STREAM.getType(); + } + + @Override + public void execute0(IotDeviceMessage message, IotDataBridgeRedisStreamConfig config) throws Exception { + // 1. 获取 RedisTemplate + RedisTemplate redisTemplate = getProducer(config); + + // 2. 创建并发送 Stream 记录 + ObjectRecord record = StreamRecords.newRecord() + .ofObject(message).withStreamKey(config.getTopic()); + String recordId = String.valueOf(redisTemplate.opsForStream().add(record)); + log.info("[executeRedisStream][消息发送成功] messageId: {}, config: {}", recordId, config); + } + + @Override + protected RedisTemplate initProducer(IotDataBridgeRedisStreamConfig config) { + // 1.1 创建 Redisson 配置 + Config redissonConfig = new Config(); + SingleServerConfig serverConfig = redissonConfig.useSingleServer() + .setAddress("redis://" + config.getHost() + ":" + config.getPort()) + .setDatabase(config.getDatabase()); + // 1.2 设置密码（如果有） + if (StrUtil.isNotBlank(config.getPassword())) { + serverConfig.setPassword(config.getPassword()); + } + + // TODO @芋艿：看看怎么优化 + // 创建 RedisTemplate 并配置 + RedissonClient redisson = Redisson.create(redissonConfig); + RedisTemplate template = new RedisTemplate<>(); + template.setConnectionFactory(new RedissonConnectionFactory(redisson)); + // 设置序列化器 + template.setKeySerializer(RedisSerializer.string()); + template.setHashKeySerializer(RedisSerializer.string()); + template.setValueSerializer(RedisSerializer.json()); + template.setHashValueSerializer(RedisSerializer.json()); + template.afterPropertiesSet(); + return template; + } + + @Override + protected void closeProducer(RedisTemplate producer) throws Exception { + RedisConnectionFactory factory = producer.getConnectionFactory(); + if (factory != null) { + ((RedissonConnectionFactory) factory).destroy(); + } + } + +} diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/action/databridge/IotRedisStreamMQDataBridgeExecute.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/action/databridge/IotRedisStreamMQDataBridgeExecute.java deleted file mode 100644 index 2aac766..0000000 --- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/action/databridge/IotRedisStreamMQDataBridgeExecute.java +++ /dev/null @@ -1,96 +0,0 @@ -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.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; -import org.redisson.config.Config; -import org.redisson.config.SingleServerConfig; -import org.redisson.spring.data.connection.RedissonConnectionFactory; -import org.springframework.data.redis.connection.RedisConnectionFactory; -import org.springframework.data.redis.connection.stream.ObjectRecord; -import org.springframework.data.redis.connection.stream.StreamRecords; -import org.springframework.data.redis.core.RedisTemplate; -import org.springframework.data.redis.serializer.RedisSerializer; -import org.springframework.stereotype.Component; - -/** - * Redis Stream MQ 的 {@link IotDataBridgeExecute} 实现类 - * - * @author HUIHUI - */ -@Component -@Slf4j -public class IotRedisStreamMQDataBridgeExecute extends - AbstractCacheableDataBridgeExecute> { - - @Override - public Integer getType() { - return IotDataBridgeTypeEnum.REDIS_STREAM.getType(); - } - - @Override - public void execute0(IotDeviceMessage message, IotDataBridgeRedisStreamMQConfig config) throws Exception { - // 1. 获取 RedisTemplate - RedisTemplate redisTemplate = getProducer(config); - - // 2. 创建并发送 Stream 记录 - ObjectRecord record = StreamRecords.newRecord() - .ofObject(message).withStreamKey(config.getTopic()); - String recordId = String.valueOf(redisTemplate.opsForStream().add(record)); - log.info("[executeRedisStream][消息发送成功] messageId: {}, config: {}", recordId, config); - } - - @Override - protected RedisTemplate initProducer(IotDataBridgeRedisStreamMQConfig config) { - // 1.1 创建 Redisson 配置 - Config redissonConfig = new Config(); - SingleServerConfig serverConfig = redissonConfig.useSingleServer() - .setAddress("redis://" + config.getHost() + ":" + config.getPort()) - .setDatabase(config.getDatabase()); - // 1.2 设置密码（如果有） - if (StrUtil.isNotBlank(config.getPassword())) { - serverConfig.setPassword(config.getPassword()); - } - - // TODO @huihui：看看能不能简化一些。按道理说，不用这么多的哈。 - // 2.1 创建 RedissonClient - RedissonClient redisson = Redisson.create(redissonConfig); - // 2.2 创建并配置 RedisTemplate - RedisTemplate 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();// 初始化 - return template; - } - - @Override - protected void closeProducer(RedisTemplate producer) throws Exception { - RedisConnectionFactory factory = producer.getConnectionFactory(); - if (factory != null) { - ((RedissonConnectionFactory) factory).destroy(); - } - } - - // TODO @huihui：看看能不能简化一些。按道理说，不用这么多的哈。 - public static RedisSerializer buildRedisSerializer() { - RedisSerializer json = RedisSerializer.json(); - // 解决 LocalDateTime 的序列化 - ObjectMapper objectMapper = (ObjectMapper) ReflectUtil.getFieldValue(json, "mapper"); - objectMapper.registerModules(new JavaTimeModule()); - return json; - } - -} diff --git a/yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/rule/action/databridge/IotDataBridgeExecuteTest.java b/yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/rule/action/databridge/IotDataBridgeExecuteTest.java index faf1358..4a4ca55 100644 --- a/yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/rule/action/databridge/IotDataBridgeExecuteTest.java +++ b/yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/rule/action/databridge/IotDataBridgeExecuteTest.java @@ -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 -- libgit2 0.21.4