From 385cea8d9069e71b3d4e2a109e56afb39849f420 Mon Sep 17 00:00:00 2001 From: YunaiV Date: Thu, 29 May 2025 07:33:34 +0800 Subject: [PATCH] reactor：【IoT 物联网】合并 messagebus 和 common 包，统一到 core 包 --- yudao-module-iot/yudao-module-iot-core/pom.xml | 56 +++++++++++++++++++++++++++++++++++++++++++++++++------- yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/biz/IotDeviceCommonApi.java | 5 +++++ yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/IotDeviceMessageIdentifierEnum.java | 45 +++++++++++++++++++++++++++++++++++++++++++++ yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/IotDeviceMessageTypeEnum.java | 37 +++++++++++++++++++++++++++++++++++++ yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/config/IotMessageBusAutoConfiguration.java | 63 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/config/IotMessageBusProperties.java | 28 ++++++++++++++++++++++++++++ yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/IotMessageBus.java | 27 +++++++++++++++++++++++++++ yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/IotMessageBusSubscriber.java | 29 +++++++++++++++++++++++++++++ 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/rocketmq/RocketMQIotMessageBus.java | 98 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/mq/message/IotDeviceMessage.java | 77 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/mq/producer/IotDeviceMessageProducer.java | 48 ++++++++++++++++++++++++++++++++++++++++++++++++ yudao-module-iot/yudao-module-iot-core/src/main/resources/META-INF/spring.factories | 2 ++ yudao-module-iot/yudao-module-iot-core/src/test/java/cn/iocoder/yudao/module/iot/core/messagebus/core/TestMessage.java | 12 ++++++++++++ yudao-module-iot/yudao-module-iot-core/src/test/java/cn/iocoder/yudao/module/iot/core/messagebus/core/local/LocalIotMessageBusIntegrationTest.java | 177 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-module-iot/yudao-module-iot-core/src/test/java/cn/iocoder/yudao/module/iot/core/messagebus/core/rocketmq/RocketMQIotMessageBusTest.java | 268 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-module-iot/yudao-module-iot-core/yudao-module-iot-common/pom.xml | 26 -------------------------- yudao-module-iot/yudao-module-iot-core/yudao-module-iot-common/src/main/java/cn/iocoder/yudao/module/iot/common/biz/IotDeviceCommonApi.java | 5 ----- yudao-module-iot/yudao-module-iot-core/yudao-module-iot-common/src/main/java/cn/iocoder/yudao/module/iot/common/enums/IotCommonConstants.java | 22 ---------------------- yudao-module-iot/yudao-module-iot-core/yudao-module-iot-common/src/main/java/cn/iocoder/yudao/module/iot/common/enums/IotDeviceMessageIdentifierEnum.java | 45 --------------------------------------------- yudao-module-iot/yudao-module-iot-core/yudao-module-iot-common/src/main/java/cn/iocoder/yudao/module/iot/common/enums/IotDeviceMessageTypeEnum.java | 37 ------------------------------------- yudao-module-iot/yudao-module-iot-core/yudao-module-iot-common/src/main/java/cn/iocoder/yudao/module/iot/common/message/IotDeviceMessage.java | 77 ----------------------------------------------------------------------------- yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/pom.xml | 65 ----------------------------------------------------------------- yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/src/main/java/cn/iocoder/yudao/module/iot/messagebus/config/IotMessageBusAutoConfiguration.java | 56 -------------------------------------------------------- yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/src/main/java/cn/iocoder/yudao/module/iot/messagebus/config/IotMessageBusProperties.java | 28 ---------------------------- yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/src/main/java/cn/iocoder/yudao/module/iot/messagebus/core/IotMessageBus.java | 27 --------------------------- yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/src/main/java/cn/iocoder/yudao/module/iot/messagebus/core/IotMessageBusSubscriber.java | 29 ----------------------------- yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/src/main/java/cn/iocoder/yudao/module/iot/messagebus/core/local/LocalIotMessage.java | 14 -------------- yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/src/main/java/cn/iocoder/yudao/module/iot/messagebus/core/local/LocalIotMessageBus.java | 69 --------------------------------------------------------------------- yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/src/main/java/cn/iocoder/yudao/module/iot/messagebus/core/rocketmq/RocketMQIotMessageBus.java | 98 -------------------------------------------------------------------------------------------------- yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/src/main/resources/META-INF/spring.factories | 2 -- yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/src/test/java/cn/iocoder/yudao/module/iot/messagebus/core/TestMessage.java | 12 ------------ yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/src/test/java/cn/iocoder/yudao/module/iot/messagebus/core/local/LocalIotMessageBusIntegrationTest.java | 177 --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/src/test/java/cn/iocoder/yudao/module/iot/messagebus/core/rocketmq/RocketMQIotMessageBusTest.java | 271 ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- 35 files changed, 1046 insertions(+), 1067 deletions(-) create mode 100644 yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/biz/IotDeviceCommonApi.java create mode 100644 yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/IotDeviceMessageIdentifierEnum.java create mode 100644 yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/IotDeviceMessageTypeEnum.java create mode 100644 yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/config/IotMessageBusAutoConfiguration.java create mode 100644 yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/config/IotMessageBusProperties.java create mode 100644 yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/IotMessageBus.java create mode 100644 yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/IotMessageBusSubscriber.java create mode 100644 yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/local/LocalIotMessage.java create 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/rocketmq/RocketMQIotMessageBus.java create mode 100644 yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/mq/message/IotDeviceMessage.java create mode 100644 yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/mq/producer/IotDeviceMessageProducer.java create mode 100644 yudao-module-iot/yudao-module-iot-core/src/main/resources/META-INF/spring.factories create mode 100644 yudao-module-iot/yudao-module-iot-core/src/test/java/cn/iocoder/yudao/module/iot/core/messagebus/core/TestMessage.java create mode 100644 yudao-module-iot/yudao-module-iot-core/src/test/java/cn/iocoder/yudao/module/iot/core/messagebus/core/local/LocalIotMessageBusIntegrationTest.java create mode 100644 yudao-module-iot/yudao-module-iot-core/src/test/java/cn/iocoder/yudao/module/iot/core/messagebus/core/rocketmq/RocketMQIotMessageBusTest.java delete mode 100644 yudao-module-iot/yudao-module-iot-core/yudao-module-iot-common/pom.xml delete mode 100644 yudao-module-iot/yudao-module-iot-core/yudao-module-iot-common/src/main/java/cn/iocoder/yudao/module/iot/common/biz/IotDeviceCommonApi.java delete mode 100644 yudao-module-iot/yudao-module-iot-core/yudao-module-iot-common/src/main/java/cn/iocoder/yudao/module/iot/common/enums/IotCommonConstants.java delete mode 100644 yudao-module-iot/yudao-module-iot-core/yudao-module-iot-common/src/main/java/cn/iocoder/yudao/module/iot/common/enums/IotDeviceMessageIdentifierEnum.java delete mode 100644 yudao-module-iot/yudao-module-iot-core/yudao-module-iot-common/src/main/java/cn/iocoder/yudao/module/iot/common/enums/IotDeviceMessageTypeEnum.java delete mode 100644 yudao-module-iot/yudao-module-iot-core/yudao-module-iot-common/src/main/java/cn/iocoder/yudao/module/iot/common/message/IotDeviceMessage.java delete mode 100644 yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/pom.xml delete mode 100644 yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/src/main/java/cn/iocoder/yudao/module/iot/messagebus/config/IotMessageBusAutoConfiguration.java delete mode 100644 yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/src/main/java/cn/iocoder/yudao/module/iot/messagebus/config/IotMessageBusProperties.java delete mode 100644 yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/src/main/java/cn/iocoder/yudao/module/iot/messagebus/core/IotMessageBus.java delete mode 100644 yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/src/main/java/cn/iocoder/yudao/module/iot/messagebus/core/IotMessageBusSubscriber.java delete mode 100644 yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/src/main/java/cn/iocoder/yudao/module/iot/messagebus/core/local/LocalIotMessage.java delete mode 100644 yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/src/main/java/cn/iocoder/yudao/module/iot/messagebus/core/local/LocalIotMessageBus.java delete mode 100644 yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/src/main/java/cn/iocoder/yudao/module/iot/messagebus/core/rocketmq/RocketMQIotMessageBus.java delete mode 100644 yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/src/main/resources/META-INF/spring.factories delete mode 100644 yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/src/test/java/cn/iocoder/yudao/module/iot/messagebus/core/TestMessage.java delete mode 100644 yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/src/test/java/cn/iocoder/yudao/module/iot/messagebus/core/local/LocalIotMessageBusIntegrationTest.java delete mode 100644 yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/src/test/java/cn/iocoder/yudao/module/iot/messagebus/core/rocketmq/RocketMQIotMessageBusTest.java diff --git a/yudao-module-iot/yudao-module-iot-core/pom.xml b/yudao-module-iot/yudao-module-iot-core/pom.xml index 3417fb0..3f4bb1f 100644 --- a/yudao-module-iot/yudao-module-iot-core/pom.xml +++ b/yudao-module-iot/yudao-module-iot-core/pom.xml @@ -7,19 +7,61 @@ cn.iocoder.boot ${revision} - - yudao-module-iot-common - yudao-module-iot-message-bus - 4.0.0 yudao-module-iot-core - pom + jar ${project.artifactId} - iot 模块下，提供 iot-biz 和 iot-gateway 模块的核心功能。 - 例如说：消息总线、消息协议（编解码）等。 + iot 模块下，提供 iot-biz 和 iot-gateway 模块的核心功能。例如说： + 1. 消息总线：跨 iot-biz 和 iot-gateway 的设备消息。可选择使用 spring event、redis stream、rocketmq、kafka、rabbitmq 等。 + 2. 查询设备信息的通用 API + + + cn.iocoder.boot + yudao-common + + + + + org.springframework.boot + spring-boot-starter + + + + + org.springframework.data + spring-data-redis + true + + + + org.apache.rocketmq + rocketmq-spring-boot-starter + true + + + + org.springframework.amqp + spring-rabbit + true + + + + org.springframework.kafka + spring-kafka + true + + + + + org.springframework.boot + spring-boot-starter-test + test + + + \ 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/biz/IotDeviceCommonApi.java b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/biz/IotDeviceCommonApi.java new file mode 100644 index 0000000..c3a57e5 --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/biz/IotDeviceCommonApi.java @@ -0,0 +1,5 @@ +package cn.iocoder.yudao.module.iot.core.biz; + +// TODO @芋艿：待实现 +public interface IotDeviceCommonApi { +} diff --git a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/IotDeviceMessageIdentifierEnum.java b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/IotDeviceMessageIdentifierEnum.java new file mode 100644 index 0000000..ae9c9de --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/IotDeviceMessageIdentifierEnum.java @@ -0,0 +1,45 @@ +package cn.iocoder.yudao.module.iot.core.enums; + +import lombok.Getter; +import lombok.RequiredArgsConstructor; + +// TODO @芋艿：需要添加对应的 DTO，以及上下行的链路，网关、网关服务、设备等 + +/** + * IoT 设备消息标识符枚举 + */ +@Getter +@RequiredArgsConstructor +public enum IotDeviceMessageIdentifierEnum { + + PROPERTY_GET("get"), // 下行 TODO 芋艿：【讨论】貌似这个“上行”更合理？device 主动拉取配置。和 IotDevicePropertyGetReqDTO 一样的配置 + PROPERTY_SET("set"), // 下行 + PROPERTY_REPORT("report"), // 上行 + + STATE_ONLINE("online"), // 上行 + STATE_OFFLINE("offline"), // 上行 + + CONFIG_GET("get"), // 上行 TODO 芋艿：【讨论】暂时没有上行的场景 + CONFIG_SET("set"), // 下行 + + SERVICE_INVOKE("${identifier}"), // 下行 + SERVICE_REPLY_SUFFIX("_reply"), // 芋艿：TODO 芋艿：【讨论】上行 or 下行 + + OTA_UPGRADE("upgrade"), // 下行 + OTA_PULL("pull"), // 上行 + OTA_PROGRESS("progress"), // 上行 + OTA_REPORT("report"), // 上行 + + REGISTER_REGISTER("register"), // 上行 + REGISTER_REGISTER_SUB("register_sub"), // 上行 + REGISTER_UNREGISTER_SUB("unregister_sub"), // 下行 + + TOPOLOGY_ADD("topology_add"), // 下行; + ; + + /** + * 标志符 + */ + private final String identifier; + +} \ 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/enums/IotDeviceMessageTypeEnum.java b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/IotDeviceMessageTypeEnum.java new file mode 100644 index 0000000..6e0feb1 --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/IotDeviceMessageTypeEnum.java @@ -0,0 +1,37 @@ +package cn.iocoder.yudao.module.iot.core.enums; + +import cn.iocoder.yudao.framework.common.core.ArrayValuable; +import lombok.Getter; +import lombok.RequiredArgsConstructor; + +import java.util.Arrays; + +/** + * IoT 设备消息类型枚举 + */ +@Getter +@RequiredArgsConstructor +public enum IotDeviceMessageTypeEnum implements ArrayValuable { + + STATE("state"), // 设备状态 + PROPERTY("property"), // 设备属性：可参考 https://help.aliyun.com/zh/iot/user-guide/device-properties-events-and-services 设备属性、事件、服务 + EVENT("event"), // 设备事件：可参考 https://help.aliyun.com/zh/iot/user-guide/device-properties-events-and-services 设备属性、事件、服务 + SERVICE("service"), // 设备服务：可参考 https://help.aliyun.com/zh/iot/user-guide/device-properties-events-and-services 设备属性、事件、服务 + CONFIG("config"), // 设备配置：可参考 https://help.aliyun.com/zh/iot/user-guide/remote-configuration-1 远程配置 + OTA("ota"), // 设备 OTA：可参考 https://help.aliyun.com/zh/iot/user-guide/ota-update OTA 升级 + REGISTER("register"), // 设备注册：可参考 https://help.aliyun.com/zh/iot/user-guide/register-devices 设备身份注册 + TOPOLOGY("topology"),; // 设备拓扑：可参考 https://help.aliyun.com/zh/iot/user-guide/manage-topological-relationships 设备拓扑 + + public static final String[] ARRAYS = Arrays.stream(values()).map(IotDeviceMessageTypeEnum::getType).toArray(String[]::new); + + /** + * 属性 + */ + private final String type; + + @Override + public String[] array() { + return ARRAYS; + } + +} 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 new file mode 100644 index 0000000..6505058 --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/config/IotMessageBusAutoConfiguration.java @@ -0,0 +1,63 @@ +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.mq.producer.IotDeviceMessageProducer; +import lombok.extern.slf4j.Slf4j; +import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties; +import org.apache.rocketmq.spring.core.RocketMQTemplate; +import org.springframework.boot.autoconfigure.AutoConfiguration; +import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.context.ApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +/** + * IoT 消息总线自动配置 + * + * @author 芋道源码 + */ +@AutoConfiguration +@EnableConfigurationProperties(IotMessageBusProperties.class) +@Slf4j +public class IotMessageBusAutoConfiguration { + + @Bean + public IotDeviceMessageProducer deviceMessageProducer(IotMessageBus messageBus) { + return new IotDeviceMessageProducer(messageBus); + } + + // ==================== Local 实现 ==================== + + @Configuration + @ConditionalOnProperty(prefix = "yudao.iot.message-bus", name = "type", havingValue = "local", matchIfMissing = true) + public static class LocalIotMessageBusConfiguration { + + @Bean + public IotMessageBus localIotMessageBus(ApplicationContext applicationContext) { + log.info("[localIotMessageBus][创建 Local IoT 消息总线]"); + return new LocalIotMessageBus(applicationContext); + } + + } + + // ==================== RocketMQ 实现 ==================== + + @Configuration + @ConditionalOnProperty(prefix = "yudao.iot.message-bus", name = "type", havingValue = "rocketmq") + @ConditionalOnClass(RocketMQTemplate.class) + public static class RocketMQIotMessageBusConfiguration { + + @Bean + @SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection") + public IotMessageBus rocketMQIotMessageBus(RocketMQProperties rocketMQProperties, RocketMQTemplate rocketMQTemplate) { + log.info("[rocketMQIotMessageBus][创建 RocketMQ IoT 消息总线]"); + return new RocketMQIotMessageBus(rocketMQProperties, rocketMQTemplate); + } + + } + +} \ 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/config/IotMessageBusProperties.java b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/config/IotMessageBusProperties.java new file mode 100644 index 0000000..501eb2b --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/config/IotMessageBusProperties.java @@ -0,0 +1,28 @@ +package cn.iocoder.yudao.module.iot.core.messagebus.config; + +import lombok.Data; +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.validation.annotation.Validated; + +import jakarta.validation.constraints.NotEmpty; +import jakarta.validation.constraints.NotNull; + +/** + * IoT 消息总线配置属性 + * + * @author 芋道源码 + */ +@ConfigurationProperties("yudao.iot.message-bus") +@Data +@Validated +public class IotMessageBusProperties { + + /** + * 消息总线类型 + * + * 可选值：local、redis、rocketmq、rabbitmq + */ + @NotNull(message = "IoT 消息总线类型不能为空") + private String type = "local"; + +} \ 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/IotMessageBus.java b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/IotMessageBus.java new file mode 100644 index 0000000..b032298 --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/IotMessageBus.java @@ -0,0 +1,27 @@ +package cn.iocoder.yudao.module.iot.core.messagebus.core; + +/** + * IoT 消息总线接口 + * + * 用于在 IoT 系统中发布和订阅消息，支持多种消息中间件实现 + * + * @author 芋道源码 + */ +public interface IotMessageBus { + + /** + * 发布消息到消息总线 + * + * @param topic 主题 + * @param message 消息内容 + */ + void post(String topic, Object message); + + /** + * 注册消息订阅者 + * + * @param subscriber 订阅者 + */ + void register(IotMessageBusSubscriber subscriber); + +} \ 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/IotMessageBusSubscriber.java b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/IotMessageBusSubscriber.java new file mode 100644 index 0000000..631fa88 --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/IotMessageBusSubscriber.java @@ -0,0 +1,29 @@ +package cn.iocoder.yudao.module.iot.core.messagebus.core; + +/** + * IoT 消息总线订阅者接口 + * + * 用于处理从消息总线接收到的消息 + * + * @author 芋道源码 + */ +public interface IotMessageBusSubscriber { + + /** + * @return 主题 + */ + String getTopic(); + + /** + * @return 分组 + */ + String getGroup(); + + /** + * 处理接收到的消息 + * + * @param message 消息内容 + */ + void onMessage(T message); + +} \ 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 new file mode 100644 index 0000000..c8c7277 --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/local/LocalIotMessage.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 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 new file mode 100644 index 0000000..af73547 --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/local/LocalIotMessageBus.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.IotMessageBusSubscriber; +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(IotMessageBusSubscriber 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 (IotMessageBusSubscriber 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/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 new file mode 100644 index 0000000..a304ef4 --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/rocketmq/RocketMQIotMessageBus.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.IotMessageBusSubscriber; +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) { + SendResult result = rocketMQTemplate.syncSend(topic, JsonUtils.toJsonString(message)); + log.info("[post][topic({}) 发送消息({}) result({})]", topic, message, result); + } + + @Override + @SneakyThrows + public void register(IotMessageBusSubscriber 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/main/java/cn/iocoder/yudao/module/iot/core/mq/message/IotDeviceMessage.java b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/mq/message/IotDeviceMessage.java new file mode 100644 index 0000000..43b9f2e --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/mq/message/IotDeviceMessage.java @@ -0,0 +1,77 @@ +package cn.iocoder.yudao.module.iot.core.mq.message; + +import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageIdentifierEnum; +import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageTypeEnum; +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Data; +import lombok.NoArgsConstructor; + +import java.time.LocalDateTime; + +// TODO @芋艿：参考阿里云的物模型，优化 IoT 上下行消息的设计，尽量保持一致（渐进式，不要一口气）！ + +/** + * IoT 设备消息 + */ +@Data +@NoArgsConstructor +@AllArgsConstructor +@Builder +public class IotDeviceMessage { + + /** + * 请求编号 + */ + private String requestId; + + /** + * 设备信息 + */ + private String productKey; + /** + * 设备名称 + */ + private String deviceName; + /** + * 设备标识 + */ + private String deviceKey; + + /** + * 消息类型 + * + * 枚举 {@link IotDeviceMessageTypeEnum} + */ + private String type; + /** + * 标识符 + * + * 枚举 {@link IotDeviceMessageIdentifierEnum} + */ + private String identifier; + + /** + * 请求参数 + * + * 例如说：属性上报的 properties、事件上报的 params + */ + private Object data; + /** + * 响应码 + * + * 目前只有 server 下行消息给 device 设备时，才会有响应码 + */ + private Integer code; + + /** + * 上报时间 + */ + private LocalDateTime reportTime; + + /** + * 租户编号 + */ + private Long tenantId; + +} diff --git a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/mq/producer/IotDeviceMessageProducer.java b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/mq/producer/IotDeviceMessageProducer.java new file mode 100644 index 0000000..7e23dc4 --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/mq/producer/IotDeviceMessageProducer.java @@ -0,0 +1,48 @@ +package cn.iocoder.yudao.module.iot.core.mq.producer; + +import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus; +import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; +import lombok.RequiredArgsConstructor; + +/** + * IoT 设备消息生产者 + * + * @author 芋道源码 + */ +@RequiredArgsConstructor +public class IotDeviceMessageProducer { + + /** + * 【消息总线】应用的设备消息 Topic，由 iot-gateway 发给 iot-biz 进行消费 + */ + private static final String MESSAGE_BUS_DEVICE_MESSAGE_TOPIC = "iot_device_message"; + + /** + * 【消息总线】设备消息 Topic，由 iot-biz 发送给 iot-gateway 的某个 “server”(protocol) 进行消费 + * + * 其中，%s 就是该“server”(protocol) 的标识 + */ + private static final String MESSAGE_BUS_GATEWAY_DEVICE_MESSAGE_TOPIC = MESSAGE_BUS_DEVICE_MESSAGE_TOPIC + "/%s"; + + private final IotMessageBus messageBus; + + /** + * 发送设备消息 + * + * @param message 设备消息 + */ + public void sendDeviceMessage(IotDeviceMessage message) { + messageBus.post(MESSAGE_BUS_DEVICE_MESSAGE_TOPIC, message); + } + + /** + * 发送网关设备消息 + * + * @param server 网关的 server 标识 + * @param message 设备消息 + */ + public void sendGatewayDeviceMessage(String server, Object message) { + messageBus.post(String.format(MESSAGE_BUS_GATEWAY_DEVICE_MESSAGE_TOPIC, server), message); + } + +} diff --git a/yudao-module-iot/yudao-module-iot-core/src/main/resources/META-INF/spring.factories b/yudao-module-iot/yudao-module-iot-core/src/main/resources/META-INF/spring.factories new file mode 100644 index 0000000..bfb4426 --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-core/src/main/resources/META-INF/spring.factories @@ -0,0 +1,2 @@ +org.springframework.boot.autoconfigure.EnableAutoConfiguration=\ +cn.iocoder.yudao.module.iot.core.messagebus.config.IotMessageBusAutoConfiguration \ 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/TestMessage.java b/yudao-module-iot/yudao-module-iot-core/src/test/java/cn/iocoder/yudao/module/iot/core/messagebus/core/TestMessage.java new file mode 100644 index 0000000..e06c9ec --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-core/src/test/java/cn/iocoder/yudao/module/iot/core/messagebus/core/TestMessage.java @@ -0,0 +1,12 @@ +package cn.iocoder.yudao.module.iot.core.messagebus.core; + +import lombok.Data; + +@Data +public class TestMessage { + + private String nickname; + + private Integer age; + +} \ 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 new file mode 100644 index 0000000..de757dd --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-core/src/test/java/cn/iocoder/yudao/module/iot/core/messagebus/core/local/LocalIotMessageBusIntegrationTest.java @@ -0,0 +1,177 @@ +package cn.iocoder.yudao.module.iot.core.messagebus.core.local; + +import cn.iocoder.yudao.module.iot.core.messagebus.config.IotMessageBusAutoConfiguration; +import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus; +import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBusSubscriber; +import jakarta.annotation.Resource; +import lombok.extern.slf4j.Slf4j; +import org.junit.jupiter.api.Test; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.context.annotation.Import; +import org.springframework.test.context.TestPropertySource; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.jupiter.api.Assertions.*; + +/** + * {@link LocalIotMessageBus} 集成测试 + * + * @author 芋道源码 + */ +@SpringBootTest(classes = LocalIotMessageBusIntegrationTest.class) +@Import(IotMessageBusAutoConfiguration.class) +@TestPropertySource(properties = { + "yudao.iot.message-bus.type=local" +}) +@Slf4j +public class LocalIotMessageBusIntegrationTest { + + @Resource + private IotMessageBus messageBus; + + /** + * 1 topic 2 subscriber + */ + @Test + public void testSendMessageWithTwoSubscribers() throws InterruptedException { + // 准备 + String topic = "test-topic"; + String testMessage = "Hello IoT Message Bus!"; + // 用于等待消息处理完成 + CountDownLatch latch = new CountDownLatch(2); + // 用于记录接收到的消息 + AtomicInteger subscriber1Count = new AtomicInteger(0); + AtomicInteger subscriber2Count = new AtomicInteger(0); + + // 创建第一个订阅者 + IotMessageBusSubscriber subscriber1 = new IotMessageBusSubscriber<>() { + + @Override + public String getTopic() { + return topic; + } + + @Override + public String getGroup() { + return "group1"; + } + + @Override + public void onMessage(String message) { + log.info("[订阅者1] 收到消息 - Topic: {}, Message: {}", getTopic(), message); + subscriber1Count.incrementAndGet(); + assertEquals(testMessage, message); + latch.countDown(); + } + + }; + // 创建第二个订阅者 + IotMessageBusSubscriber subscriber2 = new IotMessageBusSubscriber<>() { + + @Override + public String getTopic() { + return topic; + } + + @Override + public String getGroup() { + return "group2"; + } + + @Override + public void onMessage(String message) { + log.info("[订阅者2] 收到消息 - Topic: {}, Message: {}", getTopic(), message); + subscriber2Count.incrementAndGet(); + assertEquals(testMessage, message); + latch.countDown(); + } + + }; + // 注册订阅者 + messageBus.register(subscriber1); + messageBus.register(subscriber2); + + // 发送消息 + log.info("[测试] 发送消息 - Topic: {}, Message: {}", topic, testMessage); + messageBus.post(topic, testMessage); + // 等待消息处理完成（最多等待 10 秒） + boolean completed = latch.await(10, TimeUnit.SECONDS); + + // 验证结果 + assertTrue(completed, "消息处理超时"); + assertEquals(1, subscriber1Count.get(), "订阅者 1 应该收到 1 条消息"); + assertEquals(1, subscriber2Count.get(), "订阅者 2 应该收到 1 条消息"); + log.info("[测试] 测试完成 - 订阅者 1 收到{}条消息，订阅者 2 收到{}条消息", subscriber1Count.get(), subscriber2Count.get()); + } + + /** + * 2 topic 2 subscriber + */ + @Test + public void testMultipleTopics() throws InterruptedException { + // 准备 + String topic1 = "device-status"; + String topic2 = "device-data"; + String message1 = "设备在线"; + String message2 = "温度:25°C"; + CountDownLatch latch = new CountDownLatch(2); + + // 创建订阅者 1 - 只订阅设备状态 + IotMessageBusSubscriber statusSubscriber = new IotMessageBusSubscriber<>() { + + @Override + public String getTopic() { + return topic1; + } + + @Override + public String getGroup() { + return "status-group"; + } + + @Override + public void onMessage(String message) { + log.info("[状态订阅者] 收到消息 - Topic: {}, Message: {}", getTopic(), message); + assertEquals(message1, message); + latch.countDown(); + } + + }; + // 创建订阅者 2 - 只订阅设备数据 + IotMessageBusSubscriber dataSubscriber = new IotMessageBusSubscriber<>() { + + @Override + public String getTopic() { + return topic2; + } + + @Override + public String getGroup() { + return "data-group"; + } + + @Override + public void onMessage(String message) { + log.info("[数据订阅者] 收到消息 - Topic: {}, Message: {}", getTopic(), message); + assertEquals(message2, message); + latch.countDown(); + } + + }; + // 注册订阅者到不同主题 + messageBus.register(statusSubscriber); + messageBus.register(dataSubscriber); + + // 发送消息到不同主题 + messageBus.post(topic1, message1); + messageBus.post(topic2, message2); + // 等待消息处理完成 + boolean completed = latch.await(10, TimeUnit.SECONDS); + assertTrue(completed, "消息处理超时"); + log.info("[测试] 多主题测试完成"); + } + +} \ 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/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 new file mode 100644 index 0000000..babd3b2 --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-core/src/test/java/cn/iocoder/yudao/module/iot/core/messagebus/core/rocketmq/RocketMQIotMessageBusTest.java @@ -0,0 +1,268 @@ +package cn.iocoder.yudao.module.iot.core.messagebus.core.rocketmq; + +import cn.hutool.core.util.IdUtil; +import cn.iocoder.yudao.module.iot.core.messagebus.config.IotMessageBusAutoConfiguration; +import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus; +import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBusSubscriber; +import cn.iocoder.yudao.module.iot.core.messagebus.core.TestMessage; +import jakarta.annotation.Resource; +import lombok.extern.slf4j.Slf4j; +import org.apache.rocketmq.spring.autoconfigure.RocketMQAutoConfiguration; +import org.junit.jupiter.api.Test; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.context.annotation.Import; +import org.springframework.test.context.TestPropertySource; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; + +import static org.junit.jupiter.api.Assertions.*; + +/** + * {@link RocketMQIotMessageBus} 集成测试 + * + * @author 芋道源码 + */ +@SpringBootTest(classes = RocketMQIotMessageBusTest.class) +@Import({RocketMQAutoConfiguration.class, IotMessageBusAutoConfiguration.class}) +@TestPropertySource(properties = { + "yudao.iot.message-bus.type=rocketmq", + "rocketmq.name-server=127.0.0.1:9876", + "rocketmq.producer.group=test-rocketmq-group", + "rocketmq.producer.send-message-timeout=10000" +}) +@Slf4j +public class RocketMQIotMessageBusTest { + + @Resource + private IotMessageBus messageBus; + + /** + * 1 topic 1 subscriber（string） + */ + @Test + public void testSendMessageWithOneSubscriber() throws InterruptedException { + // 准备 + String topic = "test-topic-" + IdUtil.simpleUUID(); +// String topic = "test-topic-pojo"; + String testMessage = "Hello IoT Message Bus!"; + // 用于等待消息处理完成 + CountDownLatch latch = new CountDownLatch(1); + // 用于记录接收到的消息 + AtomicInteger subscriberCount = new AtomicInteger(0); + AtomicReference subscriberMessageRef = new AtomicReference<>(); + + // 发送消息（需要提前发，保证 RocketMQ 路由的创建） + log.info("[测试] 发送消息 - Topic: {}, Message: {}", topic, testMessage); + messageBus.post(topic, testMessage); + + // 创建订阅者 + IotMessageBusSubscriber subscriber1 = new IotMessageBusSubscriber<>() { + + @Override + public String getTopic() { + return topic; + } + + @Override + public String getGroup() { + return "test-topic-" + IdUtil.simpleUUID() + "-consumer"; +// return "test-topic-consumer-01"; + } + + @Override + public void onMessage(String message) { + log.info("[订阅者] 收到消息 - Topic: {}, Message: {}", getTopic(), message); + subscriberCount.incrementAndGet(); + subscriberMessageRef.set(message); + assertEquals(testMessage, message); + latch.countDown(); + } + + }; + // 注册订阅者 + messageBus.register(subscriber1); + + // 等待消息处理完成（最多等待 5 秒） + boolean completed = latch.await(10, TimeUnit.SECONDS); + + // 验证结果 + assertTrue(completed, "消息处理超时"); + assertEquals(1, subscriberCount.get(), "订阅者应该收到 1 条消息"); + log.info("[测试] 测试完成 - 订阅者收到{}条消息", subscriberCount.get()); + assertEquals(testMessage, subscriberMessageRef.get(), "接收到的消息内容不匹配"); + } + + /** + * 1 topic 2 subscriber（pojo） + */ + @Test + public void testSendMessageWithTwoSubscribers() throws InterruptedException { + // 准备 + String topic = "test-topic-" + IdUtil.simpleUUID(); +// String topic = "test-topic-pojo"; + TestMessage testMessage = new TestMessage().setNickname("yunai").setAge(18); + // 用于等待消息处理完成 + CountDownLatch latch = new CountDownLatch(2); + // 用于记录接收到的消息 + AtomicInteger subscriber1Count = new AtomicInteger(0); + AtomicReference subscriber1MessageRef = new AtomicReference<>(); + AtomicInteger subscriber2Count = new AtomicInteger(0); + AtomicReference subscriber2MessageRef = new AtomicReference<>(); + + // 发送消息（需要提前发，保证 RocketMQ 路由的创建） + log.info("[测试] 发送消息 - Topic: {}, Message: {}", topic, testMessage); + messageBus.post(topic, testMessage); + + // 创建第一个订阅者 + IotMessageBusSubscriber subscriber1 = new IotMessageBusSubscriber<>() { + + @Override + public String getTopic() { + return topic; + } + + @Override + public String getGroup() { + return "test-topic-" + IdUtil.simpleUUID() + "-consumer"; +// return "test-topic-consumer-01"; + } + + @Override + public void onMessage(TestMessage message) { + log.info("[订阅者1] 收到消息 - Topic: {}, Message: {}", getTopic(), message); + subscriber1Count.incrementAndGet(); + subscriber1MessageRef.set(message); + assertEquals(testMessage, message); + latch.countDown(); + } + + }; + // 创建第二个订阅者 + IotMessageBusSubscriber subscriber2 = new IotMessageBusSubscriber<>() { + + @Override + public String getTopic() { + return topic; + } + + @Override + public String getGroup() { + return "test-topic-" + IdUtil.simpleUUID() + "-consumer"; +// return "test-topic-consumer-02"; + } + + @Override + public void onMessage(TestMessage message) { + log.info("[订阅者2] 收到消息 - Topic: {}, Message: {}", getTopic(), message); + subscriber2Count.incrementAndGet(); + subscriber2MessageRef.set(message); + assertEquals(testMessage, message); + latch.countDown(); + } + + }; + // 注册订阅者 + messageBus.register(subscriber1); + messageBus.register(subscriber2); + + // 等待消息处理完成（最多等待 5 秒） + boolean completed = latch.await(10, TimeUnit.SECONDS); + + // 验证结果 + assertTrue(completed, "消息处理超时"); + assertEquals(1, subscriber1Count.get(), "订阅者 1 应该收到 1 条消息"); + assertEquals(1, subscriber2Count.get(), "订阅者 2 应该收到 1 条消息"); + log.info("[测试] 测试完成 - 订阅者 1 收到{}条消息，订阅者2收到{}条消息", subscriber1Count.get(), subscriber2Count.get()); + assertEquals(testMessage, subscriber1MessageRef.get(), "接收到的消息内容不匹配"); + assertEquals(testMessage, subscriber2MessageRef.get(), "接收到的消息内容不匹配"); + } + + /** + * 2 topic 2 subscriber + */ + @Test + public void testMultipleTopics() throws InterruptedException { + // 准备 + String topic1 = "device-status-" + IdUtil.simpleUUID(); + String topic2 = "device-data-" + IdUtil.simpleUUID(); + String message1 = "设备在线"; + String message2 = "温度:25°C"; + CountDownLatch latch = new CountDownLatch(2); + AtomicInteger subscriber1Count = new AtomicInteger(0); + AtomicReference subscriber1MessageRef = new AtomicReference<>(); + AtomicInteger subscriber2Count = new AtomicInteger(0); + AtomicReference subscriber2MessageRef = new AtomicReference<>(); + + + // 发送消息到不同主题（需要提前发，保证 RocketMQ 路由的创建） + log.info("[测试] 发送消息 - Topic1: {}, Message1: {}", topic1, message1); + messageBus.post(topic1, message1); + log.info("[测试] 发送消息 - Topic2: {}, Message2: {}", topic2, message2); + messageBus.post(topic2, message2); + + // 创建订阅者 1 - 只订阅设备状态 + IotMessageBusSubscriber statusSubscriber = new IotMessageBusSubscriber<>() { + + @Override + public String getTopic() { + return topic1; + } + + @Override + public String getGroup() { + return "status-group-" + IdUtil.simpleUUID(); + } + + @Override + public void onMessage(String message) { + log.info("[状态订阅者] 收到消息 - Topic: {}, Message: {}", getTopic(), message); + subscriber1Count.incrementAndGet(); + subscriber1MessageRef.set(message); + assertEquals(message1, message); + latch.countDown(); + } + + }; + // 创建订阅者 2 - 只订阅设备数据 + IotMessageBusSubscriber dataSubscriber = new IotMessageBusSubscriber<>() { + + @Override + public String getTopic() { + return topic2; + } + + @Override + public String getGroup() { + return "data-group-" + IdUtil.simpleUUID(); + } + + @Override + public void onMessage(String message) { + log.info("[数据订阅者] 收到消息 - Topic: {}, Message: {}", getTopic(), message); + subscriber2Count.incrementAndGet(); + subscriber2MessageRef.set(message); + assertEquals(message2, message); + latch.countDown(); + } + + }; + // 注册订阅者到不同主题 + messageBus.register(statusSubscriber); + messageBus.register(dataSubscriber); + + // 等待消息处理完成 + boolean completed = latch.await(10, TimeUnit.SECONDS); + + // 验证结果 + assertTrue(completed, "消息处理超时"); + assertEquals(1, subscriber1Count.get(), "状态订阅者应该收到 1 条消息"); + assertEquals(message1, subscriber1MessageRef.get(), "状态订阅者接收到的消息内容不匹配"); + assertEquals(1, subscriber2Count.get(), "数据订阅者应该收到 1 条消息"); + assertEquals(message2, subscriber2MessageRef.get(), "数据订阅者接收到的消息内容不匹配"); + log.info("[测试] 多主题测试完成 - 状态订阅者收到{}条消息，数据订阅者收到{}条消息", subscriber1Count.get(), subscriber2Count.get()); + } + +} \ No newline at end of file diff --git a/yudao-module-iot/yudao-module-iot-core/yudao-module-iot-common/pom.xml b/yudao-module-iot/yudao-module-iot-core/yudao-module-iot-common/pom.xml deleted file mode 100644 index 3cf3fb6..0000000 --- a/yudao-module-iot/yudao-module-iot-core/yudao-module-iot-common/pom.xml +++ /dev/null @@ -1,26 +0,0 @@ - - - yudao-module-iot-core - cn.iocoder.boot - ${revision} - - 4.0.0 - yudao-module-iot-common - jar - - ${project.artifactId} - - iot 模块下，提供通用的功能。 - 1. 跨 iot-biz 和 iot-gateway 的设备消息 - 2. 查询设备信息的通用 API - - - - - cn.iocoder.boot - yudao-common - - - - diff --git a/yudao-module-iot/yudao-module-iot-core/yudao-module-iot-common/src/main/java/cn/iocoder/yudao/module/iot/common/biz/IotDeviceCommonApi.java b/yudao-module-iot/yudao-module-iot-core/yudao-module-iot-common/src/main/java/cn/iocoder/yudao/module/iot/common/biz/IotDeviceCommonApi.java deleted file mode 100644 index 3b5385b..0000000 --- a/yudao-module-iot/yudao-module-iot-core/yudao-module-iot-common/src/main/java/cn/iocoder/yudao/module/iot/common/biz/IotDeviceCommonApi.java +++ /dev/null @@ -1,5 +0,0 @@ -package cn.iocoder.yudao.module.iot.common.biz; - -// TODO @芋艿：待实现 -public interface IotDeviceCommonApi { -} diff --git a/yudao-module-iot/yudao-module-iot-core/yudao-module-iot-common/src/main/java/cn/iocoder/yudao/module/iot/common/enums/IotCommonConstants.java b/yudao-module-iot/yudao-module-iot-core/yudao-module-iot-common/src/main/java/cn/iocoder/yudao/module/iot/common/enums/IotCommonConstants.java deleted file mode 100644 index 4a8860c..0000000 --- a/yudao-module-iot/yudao-module-iot-core/yudao-module-iot-common/src/main/java/cn/iocoder/yudao/module/iot/common/enums/IotCommonConstants.java +++ /dev/null @@ -1,22 +0,0 @@ -package cn.iocoder.yudao.module.iot.common.enums; - -/** - * IoT 通用的枚举 - * - * @author 芋道源码 - */ -public interface IotCommonConstants { - - /** - * 【消息总线】应用的设备消息 Topic，由 iot-gateway 发给 iot-biz 进行消费 - */ - String MESSAGE_BUS_DEVICE_MESSAGE_TOPIC = "iot_device_message"; - - /** - * 【消息总线】设备消息 Topic，由 iot-biz 发送给 iot-gateway 的某个 “server”(protocol) 进行消费 - * - * 其中，%s 就是该“server”(protocol) 的标识 - */ - String MESSAGE_BUS_GATEWAY_DEVICE_MESSAGE_TOPIC = MESSAGE_BUS_DEVICE_MESSAGE_TOPIC + "/%s"; - -} diff --git a/yudao-module-iot/yudao-module-iot-core/yudao-module-iot-common/src/main/java/cn/iocoder/yudao/module/iot/common/enums/IotDeviceMessageIdentifierEnum.java b/yudao-module-iot/yudao-module-iot-core/yudao-module-iot-common/src/main/java/cn/iocoder/yudao/module/iot/common/enums/IotDeviceMessageIdentifierEnum.java deleted file mode 100644 index 88de746..0000000 --- a/yudao-module-iot/yudao-module-iot-core/yudao-module-iot-common/src/main/java/cn/iocoder/yudao/module/iot/common/enums/IotDeviceMessageIdentifierEnum.java +++ /dev/null @@ -1,45 +0,0 @@ -package cn.iocoder.yudao.module.iot.common.enums; - -import lombok.Getter; -import lombok.RequiredArgsConstructor; - -// TODO @芋艿：需要添加对应的 DTO，以及上下行的链路，网关、网关服务、设备等 - -/** - * IoT 设备消息标识符枚举 - */ -@Getter -@RequiredArgsConstructor -public enum IotDeviceMessageIdentifierEnum { - - PROPERTY_GET("get"), // 下行 TODO 芋艿：【讨论】貌似这个“上行”更合理？device 主动拉取配置。和 IotDevicePropertyGetReqDTO 一样的配置 - PROPERTY_SET("set"), // 下行 - PROPERTY_REPORT("report"), // 上行 - - STATE_ONLINE("online"), // 上行 - STATE_OFFLINE("offline"), // 上行 - - CONFIG_GET("get"), // 上行 TODO 芋艿：【讨论】暂时没有上行的场景 - CONFIG_SET("set"), // 下行 - - SERVICE_INVOKE("${identifier}"), // 下行 - SERVICE_REPLY_SUFFIX("_reply"), // 芋艿：TODO 芋艿：【讨论】上行 or 下行 - - OTA_UPGRADE("upgrade"), // 下行 - OTA_PULL("pull"), // 上行 - OTA_PROGRESS("progress"), // 上行 - OTA_REPORT("report"), // 上行 - - REGISTER_REGISTER("register"), // 上行 - REGISTER_REGISTER_SUB("register_sub"), // 上行 - REGISTER_UNREGISTER_SUB("unregister_sub"), // 下行 - - TOPOLOGY_ADD("topology_add"), // 下行; - ; - - /** - * 标志符 - */ - private final String identifier; - -} \ No newline at end of file diff --git a/yudao-module-iot/yudao-module-iot-core/yudao-module-iot-common/src/main/java/cn/iocoder/yudao/module/iot/common/enums/IotDeviceMessageTypeEnum.java b/yudao-module-iot/yudao-module-iot-core/yudao-module-iot-common/src/main/java/cn/iocoder/yudao/module/iot/common/enums/IotDeviceMessageTypeEnum.java deleted file mode 100644 index 156e614..0000000 --- a/yudao-module-iot/yudao-module-iot-core/yudao-module-iot-common/src/main/java/cn/iocoder/yudao/module/iot/common/enums/IotDeviceMessageTypeEnum.java +++ /dev/null @@ -1,37 +0,0 @@ -package cn.iocoder.yudao.module.iot.common.enums; - -import cn.iocoder.yudao.framework.common.core.ArrayValuable; -import lombok.Getter; -import lombok.RequiredArgsConstructor; - -import java.util.Arrays; - -/** - * IoT 设备消息类型枚举 - */ -@Getter -@RequiredArgsConstructor -public enum IotDeviceMessageTypeEnum implements ArrayValuable { - - STATE("state"), // 设备状态 - PROPERTY("property"), // 设备属性：可参考 https://help.aliyun.com/zh/iot/user-guide/device-properties-events-and-services 设备属性、事件、服务 - EVENT("event"), // 设备事件：可参考 https://help.aliyun.com/zh/iot/user-guide/device-properties-events-and-services 设备属性、事件、服务 - SERVICE("service"), // 设备服务：可参考 https://help.aliyun.com/zh/iot/user-guide/device-properties-events-and-services 设备属性、事件、服务 - CONFIG("config"), // 设备配置：可参考 https://help.aliyun.com/zh/iot/user-guide/remote-configuration-1 远程配置 - OTA("ota"), // 设备 OTA：可参考 https://help.aliyun.com/zh/iot/user-guide/ota-update OTA 升级 - REGISTER("register"), // 设备注册：可参考 https://help.aliyun.com/zh/iot/user-guide/register-devices 设备身份注册 - TOPOLOGY("topology"),; // 设备拓扑：可参考 https://help.aliyun.com/zh/iot/user-guide/manage-topological-relationships 设备拓扑 - - public static final String[] ARRAYS = Arrays.stream(values()).map(IotDeviceMessageTypeEnum::getType).toArray(String[]::new); - - /** - * 属性 - */ - private final String type; - - @Override - public String[] array() { - return ARRAYS; - } - -} diff --git a/yudao-module-iot/yudao-module-iot-core/yudao-module-iot-common/src/main/java/cn/iocoder/yudao/module/iot/common/message/IotDeviceMessage.java b/yudao-module-iot/yudao-module-iot-core/yudao-module-iot-common/src/main/java/cn/iocoder/yudao/module/iot/common/message/IotDeviceMessage.java deleted file mode 100644 index c55e806..0000000 --- a/yudao-module-iot/yudao-module-iot-core/yudao-module-iot-common/src/main/java/cn/iocoder/yudao/module/iot/common/message/IotDeviceMessage.java +++ /dev/null @@ -1,77 +0,0 @@ -package cn.iocoder.yudao.module.iot.common.message; - -import cn.iocoder.yudao.module.iot.common.enums.IotDeviceMessageIdentifierEnum; -import cn.iocoder.yudao.module.iot.common.enums.IotDeviceMessageTypeEnum; -import lombok.AllArgsConstructor; -import lombok.Builder; -import lombok.Data; -import lombok.NoArgsConstructor; - -import java.time.LocalDateTime; - -// TODO @芋艿：参考阿里云的物模型，优化 IoT 上下行消息的设计，尽量保持一致（渐进式，不要一口气）！ - -/** - * IoT 设备消息 - */ -@Data -@NoArgsConstructor -@AllArgsConstructor -@Builder -public class IotDeviceMessage { - - /** - * 请求编号 - */ - private String requestId; - - /** - * 设备信息 - */ - private String productKey; - /** - * 设备名称 - */ - private String deviceName; - /** - * 设备标识 - */ - private String deviceKey; - - /** - * 消息类型 - * - * 枚举 {@link IotDeviceMessageTypeEnum} - */ - private String type; - /** - * 标识符 - * - * 枚举 {@link IotDeviceMessageIdentifierEnum} - */ - private String identifier; - - /** - * 请求参数 - * - * 例如说：属性上报的 properties、事件上报的 params - */ - private Object data; - /** - * 响应码 - * - * 目前只有 server 下行消息给 device 设备时，才会有响应码 - */ - private Integer code; - - /** - * 上报时间 - */ - private LocalDateTime reportTime; - - /** - * 租户编号 - */ - private Long tenantId; - -} diff --git a/yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/pom.xml b/yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/pom.xml deleted file mode 100644 index 436ec9e..0000000 --- a/yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/pom.xml +++ /dev/null @@ -1,65 +0,0 @@ - - - - yudao-module-iot-core - cn.iocoder.boot - ${revision} - - 4.0.0 - yudao-module-iot-message-bus - jar - - ${project.artifactId} - - iot 模块下，提供消息总线的功能。 - 可选择使用 spring event、redis stream、rocketmq、kafka、rabbitmq 等。 - - - - - cn.iocoder.boot - yudao-common - - - - - org.springframework.boot - spring-boot-starter - - - - - org.springframework.data - spring-data-redis - true - - - - org.apache.rocketmq - rocketmq-spring-boot-starter - true - - - - org.springframework.amqp - spring-rabbit - true - - - - org.springframework.kafka - spring-kafka - true - - - - - org.springframework.boot - spring-boot-starter-test - test - - - - \ No newline at end of file diff --git a/yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/src/main/java/cn/iocoder/yudao/module/iot/messagebus/config/IotMessageBusAutoConfiguration.java b/yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/src/main/java/cn/iocoder/yudao/module/iot/messagebus/config/IotMessageBusAutoConfiguration.java deleted file mode 100644 index aa216b5..0000000 --- a/yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/src/main/java/cn/iocoder/yudao/module/iot/messagebus/config/IotMessageBusAutoConfiguration.java +++ /dev/null @@ -1,56 +0,0 @@ -package cn.iocoder.yudao.module.iot.messagebus.config; - -import cn.iocoder.yudao.module.iot.messagebus.core.IotMessageBus; -import cn.iocoder.yudao.module.iot.messagebus.core.local.LocalIotMessageBus; -import cn.iocoder.yudao.module.iot.messagebus.core.rocketmq.RocketMQIotMessageBus; -import lombok.extern.slf4j.Slf4j; -import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties; -import org.apache.rocketmq.spring.core.RocketMQTemplate; -import org.springframework.boot.autoconfigure.AutoConfiguration; -import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; -import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; -import org.springframework.boot.context.properties.EnableConfigurationProperties; -import org.springframework.context.ApplicationContext; -import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Configuration; - -/** - * IoT 消息总线自动配置 - * - * @author 芋道源码 - */ -@AutoConfiguration -@EnableConfigurationProperties(IotMessageBusProperties.class) -@Slf4j -public class IotMessageBusAutoConfiguration { - - // ==================== Local 实现 ==================== - - @Configuration - @ConditionalOnProperty(prefix = "yudao.iot.message-bus", name = "type", havingValue = "local", matchIfMissing = true) - public static class LocalIotMessageBusConfiguration { - - @Bean - public IotMessageBus localIotMessageBus(ApplicationContext applicationContext) { - log.info("[localIotMessageBus][创建 Local IoT 消息总线]"); - return new LocalIotMessageBus(applicationContext); - } - - } - - // ==================== RocketMQ 实现 ==================== - - @Configuration - @ConditionalOnProperty(prefix = "yudao.iot.message-bus", name = "type", havingValue = "rocketmq") - @ConditionalOnClass(RocketMQTemplate.class) - public static class RocketMQIotMessageBusConfiguration { - - @Bean - public IotMessageBus rocketMQIotMessageBus(RocketMQProperties rocketMQProperties, RocketMQTemplate rocketMQTemplate) { - log.info("[rocketMQIotMessageBus][创建 RocketMQ IoT 消息总线]"); - return new RocketMQIotMessageBus(rocketMQProperties, rocketMQTemplate); - } - - } - -} \ No newline at end of file diff --git a/yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/src/main/java/cn/iocoder/yudao/module/iot/messagebus/config/IotMessageBusProperties.java b/yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/src/main/java/cn/iocoder/yudao/module/iot/messagebus/config/IotMessageBusProperties.java deleted file mode 100644 index eac974e..0000000 --- a/yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/src/main/java/cn/iocoder/yudao/module/iot/messagebus/config/IotMessageBusProperties.java +++ /dev/null @@ -1,28 +0,0 @@ -package cn.iocoder.yudao.module.iot.messagebus.config; - -import lombok.Data; -import org.springframework.boot.context.properties.ConfigurationProperties; -import org.springframework.validation.annotation.Validated; - -import jakarta.validation.constraints.NotEmpty; -import jakarta.validation.constraints.NotNull; - -/** - * IoT 消息总线配置属性 - * - * @author 芋道源码 - */ -@ConfigurationProperties("yudao.iot.message-bus") -@Data -@Validated -public class IotMessageBusProperties { - - /** - * 消息总线类型 - * - * 可选值：local、redis、rocketmq、rabbitmq - */ - @NotNull(message = "IoT 消息总线类型不能为空") - private String type = "local"; - -} \ No newline at end of file diff --git a/yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/src/main/java/cn/iocoder/yudao/module/iot/messagebus/core/IotMessageBus.java b/yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/src/main/java/cn/iocoder/yudao/module/iot/messagebus/core/IotMessageBus.java deleted file mode 100644 index 931e963..0000000 --- a/yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/src/main/java/cn/iocoder/yudao/module/iot/messagebus/core/IotMessageBus.java +++ /dev/null @@ -1,27 +0,0 @@ -package cn.iocoder.yudao.module.iot.messagebus.core; - -/** - * IoT 消息总线接口 - * - * 用于在 IoT 系统中发布和订阅消息，支持多种消息中间件实现 - * - * @author 芋道源码 - */ -public interface IotMessageBus { - - /** - * 发布消息到消息总线 - * - * @param topic 主题 - * @param message 消息内容 - */ - void post(String topic, Object message); - - /** - * 注册消息订阅者 - * - * @param subscriber 订阅者 - */ - void register(IotMessageBusSubscriber subscriber); - -} \ No newline at end of file diff --git a/yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/src/main/java/cn/iocoder/yudao/module/iot/messagebus/core/IotMessageBusSubscriber.java b/yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/src/main/java/cn/iocoder/yudao/module/iot/messagebus/core/IotMessageBusSubscriber.java deleted file mode 100644 index a8bdff9..0000000 --- a/yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/src/main/java/cn/iocoder/yudao/module/iot/messagebus/core/IotMessageBusSubscriber.java +++ /dev/null @@ -1,29 +0,0 @@ -package cn.iocoder.yudao.module.iot.messagebus.core; - -/** - * IoT 消息总线订阅者接口 - * - * 用于处理从消息总线接收到的消息 - * - * @author 芋道源码 - */ -public interface IotMessageBusSubscriber { - - /** - * @return 主题 - */ - String getTopic(); - - /** - * @return 分组 - */ - String getGroup(); - - /** - * 处理接收到的消息 - * - * @param message 消息内容 - */ - void onMessage(T message); - -} \ No newline at end of file diff --git a/yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/src/main/java/cn/iocoder/yudao/module/iot/messagebus/core/local/LocalIotMessage.java b/yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/src/main/java/cn/iocoder/yudao/module/iot/messagebus/core/local/LocalIotMessage.java deleted file mode 100644 index bd048e5..0000000 --- a/yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/src/main/java/cn/iocoder/yudao/module/iot/messagebus/core/local/LocalIotMessage.java +++ /dev/null @@ -1,14 +0,0 @@ -package cn.iocoder.yudao.module.iot.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/yudao-module-iot-message-bus/src/main/java/cn/iocoder/yudao/module/iot/messagebus/core/local/LocalIotMessageBus.java b/yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/src/main/java/cn/iocoder/yudao/module/iot/messagebus/core/local/LocalIotMessageBus.java deleted file mode 100644 index 5a27a67..0000000 --- a/yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/src/main/java/cn/iocoder/yudao/module/iot/messagebus/core/local/LocalIotMessageBus.java +++ /dev/null @@ -1,69 +0,0 @@ -package cn.iocoder.yudao.module.iot.messagebus.core.local; - -import cn.hutool.core.collection.CollUtil; -import cn.iocoder.yudao.module.iot.messagebus.core.IotMessageBus; -import cn.iocoder.yudao.module.iot.messagebus.core.IotMessageBusSubscriber; -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; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.CopyOnWriteArrayList; - -/** - * 本地的 {@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(IotMessageBusSubscriber 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 (IotMessageBusSubscriber 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/yudao-module-iot-message-bus/src/main/java/cn/iocoder/yudao/module/iot/messagebus/core/rocketmq/RocketMQIotMessageBus.java b/yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/src/main/java/cn/iocoder/yudao/module/iot/messagebus/core/rocketmq/RocketMQIotMessageBus.java deleted file mode 100644 index 346d66e..0000000 --- a/yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/src/main/java/cn/iocoder/yudao/module/iot/messagebus/core/rocketmq/RocketMQIotMessageBus.java +++ /dev/null @@ -1,98 +0,0 @@ -package cn.iocoder.yudao.module.iot.messagebus.core.rocketmq; - -import cn.hutool.core.util.TypeUtil; -import cn.iocoder.yudao.framework.common.util.json.JsonUtils; -import cn.iocoder.yudao.module.iot.messagebus.core.IotMessageBus; -import cn.iocoder.yudao.module.iot.messagebus.core.IotMessageBusSubscriber; -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) { - SendResult result = rocketMQTemplate.syncSend(topic, JsonUtils.toJsonString(message)); - log.info("[post][topic({}) 发送消息({}) result({})]", topic, message, result); - } - - @Override - @SneakyThrows - public void register(IotMessageBusSubscriber 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/yudao-module-iot-message-bus/src/main/resources/META-INF/spring.factories b/yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/src/main/resources/META-INF/spring.factories deleted file mode 100644 index 04096de..0000000 --- a/yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/src/main/resources/META-INF/spring.factories +++ /dev/null @@ -1,2 +0,0 @@ -org.springframework.boot.autoconfigure.EnableAutoConfiguration=\ -cn.iocoder.yudao.module.iot.messagebus.config.IotMessageBusAutoConfiguration \ No newline at end of file diff --git a/yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/src/test/java/cn/iocoder/yudao/module/iot/messagebus/core/TestMessage.java b/yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/src/test/java/cn/iocoder/yudao/module/iot/messagebus/core/TestMessage.java deleted file mode 100644 index df12d60..0000000 --- a/yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/src/test/java/cn/iocoder/yudao/module/iot/messagebus/core/TestMessage.java +++ /dev/null @@ -1,12 +0,0 @@ -package cn.iocoder.yudao.module.iot.messagebus.core; - -import lombok.Data; - -@Data -public class TestMessage { - - private String nickname; - - private Integer age; - -} \ No newline at end of file diff --git a/yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/src/test/java/cn/iocoder/yudao/module/iot/messagebus/core/local/LocalIotMessageBusIntegrationTest.java b/yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/src/test/java/cn/iocoder/yudao/module/iot/messagebus/core/local/LocalIotMessageBusIntegrationTest.java deleted file mode 100644 index c44997f..0000000 --- a/yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/src/test/java/cn/iocoder/yudao/module/iot/messagebus/core/local/LocalIotMessageBusIntegrationTest.java +++ /dev/null @@ -1,177 +0,0 @@ -package cn.iocoder.yudao.module.iot.messagebus.core.local; - -import cn.iocoder.yudao.module.iot.messagebus.config.IotMessageBusAutoConfiguration; -import cn.iocoder.yudao.module.iot.messagebus.core.IotMessageBus; -import cn.iocoder.yudao.module.iot.messagebus.core.IotMessageBusSubscriber; -import jakarta.annotation.Resource; -import lombok.extern.slf4j.Slf4j; -import org.junit.jupiter.api.Test; -import org.springframework.boot.test.context.SpringBootTest; -import org.springframework.context.annotation.Import; -import org.springframework.test.context.TestPropertySource; - -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicInteger; - -import static org.junit.jupiter.api.Assertions.*; - -/** - * {@link LocalIotMessageBus} 集成测试 - * - * @author 芋道源码 - */ -@SpringBootTest(classes = LocalIotMessageBusIntegrationTest.class) -@Import(IotMessageBusAutoConfiguration.class) -@TestPropertySource(properties = { - "yudao.iot.message-bus.type=local" -}) -@Slf4j -public class LocalIotMessageBusIntegrationTest { - - @Resource - private IotMessageBus messageBus; - - /** - * 1 topic 2 subscriber - */ - @Test - public void testSendMessageWithTwoSubscribers() throws InterruptedException { - // 准备 - String topic = "test-topic"; - String testMessage = "Hello IoT Message Bus!"; - // 用于等待消息处理完成 - CountDownLatch latch = new CountDownLatch(2); - // 用于记录接收到的消息 - AtomicInteger subscriber1Count = new AtomicInteger(0); - AtomicInteger subscriber2Count = new AtomicInteger(0); - - // 创建第一个订阅者 - IotMessageBusSubscriber subscriber1 = new IotMessageBusSubscriber<>() { - - @Override - public String getTopic() { - return topic; - } - - @Override - public String getGroup() { - return "group1"; - } - - @Override - public void onMessage(String message) { - log.info("[订阅者1] 收到消息 - Topic: {}, Message: {}", getTopic(), message); - subscriber1Count.incrementAndGet(); - assertEquals(testMessage, message); - latch.countDown(); - } - - }; - // 创建第二个订阅者 - IotMessageBusSubscriber subscriber2 = new IotMessageBusSubscriber<>() { - - @Override - public String getTopic() { - return topic; - } - - @Override - public String getGroup() { - return "group2"; - } - - @Override - public void onMessage(String message) { - log.info("[订阅者2] 收到消息 - Topic: {}, Message: {}", getTopic(), message); - subscriber2Count.incrementAndGet(); - assertEquals(testMessage, message); - latch.countDown(); - } - - }; - // 注册订阅者 - messageBus.register(subscriber1); - messageBus.register(subscriber2); - - // 发送消息 - log.info("[测试] 发送消息 - Topic: {}, Message: {}", topic, testMessage); - messageBus.post(topic, testMessage); - // 等待消息处理完成（最多等待 10 秒） - boolean completed = latch.await(10, TimeUnit.SECONDS); - - // 验证结果 - assertTrue(completed, "消息处理超时"); - assertEquals(1, subscriber1Count.get(), "订阅者 1 应该收到 1 条消息"); - assertEquals(1, subscriber2Count.get(), "订阅者 2 应该收到 1 条消息"); - log.info("[测试] 测试完成 - 订阅者 1 收到{}条消息，订阅者 2 收到{}条消息", subscriber1Count.get(), subscriber2Count.get()); - } - - /** - * 2 topic 2 subscriber - */ - @Test - public void testMultipleTopics() throws InterruptedException { - // 准备 - String topic1 = "device-status"; - String topic2 = "device-data"; - String message1 = "设备在线"; - String message2 = "温度:25°C"; - CountDownLatch latch = new CountDownLatch(2); - - // 创建订阅者 1 - 只订阅设备状态 - IotMessageBusSubscriber statusSubscriber = new IotMessageBusSubscriber<>() { - - @Override - public String getTopic() { - return topic1; - } - - @Override - public String getGroup() { - return "status-group"; - } - - @Override - public void onMessage(String message) { - log.info("[状态订阅者] 收到消息 - Topic: {}, Message: {}", getTopic(), message); - assertEquals(message1, message); - latch.countDown(); - } - - }; - // 创建订阅者 2 - 只订阅设备数据 - IotMessageBusSubscriber dataSubscriber = new IotMessageBusSubscriber<>() { - - @Override - public String getTopic() { - return topic2; - } - - @Override - public String getGroup() { - return "data-group"; - } - - @Override - public void onMessage(String message) { - log.info("[数据订阅者] 收到消息 - Topic: {}, Message: {}", getTopic(), message); - assertEquals(message2, message); - latch.countDown(); - } - - }; - // 注册订阅者到不同主题 - messageBus.register(statusSubscriber); - messageBus.register(dataSubscriber); - - // 发送消息到不同主题 - messageBus.post(topic1, message1); - messageBus.post(topic2, message2); - // 等待消息处理完成 - boolean completed = latch.await(10, TimeUnit.SECONDS); - assertTrue(completed, "消息处理超时"); - log.info("[测试] 多主题测试完成"); - } - -} \ No newline at end of file diff --git a/yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/src/test/java/cn/iocoder/yudao/module/iot/messagebus/core/rocketmq/RocketMQIotMessageBusTest.java b/yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/src/test/java/cn/iocoder/yudao/module/iot/messagebus/core/rocketmq/RocketMQIotMessageBusTest.java deleted file mode 100644 index bd8ab07..0000000 --- a/yudao-module-iot/yudao-module-iot-core/yudao-module-iot-message-bus/src/test/java/cn/iocoder/yudao/module/iot/messagebus/core/rocketmq/RocketMQIotMessageBusTest.java +++ /dev/null @@ -1,271 +0,0 @@ -package cn.iocoder.yudao.module.iot.messagebus.core.rocketmq; - -import cn.hutool.core.util.IdUtil; -import cn.iocoder.yudao.module.iot.messagebus.config.IotMessageBusAutoConfiguration; -import cn.iocoder.yudao.module.iot.messagebus.core.IotMessageBus; -import cn.iocoder.yudao.module.iot.messagebus.core.IotMessageBusSubscriber; -import cn.iocoder.yudao.module.iot.messagebus.core.TestMessage; -import jakarta.annotation.Resource; -import lombok.Data; -import lombok.extern.slf4j.Slf4j; -import org.apache.rocketmq.spring.autoconfigure.RocketMQAutoConfiguration; -import org.apache.rocketmq.spring.core.RocketMQTemplate; -import org.junit.jupiter.api.Test; -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.boot.test.context.SpringBootTest; -import org.springframework.context.annotation.Import; -import org.springframework.test.context.TestPropertySource; - -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicInteger; -import java.util.concurrent.atomic.AtomicReference; - -import static org.junit.jupiter.api.Assertions.*; - -/** - * {@link RocketMQIotMessageBus} 集成测试 - * - * @author 芋道源码 - */ -@SpringBootTest(classes = RocketMQIotMessageBusTest.class) -@Import({RocketMQAutoConfiguration.class, IotMessageBusAutoConfiguration.class}) -@TestPropertySource(properties = { - "yudao.iot.message-bus.type=rocketmq", - "rocketmq.name-server=127.0.0.1:9876", - "rocketmq.producer.group=test-rocketmq-group", - "rocketmq.producer.send-message-timeout=10000" -}) -@Slf4j -public class RocketMQIotMessageBusTest { - - @Resource - private IotMessageBus messageBus; - - /** - * 1 topic 1 subscriber（string） - */ - @Test - public void testSendMessageWithOneSubscriber() throws InterruptedException { - // 准备 - String topic = "test-topic-" + IdUtil.simpleUUID(); -// String topic = "test-topic-pojo"; - String testMessage = "Hello IoT Message Bus!"; - // 用于等待消息处理完成 - CountDownLatch latch = new CountDownLatch(1); - // 用于记录接收到的消息 - AtomicInteger subscriberCount = new AtomicInteger(0); - AtomicReference subscriberMessageRef = new AtomicReference<>(); - - // 发送消息（需要提前发，保证 RocketMQ 路由的创建） - log.info("[测试] 发送消息 - Topic: {}, Message: {}", topic, testMessage); - messageBus.post(topic, testMessage); - - // 创建订阅者 - IotMessageBusSubscriber subscriber1 = new IotMessageBusSubscriber<>() { - - @Override - public String getTopic() { - return topic; - } - - @Override - public String getGroup() { - return "test-topic-" + IdUtil.simpleUUID() + "-consumer"; -// return "test-topic-consumer-01"; - } - - @Override - public void onMessage(String message) { - log.info("[订阅者] 收到消息 - Topic: {}, Message: {}", getTopic(), message); - subscriberCount.incrementAndGet(); - subscriberMessageRef.set(message); - assertEquals(testMessage, message); - latch.countDown(); - } - - }; - // 注册订阅者 - messageBus.register(subscriber1); - - // 等待消息处理完成（最多等待 5 秒） - boolean completed = latch.await(10, TimeUnit.SECONDS); - - // 验证结果 - assertTrue(completed, "消息处理超时"); - assertEquals(1, subscriberCount.get(), "订阅者应该收到 1 条消息"); - log.info("[测试] 测试完成 - 订阅者收到{}条消息", subscriberCount.get()); - assertEquals(testMessage, subscriberMessageRef.get(), "接收到的消息内容不匹配"); - } - - /** - * 1 topic 2 subscriber（pojo） - */ - @Test - public void testSendMessageWithTwoSubscribers() throws InterruptedException { - // 准备 - String topic = "test-topic-" + IdUtil.simpleUUID(); -// String topic = "test-topic-pojo"; - TestMessage testMessage = new TestMessage().setNickname("yunai").setAge(18); - // 用于等待消息处理完成 - CountDownLatch latch = new CountDownLatch(2); - // 用于记录接收到的消息 - AtomicInteger subscriber1Count = new AtomicInteger(0); - AtomicReference subscriber1MessageRef = new AtomicReference<>(); - AtomicInteger subscriber2Count = new AtomicInteger(0); - AtomicReference subscriber2MessageRef = new AtomicReference<>(); - - // 发送消息（需要提前发，保证 RocketMQ 路由的创建） - log.info("[测试] 发送消息 - Topic: {}, Message: {}", topic, testMessage); - messageBus.post(topic, testMessage); - - // 创建第一个订阅者 - IotMessageBusSubscriber subscriber1 = new IotMessageBusSubscriber<>() { - - @Override - public String getTopic() { - return topic; - } - - @Override - public String getGroup() { - return "test-topic-" + IdUtil.simpleUUID() + "-consumer"; -// return "test-topic-consumer-01"; - } - - @Override - public void onMessage(TestMessage message) { - log.info("[订阅者1] 收到消息 - Topic: {}, Message: {}", getTopic(), message); - subscriber1Count.incrementAndGet(); - subscriber1MessageRef.set(message); - assertEquals(testMessage, message); - latch.countDown(); - } - - }; - // 创建第二个订阅者 - IotMessageBusSubscriber subscriber2 = new IotMessageBusSubscriber<>() { - - @Override - public String getTopic() { - return topic; - } - - @Override - public String getGroup() { - return "test-topic-" + IdUtil.simpleUUID() + "-consumer"; -// return "test-topic-consumer-02"; - } - - @Override - public void onMessage(TestMessage message) { - log.info("[订阅者2] 收到消息 - Topic: {}, Message: {}", getTopic(), message); - subscriber2Count.incrementAndGet(); - subscriber2MessageRef.set(message); - assertEquals(testMessage, message); - latch.countDown(); - } - - }; - // 注册订阅者 - messageBus.register(subscriber1); - messageBus.register(subscriber2); - - // 等待消息处理完成（最多等待 5 秒） - boolean completed = latch.await(10, TimeUnit.SECONDS); - - // 验证结果 - assertTrue(completed, "消息处理超时"); - assertEquals(1, subscriber1Count.get(), "订阅者 1 应该收到 1 条消息"); - assertEquals(1, subscriber2Count.get(), "订阅者 2 应该收到 1 条消息"); - log.info("[测试] 测试完成 - 订阅者 1 收到{}条消息，订阅者2收到{}条消息", subscriber1Count.get(), subscriber2Count.get()); - assertEquals(testMessage, subscriber1MessageRef.get(), "接收到的消息内容不匹配"); - assertEquals(testMessage, subscriber2MessageRef.get(), "接收到的消息内容不匹配"); - } - - /** - * 2 topic 2 subscriber - */ - @Test - public void testMultipleTopics() throws InterruptedException { - // 准备 - String topic1 = "device-status-" + IdUtil.simpleUUID(); - String topic2 = "device-data-" + IdUtil.simpleUUID(); - String message1 = "设备在线"; - String message2 = "温度:25°C"; - CountDownLatch latch = new CountDownLatch(2); - AtomicInteger subscriber1Count = new AtomicInteger(0); - AtomicReference subscriber1MessageRef = new AtomicReference<>(); - AtomicInteger subscriber2Count = new AtomicInteger(0); - AtomicReference subscriber2MessageRef = new AtomicReference<>(); - - - // 发送消息到不同主题（需要提前发，保证 RocketMQ 路由的创建） - log.info("[测试] 发送消息 - Topic1: {}, Message1: {}", topic1, message1); - messageBus.post(topic1, message1); - log.info("[测试] 发送消息 - Topic2: {}, Message2: {}", topic2, message2); - messageBus.post(topic2, message2); - - // 创建订阅者 1 - 只订阅设备状态 - IotMessageBusSubscriber statusSubscriber = new IotMessageBusSubscriber<>() { - - @Override - public String getTopic() { - return topic1; - } - - @Override - public String getGroup() { - return "status-group-" + IdUtil.simpleUUID(); - } - - @Override - public void onMessage(String message) { - log.info("[状态订阅者] 收到消息 - Topic: {}, Message: {}", getTopic(), message); - subscriber1Count.incrementAndGet(); - subscriber1MessageRef.set(message); - assertEquals(message1, message); - latch.countDown(); - } - - }; - // 创建订阅者 2 - 只订阅设备数据 - IotMessageBusSubscriber dataSubscriber = new IotMessageBusSubscriber<>() { - - @Override - public String getTopic() { - return topic2; - } - - @Override - public String getGroup() { - return "data-group-" + IdUtil.simpleUUID(); - } - - @Override - public void onMessage(String message) { - log.info("[数据订阅者] 收到消息 - Topic: {}, Message: {}", getTopic(), message); - subscriber2Count.incrementAndGet(); - subscriber2MessageRef.set(message); - assertEquals(message2, message); - latch.countDown(); - } - - }; - // 注册订阅者到不同主题 - messageBus.register(statusSubscriber); - messageBus.register(dataSubscriber); - - // 等待消息处理完成 - boolean completed = latch.await(10, TimeUnit.SECONDS); - - // 验证结果 - assertTrue(completed, "消息处理超时"); - assertEquals(1, subscriber1Count.get(), "状态订阅者应该收到 1 条消息"); - assertEquals(message1, subscriber1MessageRef.get(), "状态订阅者接收到的消息内容不匹配"); - assertEquals(1, subscriber2Count.get(), "数据订阅者应该收到 1 条消息"); - assertEquals(message2, subscriber2MessageRef.get(), "数据订阅者接收到的消息内容不匹配"); - log.info("[测试] 多主题测试完成 - 状态订阅者收到{}条消息，数据订阅者收到{}条消息", subscriber1Count.get(), subscriber2Count.get()); - } - -} \ No newline at end of file -- libgit2 0.21.4