From 19cf311b7e476076333ab8981afd093445e5213c Mon Sep 17 00:00:00 2001 From: YunaiV Date: Sat, 14 Jun 2025 20:53:29 +0800 Subject: [PATCH] feat：【IoT 物联网】增加 redis + event-bus 的实现 --- yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/config/YudaoRedisMQConsumerAutoConfiguration.java | 7 +++---- yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/job/RedisPendingMessageResendJob.java | 7 +++---- yudao-module-iot/yudao-module-iot-core/pom.xml | 5 +++++ yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/config/IotMessageBusAutoConfiguration.java | 40 +++++++++++++++++++++++++++++----------- yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/local/IotLocalMessage.java | 14 ++++++++++++++ yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/local/IotLocalMessageBus.java | 67 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/local/LocalIotMessage.java | 14 -------------- yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/local/LocalIotMessageBus.java | 67 ------------------------------------------------------------------- yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/redis/IotRedisMessageBus.java | 92 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/rocketmq/IotRocketMQMessageBus.java | 98 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/rocketmq/RocketMQIotMessageBus.java | 99 --------------------------------------------------------------------------------------------------- yudao-module-iot/yudao-module-iot-core/src/test/java/cn/iocoder/yudao/module/iot/core/messagebus/core/local/LocalIotMessageBusIntegrationTest.java | 2 +- yudao-module-iot/yudao-module-iot-core/src/test/java/cn/iocoder/yudao/module/iot/core/messagebus/core/rocketmq/RocketMQIotMessageBusTest.java | 2 +- yudao-module-iot/yudao-module-iot-gateway/src/main/resources/application-local.yaml | 4 ++-- yudao-module-iot/yudao-module-iot-gateway/src/main/resources/application.yaml | 4 ++-- yudao-server/src/main/resources/application.yaml | 2 +- 16 files changed, 318 insertions(+), 206 deletions(-) create mode 100644 yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/local/IotLocalMessage.java create mode 100644 yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/local/IotLocalMessageBus.java delete mode 100644 yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/local/LocalIotMessage.java delete mode 100644 yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/local/LocalIotMessageBus.java create mode 100644 yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/redis/IotRedisMessageBus.java create mode 100644 yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/rocketmq/IotRocketMQMessageBus.java delete mode 100644 yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/rocketmq/RocketMQIotMessageBus.java diff --git a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/config/YudaoRedisMQConsumerAutoConfiguration.java b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/config/YudaoRedisMQConsumerAutoConfiguration.java index c9ab3e5..b802444 100644 --- a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/config/YudaoRedisMQConsumerAutoConfiguration.java +++ b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/config/YudaoRedisMQConsumerAutoConfiguration.java @@ -69,9 +69,8 @@ public class YudaoRedisMQConsumerAutoConfiguration { @ConditionalOnBean(AbstractRedisStreamMessageListener.class) // 只有 AbstractStreamMessageListener 存在的时候，才需要注册 Redis pubsub 监听 public RedisPendingMessageResendJob redisPendingMessageResendJob(List> listeners, RedisMQTemplate redisTemplate, - @Value("${spring.application.name}") String groupName, RedissonClient redissonClient) { - return new RedisPendingMessageResendJob(listeners, redisTemplate, groupName, redissonClient); + return new RedisPendingMessageResendJob(listeners, redisTemplate, redissonClient); } /** @@ -141,14 +140,14 @@ public class YudaoRedisMQConsumerAutoConfiguration { * * @return 消费者名字 */ - private static String buildConsumerName() { + public static String buildConsumerName() { return String.format("%s@%d", SystemUtil.getHostInfo().getAddress(), SystemUtil.getCurrentPID()); } /** * 校验 Redis 版本号，是否满足最低的版本号要求！ */ - private static void checkRedisVersion(RedisTemplate redisTemplate) { + public static void checkRedisVersion(RedisTemplate redisTemplate) { // 获得 Redis 版本 Properties info = redisTemplate.execute((RedisCallback) RedisServerCommands::info); String version = MapUtil.getStr(info, "redis_version"); diff --git a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/job/RedisPendingMessageResendJob.java b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/job/RedisPendingMessageResendJob.java index cb4e399..bb16be0 100644 --- a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/job/RedisPendingMessageResendJob.java +++ b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/job/RedisPendingMessageResendJob.java @@ -35,7 +35,6 @@ public class RedisPendingMessageResendJob { private final List> listeners; private final RedisMQTemplate redisTemplate; - private final String groupName; private final RedissonClient redissonClient; /** @@ -64,13 +63,13 @@ public class RedisPendingMessageResendJob { private void execute() { StreamOperations ops = redisTemplate.getRedisTemplate().opsForStream(); listeners.forEach(listener -> { - PendingMessagesSummary pendingMessagesSummary = Objects.requireNonNull(ops.pending(listener.getStreamKey(), groupName)); + PendingMessagesSummary pendingMessagesSummary = Objects.requireNonNull(ops.pending(listener.getStreamKey(), listener.getGroup())); // 每个消费者的 pending 队列消息数量 Map pendingMessagesPerConsumer = pendingMessagesSummary.getPendingMessagesPerConsumer(); pendingMessagesPerConsumer.forEach((consumerName, pendingMessageCount) -> { log.info("[processPendingMessage][消费者({}) 消息数量({})]", consumerName, pendingMessageCount); // 每个消费者的 pending消息的详情信息 - PendingMessages pendingMessages = ops.pending(listener.getStreamKey(), Consumer.from(groupName, consumerName), Range.unbounded(), pendingMessageCount); + PendingMessages pendingMessages = ops.pending(listener.getStreamKey(), Consumer.from(listener.getGroup(), consumerName), Range.unbounded(), pendingMessageCount); if (pendingMessages.isEmpty()) { return; } @@ -91,7 +90,7 @@ public class RedisPendingMessageResendJob { .ofObject(records.get(0).getValue()) // 设置内容 .withStreamKey(listener.getStreamKey())); // ack 消息消费完成 - redisTemplate.getRedisTemplate().opsForStream().acknowledge(groupName, records.get(0)); + redisTemplate.getRedisTemplate().opsForStream().acknowledge(listener.getGroup(), records.get(0)); log.info("[processPendingMessage][消息({})重新投递成功]", records.get(0).getId()); }); }); diff --git a/yudao-module-iot/yudao-module-iot-core/pom.xml b/yudao-module-iot/yudao-module-iot-core/pom.xml index 3f4bb1f..30ebc2d 100644 --- a/yudao-module-iot/yudao-module-iot-core/pom.xml +++ b/yudao-module-iot/yudao-module-iot-core/pom.xml @@ -33,6 +33,11 @@ + cn.iocoder.boot + yudao-spring-boot-starter-mq + + + org.springframework.data spring-data-redis true diff --git a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/config/IotMessageBusAutoConfiguration.java b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/config/IotMessageBusAutoConfiguration.java index 6505058..4a5aaff 100644 --- a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/config/IotMessageBusAutoConfiguration.java +++ b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/config/IotMessageBusAutoConfiguration.java @@ -1,8 +1,9 @@ package cn.iocoder.yudao.module.iot.core.messagebus.config; import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus; -import cn.iocoder.yudao.module.iot.core.messagebus.core.local.LocalIotMessageBus; -import cn.iocoder.yudao.module.iot.core.messagebus.core.rocketmq.RocketMQIotMessageBus; +import cn.iocoder.yudao.module.iot.core.messagebus.core.local.IotLocalMessageBus; +import cn.iocoder.yudao.module.iot.core.messagebus.core.redis.IotRedisMessageBus; +import cn.iocoder.yudao.module.iot.core.messagebus.core.rocketmq.IotRocketMQMessageBus; import cn.iocoder.yudao.module.iot.core.mq.producer.IotDeviceMessageProducer; import lombok.extern.slf4j.Slf4j; import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties; @@ -14,6 +15,8 @@ import org.springframework.boot.context.properties.EnableConfigurationProperties import org.springframework.context.ApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; +import org.springframework.data.redis.core.RedisTemplate; +import org.springframework.data.redis.core.StringRedisTemplate; /** * IoT 消息总线自动配置 @@ -34,12 +37,12 @@ public class IotMessageBusAutoConfiguration { @Configuration @ConditionalOnProperty(prefix = "yudao.iot.message-bus", name = "type", havingValue = "local", matchIfMissing = true) - public static class LocalIotMessageBusConfiguration { + public static class IotLocalMessageBusConfiguration { @Bean - public IotMessageBus localIotMessageBus(ApplicationContext applicationContext) { - log.info("[localIotMessageBus][创建 Local IoT 消息总线]"); - return new LocalIotMessageBus(applicationContext); + public IotMessageBus iotLocalMessageBus(ApplicationContext applicationContext) { + log.info("[iotLocalMessageBus][创建 IoT Local 消息总线]"); + return new IotLocalMessageBus(applicationContext); } } @@ -49,13 +52,28 @@ public class IotMessageBusAutoConfiguration { @Configuration @ConditionalOnProperty(prefix = "yudao.iot.message-bus", name = "type", havingValue = "rocketmq") @ConditionalOnClass(RocketMQTemplate.class) - public static class RocketMQIotMessageBusConfiguration { + public static class IotRocketMQMessageBusConfiguration { @Bean - @SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection") - public IotMessageBus rocketMQIotMessageBus(RocketMQProperties rocketMQProperties, RocketMQTemplate rocketMQTemplate) { - log.info("[rocketMQIotMessageBus][创建 RocketMQ IoT 消息总线]"); - return new RocketMQIotMessageBus(rocketMQProperties, rocketMQTemplate); + public IotMessageBus iotRocketMQMessageBus(RocketMQProperties rocketMQProperties, + RocketMQTemplate rocketMQTemplate) { + log.info("[iotRocketMQMessageBus][创建 IoT RocketMQ 消息总线]"); + return new IotRocketMQMessageBus(rocketMQProperties, rocketMQTemplate); + } + + } + + // ==================== Redis 实现 ==================== + + @Configuration + @ConditionalOnProperty(prefix = "yudao.iot.message-bus", name = "type", havingValue = "redis") + @ConditionalOnClass(RedisTemplate.class) + public static class IotRedisMessageBusConfiguration { + + @Bean + public IotMessageBus iotRedisMessageBus(StringRedisTemplate redisTemplate) { + log.info("[iotRedisMessageBus][创建 IoT Redis 消息总线]"); + return new IotRedisMessageBus(redisTemplate); } } diff --git a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/local/IotLocalMessage.java b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/local/IotLocalMessage.java new file mode 100644 index 0000000..5a9841a --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/local/IotLocalMessage.java @@ -0,0 +1,14 @@ +package cn.iocoder.yudao.module.iot.core.messagebus.core.local; + +import lombok.AllArgsConstructor; +import lombok.Data; + +@Data +@AllArgsConstructor +public class IotLocalMessage { + + private String topic; + + private Object message; + +} diff --git a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/local/IotLocalMessageBus.java b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/local/IotLocalMessageBus.java new file mode 100644 index 0000000..1fc608b --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/local/IotLocalMessageBus.java @@ -0,0 +1,67 @@ +package cn.iocoder.yudao.module.iot.core.messagebus.core.local; + +import cn.hutool.core.collection.CollUtil; +import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus; +import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageSubscriber; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.context.ApplicationContext; +import org.springframework.context.event.EventListener; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +/** + * 本地的 {@link IotMessageBus} 实现类 + * + * 注意：仅适用于单机场景！！！ + * + * @author 芋道源码 + */ +@RequiredArgsConstructor +@Slf4j +public class IotLocalMessageBus implements IotMessageBus { + + private final ApplicationContext applicationContext; + + /** + * 订阅者映射表 + * Key: topic + */ + private final Map>> subscribers = new HashMap<>(); + + @Override + public void post(String topic, Object message) { + applicationContext.publishEvent(new IotLocalMessage(topic, message)); + } + + @Override + public void register(IotMessageSubscriber subscriber) { + String topic = subscriber.getTopic(); + List> topicSubscribers = subscribers.computeIfAbsent(topic, k -> new ArrayList<>()); + topicSubscribers.add(subscriber); + log.info("[register][topic({}/{}) 注册消费者({})成功]", + topic, subscriber.getGroup(), subscriber.getClass().getName()); + } + + @EventListener + @SuppressWarnings({"unchecked", "rawtypes"}) + public void onMessage(IotLocalMessage message) { + String topic = message.getTopic(); + List> topicSubscribers = subscribers.get(topic); + if (CollUtil.isEmpty(topicSubscribers)) { + return; + } + for (IotMessageSubscriber subscriber : topicSubscribers) { + try { + subscriber.onMessage(message.getMessage()); + } catch (Exception ex) { + log.error("[onMessage][topic({}/{}) message({}) 消费者({}) 处理异常]", + subscriber.getTopic(), subscriber.getGroup(), message.getMessage(), subscriber.getClass().getName(), ex); + } + } + } + +} \ No newline at end of file diff --git a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/local/LocalIotMessage.java b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/local/LocalIotMessage.java deleted file mode 100644 index c8c7277..0000000 --- a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/local/LocalIotMessage.java +++ /dev/null @@ -1,14 +0,0 @@ -package cn.iocoder.yudao.module.iot.core.messagebus.core.local; - -import lombok.AllArgsConstructor; -import lombok.Data; - -@Data -@AllArgsConstructor -public class LocalIotMessage { - - private String topic; - - private Object message; - -} diff --git a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/local/LocalIotMessageBus.java b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/local/LocalIotMessageBus.java deleted file mode 100644 index 76bd6a4..0000000 --- a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/local/LocalIotMessageBus.java +++ /dev/null @@ -1,67 +0,0 @@ -package cn.iocoder.yudao.module.iot.core.messagebus.core.local; - -import cn.hutool.core.collection.CollUtil; -import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus; -import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageSubscriber; -import lombok.RequiredArgsConstructor; -import lombok.extern.slf4j.Slf4j; -import org.springframework.context.ApplicationContext; -import org.springframework.context.event.EventListener; - -import java.util.ArrayList; -import java.util.HashMap; -import java.util.List; -import java.util.Map; - -/** - * 本地的 {@link IotMessageBus} 实现类 - * - * 注意：仅适用于单机场景！！！ - * - * @author 芋道源码 - */ -@RequiredArgsConstructor -@Slf4j -public class LocalIotMessageBus implements IotMessageBus { - - private final ApplicationContext applicationContext; - - /** - * 订阅者映射表 - * Key: topic - */ - private final Map>> subscribers = new HashMap<>(); - - @Override - public void post(String topic, Object message) { - applicationContext.publishEvent(new LocalIotMessage(topic, message)); - } - - @Override - public void register(IotMessageSubscriber subscriber) { - String topic = subscriber.getTopic(); - List> topicSubscribers = subscribers.computeIfAbsent(topic, k -> new ArrayList<>()); - topicSubscribers.add(subscriber); - log.info("[register][topic({}/{}) 注册消费者({})成功]", - topic, subscriber.getGroup(), subscriber.getClass().getName()); - } - - @EventListener - @SuppressWarnings({"unchecked", "rawtypes"}) - public void onMessage(LocalIotMessage message) { - String topic = message.getTopic(); - List> topicSubscribers = subscribers.get(topic); - if (CollUtil.isEmpty(topicSubscribers)) { - return; - } - for (IotMessageSubscriber subscriber : topicSubscribers) { - try { - subscriber.onMessage(message.getMessage()); - } catch (Exception ex) { - log.error("[onMessage][topic({}/{}) message({}) 消费者({}) 处理异常]", - subscriber.getTopic(), subscriber.getGroup(), message.getMessage(), subscriber.getClass().getName(), ex); - } - } - } - -} \ No newline at end of file diff --git a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/redis/IotRedisMessageBus.java b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/redis/IotRedisMessageBus.java new file mode 100644 index 0000000..5736345 --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/redis/IotRedisMessageBus.java @@ -0,0 +1,92 @@ +package cn.iocoder.yudao.module.iot.core.messagebus.core.redis; + +import cn.hutool.core.util.TypeUtil; +import cn.iocoder.yudao.framework.common.util.json.JsonUtils; +import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus; +import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageSubscriber; +import jakarta.annotation.PostConstruct; +import jakarta.annotation.PreDestroy; +import lombok.extern.slf4j.Slf4j; +import org.springframework.data.redis.connection.stream.*; +import org.springframework.data.redis.core.RedisTemplate; +import org.springframework.data.redis.stream.StreamMessageListenerContainer; + +import java.lang.reflect.Type; + +import static cn.iocoder.yudao.framework.mq.redis.config.YudaoRedisMQConsumerAutoConfiguration.buildConsumerName; +import static cn.iocoder.yudao.framework.mq.redis.config.YudaoRedisMQConsumerAutoConfiguration.checkRedisVersion; + +/** + * Redis 的 {@link IotMessageBus} 实现类 + * + * @author 芋道源码 + */ +@Slf4j +public class IotRedisMessageBus implements IotMessageBus { + + private final RedisTemplate redisTemplate; + + private final StreamMessageListenerContainer> redisStreamMessageListenerContainer; + + public IotRedisMessageBus(RedisTemplate redisTemplate) { + this.redisTemplate = redisTemplate; + checkRedisVersion(redisTemplate); + // 创建 options 配置 + StreamMessageListenerContainer.StreamMessageListenerContainerOptions> containerOptions = + StreamMessageListenerContainer.StreamMessageListenerContainerOptions.builder() + .batchSize(10) // 一次性最多拉取多少条消息 + .targetType(String.class) // 目标类型。统一使用 String，通过自己封装的 AbstractStreamMessageListener 去反序列化 + .build(); + // 创建 container 对象 + this.redisStreamMessageListenerContainer = + StreamMessageListenerContainer.create(redisTemplate.getRequiredConnectionFactory(), containerOptions); + } + + @PostConstruct + public void init() { + this.redisStreamMessageListenerContainer.start(); + } + + @PreDestroy + public void destroy() { + this.redisStreamMessageListenerContainer.stop(); + } + + @Override + public void post(String topic, Object message) { + redisTemplate.opsForStream().add(StreamRecords.newRecord() + .ofObject(JsonUtils.toJsonString(message)) // 设置内容 + .withStreamKey(topic)); // 设置 stream key + } + + @Override + public void register(IotMessageSubscriber subscriber) { + Type type = TypeUtil.getTypeArgument(subscriber.getClass(), 0); + if (type == null) { + throw new IllegalStateException(String.format("类型(%s) 需要设置消息类型", getClass().getName())); + } + + // 创建 listener 对应的消费者分组 + try { + redisTemplate.opsForStream().createGroup(subscriber.getTopic(), subscriber.getGroup()); + } catch (Exception ignore) { + } + // 创建 Consumer 对象 + String consumerName = buildConsumerName(); + Consumer consumer = Consumer.from(subscriber.getGroup(), consumerName); + // 设置 Consumer 消费进度，以最小消费进度为准 + StreamOffset streamOffset = StreamOffset.create(subscriber.getTopic(), ReadOffset.lastConsumed()); + // 设置 Consumer 监听 + StreamMessageListenerContainer.StreamReadRequestBuilder builder = StreamMessageListenerContainer.StreamReadRequest + .builder(streamOffset).consumer(consumer) + .autoAcknowledge(false) // 不自动 ack + .cancelOnError(throwable -> false); // 默认配置，发生异常就取消消费，显然不符合预期；因此，我们设置为 false + redisStreamMessageListenerContainer.register(builder.build(), message -> { + // 消费消息 + subscriber.onMessage(JsonUtils.parseObject(message.getValue(), type)); + // ack 消息消费完成 + redisTemplate.opsForStream().acknowledge(subscriber.getGroup(), message); + }); + } + +} diff --git a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/rocketmq/IotRocketMQMessageBus.java b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/rocketmq/IotRocketMQMessageBus.java new file mode 100644 index 0000000..48218b2 --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/rocketmq/IotRocketMQMessageBus.java @@ -0,0 +1,98 @@ +package cn.iocoder.yudao.module.iot.core.messagebus.core.rocketmq; + +import cn.hutool.core.util.TypeUtil; +import cn.iocoder.yudao.framework.common.util.json.JsonUtils; +import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus; +import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageSubscriber; +import jakarta.annotation.PreDestroy; +import lombok.RequiredArgsConstructor; +import lombok.SneakyThrows; +import lombok.extern.slf4j.Slf4j; +import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; +import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; +import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently; +import org.apache.rocketmq.client.producer.SendResult; +import org.apache.rocketmq.common.message.MessageExt; +import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties; +import org.apache.rocketmq.spring.core.RocketMQTemplate; + +import java.lang.reflect.Type; +import java.util.ArrayList; +import java.util.List; + +/** + * 基于 RocketMQ 的 {@link IotMessageBus} 实现类 + * + * @author 芋道源码 + */ +@RequiredArgsConstructor +@Slf4j +public class IotRocketMQMessageBus implements IotMessageBus { + + private final RocketMQProperties rocketMQProperties; + + private final RocketMQTemplate rocketMQTemplate; + + /** + * 主题对应的消费者映射 + */ + private final List topicConsumers = new ArrayList<>(); + + /** + * 销毁时关闭所有消费者 + */ + @PreDestroy + public void destroy() { + for (DefaultMQPushConsumer consumer : topicConsumers) { + try { + consumer.shutdown(); + log.info("[destroy][关闭 group({}) 的消费者成功]", consumer.getConsumerGroup()); + } catch (Exception e) { + log.error("[destroy]关闭 group({}) 的消费者异常]", consumer.getConsumerGroup(), e); + } + } + } + + @Override + public void post(String topic, Object message) { + // TODO @芋艿：需要 orderly！ + SendResult result = rocketMQTemplate.syncSend(topic, JsonUtils.toJsonString(message)); + log.info("[post][topic({}) 发送消息({}) result({})]", topic, message, result); + } + + @Override + @SneakyThrows + public void register(IotMessageSubscriber subscriber) { + Type type = TypeUtil.getTypeArgument(subscriber.getClass(), 0); + if (type == null) { + throw new IllegalStateException(String.format("类型(%s) 需要设置消息类型", getClass().getName())); + } + + // 1.1 创建 DefaultMQPushConsumer + DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(); + consumer.setNamesrvAddr(rocketMQProperties.getNameServer()); + consumer.setConsumerGroup(subscriber.getGroup()); + // 1.2 订阅主题 + consumer.subscribe(subscriber.getTopic(), "*"); + // 1.3 设置消息监听器 + consumer.setMessageListener((MessageListenerConcurrently) (messages, context) -> { + for (MessageExt messageExt : messages) { + try { + byte[] body = messageExt.getBody(); + subscriber.onMessage(JsonUtils.parseObject(body, type)); + } catch (Exception ex) { + log.error("[onMessage][topic({}/{}) message({}) 消费者({}) 处理异常]", + subscriber.getTopic(), subscriber.getGroup(), messageExt, subscriber.getClass().getName(), ex); + return ConsumeConcurrentlyStatus.RECONSUME_LATER; + } + } + return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; + }); + // 1.4 启动消费者 + consumer.start(); + + // 2. 保存消费者引用 + topicConsumers.add(consumer); + } + +} \ No newline at end of file diff --git a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/rocketmq/RocketMQIotMessageBus.java b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/rocketmq/RocketMQIotMessageBus.java deleted file mode 100644 index 5d6d72a..0000000 --- a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/rocketmq/RocketMQIotMessageBus.java +++ /dev/null @@ -1,99 +0,0 @@ -package cn.iocoder.yudao.module.iot.core.messagebus.core.rocketmq; - -import cn.hutool.core.util.TypeUtil; -import cn.iocoder.yudao.framework.common.util.json.JsonUtils; -import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus; -import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageSubscriber; -import lombok.RequiredArgsConstructor; -import lombok.SneakyThrows; -import lombok.extern.slf4j.Slf4j; -import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; -import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; -import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently; -import org.apache.rocketmq.client.producer.SendResult; -import org.apache.rocketmq.common.message.MessageExt; -import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties; -import org.apache.rocketmq.spring.core.RocketMQTemplate; - -import jakarta.annotation.PreDestroy; - -import java.lang.reflect.Type; -import java.util.ArrayList; -import java.util.List; - -/** - * 基于 RocketMQ 的 {@link IotMessageBus} 实现类 - * - * @author 芋道源码 - */ -@RequiredArgsConstructor -@Slf4j -public class RocketMQIotMessageBus implements IotMessageBus { - - private final RocketMQProperties rocketMQProperties; - - private final RocketMQTemplate rocketMQTemplate; - - /** - * 主题对应的消费者映射 - */ - private final List topicConsumers = new ArrayList<>(); - - @Override - public void post(String topic, Object message) { - // TODO @芋艿：需要 orderly！ - SendResult result = rocketMQTemplate.syncSend(topic, JsonUtils.toJsonString(message)); - log.info("[post][topic({}) 发送消息({}) result({})]", topic, message, result); - } - - @Override - @SneakyThrows - public void register(IotMessageSubscriber subscriber) { - Type type = TypeUtil.getTypeArgument(subscriber.getClass(), 0); - if (type == null) { - throw new IllegalStateException(String.format("类型(%s) 需要设置消息类型", getClass().getName())); - } - - // 1.1 创建 DefaultMQPushConsumer - DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(); - consumer.setNamesrvAddr(rocketMQProperties.getNameServer()); - consumer.setConsumerGroup(subscriber.getGroup()); - // 1.2 订阅主题 - consumer.subscribe(subscriber.getTopic(), "*"); - // 1.3 设置消息监听器 - consumer.setMessageListener((MessageListenerConcurrently) (messages, context) -> { - for (MessageExt messageExt : messages) { - try { - byte[] body = messageExt.getBody(); - subscriber.onMessage(JsonUtils.parseObject(body, type)); - } catch (Exception ex) { - log.error("[onMessage][topic({}/{}) message({}) 消费者({}) 处理异常]", - subscriber.getTopic(), subscriber.getGroup(), messageExt, subscriber.getClass().getName(), ex); - return ConsumeConcurrentlyStatus.RECONSUME_LATER; - } - } - return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; - }); - // 1.4 启动消费者 - consumer.start(); - - // 2. 保存消费者引用 - topicConsumers.add(consumer); - } - - /** - * 销毁时关闭所有消费者 - */ - @PreDestroy - public void destroy() { - for (DefaultMQPushConsumer consumer : topicConsumers) { - try { - consumer.shutdown(); - log.info("[destroy][关闭 group({}) 的消费者成功]", consumer.getConsumerGroup()); - } catch (Exception e) { - log.error("[destroy]关闭 group({}) 的消费者异常]", consumer.getConsumerGroup(), e); - } - } - } - -} \ No newline at end of file diff --git a/yudao-module-iot/yudao-module-iot-core/src/test/java/cn/iocoder/yudao/module/iot/core/messagebus/core/local/LocalIotMessageBusIntegrationTest.java b/yudao-module-iot/yudao-module-iot-core/src/test/java/cn/iocoder/yudao/module/iot/core/messagebus/core/local/LocalIotMessageBusIntegrationTest.java index 341ad89..b282bc8 100644 --- a/yudao-module-iot/yudao-module-iot-core/src/test/java/cn/iocoder/yudao/module/iot/core/messagebus/core/local/LocalIotMessageBusIntegrationTest.java +++ b/yudao-module-iot/yudao-module-iot-core/src/test/java/cn/iocoder/yudao/module/iot/core/messagebus/core/local/LocalIotMessageBusIntegrationTest.java @@ -17,7 +17,7 @@ import java.util.concurrent.atomic.AtomicInteger; import static org.junit.jupiter.api.Assertions.*; /** - * {@link LocalIotMessageBus} 集成测试 + * {@link IotLocalMessageBus} 集成测试 * * @author 芋道源码 */ diff --git a/yudao-module-iot/yudao-module-iot-core/src/test/java/cn/iocoder/yudao/module/iot/core/messagebus/core/rocketmq/RocketMQIotMessageBusTest.java b/yudao-module-iot/yudao-module-iot-core/src/test/java/cn/iocoder/yudao/module/iot/core/messagebus/core/rocketmq/RocketMQIotMessageBusTest.java index 01b97ce..b7270f2 100644 --- a/yudao-module-iot/yudao-module-iot-core/src/test/java/cn/iocoder/yudao/module/iot/core/messagebus/core/rocketmq/RocketMQIotMessageBusTest.java +++ b/yudao-module-iot/yudao-module-iot-core/src/test/java/cn/iocoder/yudao/module/iot/core/messagebus/core/rocketmq/RocketMQIotMessageBusTest.java @@ -21,7 +21,7 @@ import java.util.concurrent.atomic.AtomicReference; import static org.junit.jupiter.api.Assertions.*; /** - * {@link RocketMQIotMessageBus} 集成测试 + * {@link IotRocketMQMessageBus} 集成测试 * * @author 芋道源码 */ diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/resources/application-local.yaml b/yudao-module-iot/yudao-module-iot-gateway/src/main/resources/application-local.yaml index 384799e..ab3eda8 100644 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/resources/application-local.yaml +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/resources/application-local.yaml @@ -31,7 +31,7 @@ yudao: # 针对引入的 EMQX 组件的配置 # ==================================== emqx: - enabled: true + enabled: false http-port: 8090 # MQTT HTTP 服务端口 mqtt-host: 127.0.0.1 # MQTT Broker 地址 mqtt-port: 1883 # MQTT Broker 端口 @@ -44,7 +44,7 @@ yudao: # 消息总线配置 message-bus: - type: rocketmq # 本地开发使用 RocketMQ + type: redis # 本地开发使用 RocketMQ --- #################### 日志相关配置 #################### diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/resources/application.yaml b/yudao-module-iot/yudao-module-iot-gateway/src/main/resources/application.yaml index ae42202..b12b2f7 100644 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/resources/application.yaml +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/resources/application.yaml @@ -37,14 +37,14 @@ yudao: # 针对引入的 EMQX 组件的配置 # ==================================== emqx: - enabled: true + enabled: false mqtt-ssl: false mqtt-topics: - "/sys/#" # 系统主题 # 消息总线配置 message-bus: - type: rocketmq # 消息总线的类型 + type: redis # 消息总线的类型 --- #################### 日志相关配置 #################### diff --git a/yudao-server/src/main/resources/application.yaml b/yudao-server/src/main/resources/application.yaml index 1fcb2df..025fbbe 100644 --- a/yudao-server/src/main/resources/application.yaml +++ b/yudao-server/src/main/resources/application.yaml @@ -313,6 +313,6 @@ yudao: customer: E77DF18BE109F454A5CD319E44BF5177 iot: message-bus: - type: rocketmq # 消息总线的类型 + type: redis # 消息总线的类型 debug: false \ No newline at end of file -- libgit2 0.21.4