更新

ruoyi-receive-server/src/main/java/com/ruoyi/receive/kafka/config/Consumer.java

@@ -0,0 +1,53 @@
+package com.ruoyi.receive.kafka.config;
+
+import com.sun.corba.se.internal.CosNaming.BootstrapServer;
+import lombok.AllArgsConstructor;
+import net.sf.jsqlparser.statement.select.Top;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.springframework.context.annotation.Bean;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.PostConstruct;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Properties;
+
+/**
+ * 消费者
+ */
+
+@Component
+public class Consumer {
+
+    private final KafkaConsumerConfig kafkaConsumerConfig;
+
+    private final ConsumerCustomConfig consumerCustomConfig;
+
+    public Consumer(KafkaConsumerConfig kafkaConsumerConfig, ConsumerCustomConfig consumerCustomConfig) {
+        this.kafkaConsumerConfig = kafkaConsumerConfig;
+        this.consumerCustomConfig = consumerCustomConfig;
+    }
+
+
+
+
+    @Bean
+    public KafkaConsumer<String,String> consumerInit(){
+        //创建kafka消费者配置
+        Properties properties = new Properties();
+        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,kafkaConsumerConfig.consumerConfigs() );
+        properties.put(ConsumerConfig.GROUP_ID_CONFIG,consumerCustomConfig.getGroup());
+        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
+        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());
+        //创建kafka消费者
+        KafkaConsumer<String,String> consumer= new KafkaConsumer<>(properties);
+        //订阅主题
+        consumer.subscribe(Collections.singletonList(consumerCustomConfig.getTopic()));
+        return consumer;
+    }
+
+
+
+}