框架结构调整

iot-dependencies/pom.xml

@@ -75,6 +75,7 @@
         <xercesImpl.version>2.12.2</xercesImpl.version>
         <weixin-java.version>4.5.5.B</weixin-java.version>
         <coap.version>3.11.0</coap.version>
+        <netty.version>4.1.107.Final</netty.version>
     </properties>
 
     <dependencyManagement>
@@ -130,6 +131,12 @@
                 <artifactId>iot-common-core</artifactId>
                 <version>${revision}</version>
             </dependency>
+
+            <dependency>
+                <groupId>io.netty</groupId>
+                <artifactId>netty-all</artifactId>
+                <version>${netty.version}</version>
+            </dependency>
             <!-- 组件 -->
             <dependency>
                 <groupId>com.middle.platform</groupId>

iot-framework/iot-starter-coap/src/main/java/com/middle/platform/coap/core/Bootstrap.java

@@ -1,11 +0,0 @@
-package com.middle.platform.coap.core;
-
-import org.springframework.boot.web.embedded.netty.NettyWebServer;
-
-/**
- * @author xucaiqin
- * @date 2024-04-02 09:16:43
- */
-public interface Bootstrap {
-
-}

iot-framework/iot-starter-kafka/pom.xml

@@ -21,5 +21,10 @@
             <groupId>org.springframework.kafka</groupId>
             <artifactId>spring-kafka</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-autoconfigure</artifactId>
+            <scope>provided</scope>
+        </dependency>
     </dependencies>
 </project>

iot-framework/iot-starter-kafka/src/main/java/com/middle/platform/kafka/config/KafkaConfig.java

@@ -0,0 +1,45 @@
+package com.middle.platform.kafka.config;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.common.serialization.IntegerDeserializer;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
+import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * @author xucaiqin
+ * @date 2024-04-15 15:47:13
+ */
+@Configuration
+public class KafkaConfig {
+    private Map<String, Object> consumerProps(KafkaProperties kafkaProperties) {
+        Map<String, Object> props = new HashMap<>();
+        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getBootstrapServers());
+        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
+        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
+        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "15000");
+        //一次拉取消息数量
+        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "5");
+        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class);
+        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+        return props;
+    }
+
+
+    @Bean("batchContainerFactory")
+    public ConcurrentKafkaListenerContainerFactory<String, Object> listenerContainer(KafkaProperties kafkaProperties) {
+        ConcurrentKafkaListenerContainerFactory<String, Object> container = new ConcurrentKafkaListenerContainerFactory<>();
+        container.setConsumerFactory(new DefaultKafkaConsumerFactory<>(consumerProps(kafkaProperties)));
+        //设置并发量，小于或等于Topic的分区数
+        container.setConcurrency(5);
+        //设置为批量监听
+        container.setBatchListener(true);
+        return container;
+    }
+}

iot-framework/iot-starter-kafka/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports

@@ -0,0 +1 @@
+com.middle.platform.kafka.config.KafkaConfig

iot-framework/iot-starter-rabbitmq/src/main/java/com/middle/platform/rabbitmq/config/RabbitConfig.java

@@ -1,7 +1,12 @@
 package com.middle.platform.rabbitmq.config;
 
 import com.middle.platform.common.core.constant.RabbitConstant;
+import org.springframework.amqp.core.Binding;
+import org.springframework.amqp.core.BindingBuilder;
+import org.springframework.amqp.core.Queue;
 import org.springframework.amqp.core.TopicExchange;
+import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
@@ -20,4 +25,57 @@ public class RabbitConfig {
     public TopicExchange dataExchange() {
         return new TopicExchange(RabbitConstant.TOPIC_EXCHANGE_DATA, true, false);
     }
+    @Bean
+    public SimpleRabbitListenerContainerFactory rabbitFactory(ConnectionFactory connectionFactory) {
+        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
+        factory.setConnectionFactory(connectionFactory);
+        //设置批量
+        factory.setBatchListener(true);
+        factory.setConsumerBatchEnabled(true);//设置BatchMessageListener生效
+        factory.setBatchSize(500);//设置监听器一次批量处理的消息数量
+        return factory;
+    }
+    @Bean
+    public Queue rawQueue() {
+        return new Queue(RabbitConstant.RAW_QUEUE);
+    }
+
+    @Bean
+    public Queue dealQueue() {
+        return new Queue(RabbitConstant.DEAL_QUEUE);
+    }
+
+    @Bean
+    public Queue dbQueue() {
+        return new Queue(RabbitConstant.DB_QUEUE);
+    }
+
+
+    @Bean
+    public Binding dbBinding(TopicExchange dataExchange, Queue dbQueue) {
+        Binding binding = BindingBuilder
+                .bind(dbQueue)
+                .to(dataExchange)
+                .with(RabbitConstant.ROUTING_POINT_DATA_PREFIX + dbQueue.getName());
+        binding.addArgument(RabbitConstant.AUTO_DELETE, true);
+        return binding;
+    }
+    @Bean
+    public Binding dealBinding(TopicExchange dataExchange, Queue dealQueue) {
+        Binding binding = BindingBuilder
+                .bind(dealQueue)
+                .to(dataExchange)
+                .with(RabbitConstant.ROUTING_POINT_DATA_PREFIX + dealQueue.getName());
+        binding.addArgument(RabbitConstant.AUTO_DELETE, true);
+        return binding;
+    }
+    @Bean
+    public Binding rawBinding(TopicExchange dataExchange, Queue rawQueue) {
+        Binding binding = BindingBuilder
+                .bind(rawQueue)
+                .to(dataExchange)
+                .with(RabbitConstant.ROUTING_POINT_DATA_PREFIX + rawQueue.getName());
+        binding.addArgument(RabbitConstant.AUTO_DELETE, true);
+        return binding;
+    }
 }

iot-framework/iot-starter-rabbitmq/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports

@@ -1 +1 @@
-com.middle.platform.common.rabbitmq.config.RabbitConfig
+com.middle.platform.rabbitmq.config.RabbitConfig

iot-module/iot-module-data/iot-module-data-biz/pom.xml

@@ -51,6 +51,10 @@
             <groupId>com.middle.platform</groupId>
             <artifactId>iot-starter-mybatis</artifactId>
         </dependency>
+        <dependency>
+            <groupId>com.middle.platform</groupId>
+            <artifactId>iot-starter-kafka</artifactId>
+        </dependency>
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-actuator</artifactId>

iot-module/iot-module-data/iot-module-data-biz/src/main/java/com/middle/platform/data/biz/config/RabbitmqConfig.java

@@ -1,73 +0,0 @@
-package com.middle.platform.data.biz.config;
-
-import com.middle.platform.common.core.constant.RabbitConstant;
-import org.springframework.amqp.core.Binding;
-import org.springframework.amqp.core.BindingBuilder;
-import org.springframework.amqp.core.Queue;
-import org.springframework.amqp.core.TopicExchange;
-import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
-import org.springframework.amqp.rabbit.connection.ConnectionFactory;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-
-/**
- * @author xucaiqin
- * @date 2024-04-01 20:02:08
- */
-@Configuration
-public class RabbitmqConfig {
-    @Bean
-    public SimpleRabbitListenerContainerFactory rabbitFactory(ConnectionFactory connectionFactory) {
-        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
-        factory.setConnectionFactory(connectionFactory);
-        //设置批量
-        factory.setBatchListener(true);
-        factory.setConsumerBatchEnabled(true);//设置BatchMessageListener生效
-        factory.setBatchSize(500);//设置监听器一次批量处理的消息数量
-        return factory;
-    }
-    @Bean
-    public Queue rawQueue() {
-        return new Queue(RabbitConstant.RAW_QUEUE);
-    }
-
-    @Bean
-    public Queue dealQueue() {
-        return new Queue(RabbitConstant.DEAL_QUEUE);
-    }
-
-    @Bean
-    public Queue dbQueue() {
-        return new Queue(RabbitConstant.DB_QUEUE);
-    }
-
-
-    @Bean
-    public Binding dbBinding(TopicExchange dataExchange, Queue dbQueue) {
-        Binding binding = BindingBuilder
-                .bind(dbQueue)
-                .to(dataExchange)
-                .with(RabbitConstant.ROUTING_POINT_DATA_PREFIX + dbQueue.getName());
-        binding.addArgument(RabbitConstant.AUTO_DELETE, true);
-        return binding;
-    }
-    @Bean
-    public Binding dealBinding(TopicExchange dataExchange, Queue dealQueue) {
-        Binding binding = BindingBuilder
-                .bind(dealQueue)
-                .to(dataExchange)
-                .with(RabbitConstant.ROUTING_POINT_DATA_PREFIX + dealQueue.getName());
-        binding.addArgument(RabbitConstant.AUTO_DELETE, true);
-        return binding;
-    }
-    @Bean
-    public Binding rawBinding(TopicExchange dataExchange, Queue rawQueue) {
-        Binding binding = BindingBuilder
-                .bind(rawQueue)
-                .to(dataExchange)
-                .with(RabbitConstant.ROUTING_POINT_DATA_PREFIX + rawQueue.getName());
-        binding.addArgument(RabbitConstant.AUTO_DELETE, true);
-        return binding;
-    }
-
-}