feat: 增加 IoT Kafka 消息总线实现

- 新增 IotKafkaMessageBus，支持 IoT gateway 与 yudao-server 通过 Kafka 通信 - 增加 type=kafka 自动配置，并补充消息总线类型说明 - 为 iot-gateway 补充 spring-kafka 依赖和 Kafka 连接配置 - 将 iot-gateway 与 yudao-server 的 IoT message-bus 默认切换为 kafka

feat: 增加 IoT Kafka 消息总线实现
- 新增 IotKafkaMessageBus，支持 IoT gateway 与 yudao-server 通过 Kafka 通信 - 增加 type=kafka 自动配置，并补充消息总线类型说明 - 为 iot-gateway 补充 spring-kafka 依赖和 Kafka 连接配置 - 将 iot-gateway 与 yudao-server 的 IoT message-bus 默认切换为 kafka
YunaiV
1 parent 93dd97af
Showing 4 changed files with 28 additions and 3 deletions
yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/config/IotMessageBusAutoConfiguration.java
yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/config/IotMessageBusProperties.java
yudao-module-iot/yudao-module-iot-gateway/pom.xml
yudao-module-iot/yudao-module-iot-gateway/src/main/resources/application.yaml
@@ -6,6 +6,7 @@ import cn.iocoder.yudao.framework.mq.redis.core.job.RedisStreamMessageCleanupJob
 import cn.iocoder.yudao.framework.mq.redis.core.stream.AbstractRedisStreamMessage;
 import cn.iocoder.yudao.framework.mq.redis.core.stream.AbstractRedisStreamMessageListener;
 import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus;
+import cn.iocoder.yudao.module.iot.core.messagebus.core.kafka.IotKafkaMessageBus;
 import cn.iocoder.yudao.module.iot.core.messagebus.core.local.IotLocalMessageBus;
 import cn.iocoder.yudao.module.iot.core.messagebus.core.rabbitmq.IotRabbitMQMessageBus;
 import cn.iocoder.yudao.module.iot.core.messagebus.core.redis.IotRedisMessageBus;
@@ -18,6 +19,7 @@ import org.redisson.api.RedissonClient;
 import org.springframework.amqp.rabbit.core.RabbitAdmin;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.boot.autoconfigure.AutoConfiguration;
+import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
@@ -27,6 +29,7 @@ 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;
+import org.springframework.kafka.core.KafkaTemplate;
 import java.util.List;
@@ -77,6 +80,21 @@ public class IotMessageBusAutoConfiguration {
     }
+    // ==================== Kafka 实现 ====================
+
+    @Configuration
+    @ConditionalOnProperty(prefix = "yudao.iot.message-bus", name = "type", havingValue = "kafka")
+    @ConditionalOnClass(KafkaTemplate.class)
+    public static class IotKafkaMessageBusConfiguration {
+
+        @Bean
+        public IotKafkaMessageBus iotKafkaMessageBus(KafkaProperties kafkaProperties) {
+            log.info("[iotKafkaMessageBus][创建 IoT Kafka 消息总线]");
+            return new IotKafkaMessageBus(kafkaProperties);
+        }
+
+    }
+
     // ==================== Redis 实现 ====================
     /**
@@ -20,9 +20,9 @@ public class IotMessageBusProperties {
     /**
      * 消息总线类型
      *
-     * 可选值：local、redis、rocketmq、rabbitmq
+     * 可选值：local、redis、rocketmq、kafka、rabbitmq
      */
     @NotNull(message = "IoT 消息总线类型不能为空")
     private String type = "local";
-}
 \ No newline at end of file
+}
@@ -30,9 +30,13 @@
         <!-- 消息队列相关 -->
         <dependency>
+            <groupId>org.springframework.kafka</groupId>
+            <artifactId>spring-kafka</artifactId>
+            <optional>true</optional>
+        </dependency>
+        <dependency>
             <groupId>org.apache.rocketmq</groupId>
             <artifactId>rocketmq-spring-boot-starter</artifactId>
-            <!-- TODO @芋艿：消息队列，后续可能去掉，默认不使用 rocketmq -->
             <optional>true</optional>
         </dependency>
@@ -12,6 +12,9 @@ spring:
       database: 0 # Redis 数据库索引
       # password: # Redis 密码，如果有的话
       timeout: 30000ms # 连接超时时间
+  # Kafka 配置项，对应 KafkaProperties 配置类
+  kafka:
+    bootstrap-servers: 127.0.0.1:9092 # 指定 Kafka Broker 地址，可以设置多个，以逗号分隔
 --- #################### 消息队列相关 ####################