一、死循环无限拉取kafka数据

1.1 整体框架剖析

1、要想从Kafka中读取数据，就要先对消费者信息进行配置

//1、创建消费者配置信息Properties properties = new Properties();//2、给配置信息赋值//2.1 kafka集群信息properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.9.220:9092,192.168.9.221:9092,192.168.9.222:9092");//2.2 开启自动提交offset 提交以后每次offset都在消费的最新位置properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"true");//2.3 自动提交offset延时 1秒钟提交一次properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,"1000");//2.4 key value的反序列化properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");//2.5 消费者组properties.put(ConsumerConfig.GROUP_ID_CONFIG,"base_db_app_210325");

2、消费者基本配置信息完成以后，创建消费者、订阅主题、为了后面的消费

 //创建消费者KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);//订阅主题consumer.subscribe(Collections.singletonList("ticdc-paperfree-monitor"));

3、订阅主题，就相当于已经订阅了kafka中的消息，下一步就是消费。而kafka消费消息的方式是poll拉取，我们这里对kafka中的数据进行消费，上面我们选择了自动提交offset，那么每次offset就是在上一次消费完成以后的最新位置，所以我们接下来的每次消费得到的都是最新未消费的数据！

while (true) {//获取数据ConsumerRecords<String, String> poll = consumer.poll(100);//解析并打印for (ConsumerRecord<String, String> stringStringConsumerRecord : poll) {System.out.println("stringStringConsumerRecord.key() = " + stringStringConsumerRecord.key() + "------" + stringStringConsumerRecord.value() + "----" + stringStringConsumerRecord.topic());}}

1.2 测试

方法一：

1、创建MyConsumer1类，根据上面整体结构的剖析，添加如下代码，并进行测试。

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;import java.util.Collections;
import java.util.Properties;/*** @author potential*/
public class MyConsumer1 {public static void main(String[] args) {//1、创建消费者配置信息Properties properties = new Properties();//2、给配置信息赋值//2.1 kafka集群信息properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.9.220:9092,192.168.9.221:9092,192.168.9.222:9092");//2.2 开启自动提交offsetproperties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"true");//2.3 自动提交offset延时 1秒钟提交一次properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,"1000");//2.4 key value的反序列化properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");//2.5 消费者组properties.put(ConsumerConfig.GROUP_ID_CONFIG,"base_db_app_210325");//创建消费者KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);//订阅主题consumer.subscribe(Collections.singletonList("ticdc-paperfree-monitor"));while (true) {//获取数据ConsumerRecords<String, String> poll = consumer.poll(100);//解析并打印for (ConsumerRecord<String, String> stringStringConsumerRecord : poll) {System.out.println("stringStringConsumerRecord.key() = " + stringStringConsumerRecord.key() + "------" + stringStringConsumerRecord.value() + "----" + stringStringConsumerRecord.topic());}}
//        //关闭连接
//        consumer.close();}
}

方法二：
2、创建MyConsumer2类，根据上面整体结构的剖析，添加如下代码，并进行测试。

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.serialization.IntegerDeserializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.util.*;/*** @author potential*/
public class MyConsumer2 {public static void main(String[] args) {//配置必要的参数//准备一个map集合放置参数Map<String, Object> config = new HashMap<String, Object>();//bootserversconconfig.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.9.220:9092,192.168.9.221:9092,192.168.9.222:9092");//开启自动提交offsetconfig.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"true");//自动提交offset延时 1秒钟提交一次config.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,"1000");//valuedeserilizerconfig.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class);config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG , StringDeserializer.class);//groupidconfig.put(ConsumerConfig.GROUP_ID_CONFIG, "base_db_app_210325");//如果找不到偏移量，设置earliest,则从最新消费开始,也就是消费者一开始最新消费的时候//一定要注意顺序，读取时候的顺序会影响config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");//        //此处是把消费者的偏移量重置到生产者最顶端
//        Map<TopicPartition, OffsetAndMetadata> hashMaps = new HashMap<TopicPartition, OffsetAndMetadata>();
//        hashMaps.put(new TopicPartition("ticdc-paperfree-monitor", 0), new OffsetAndMetadata(129));//消费者KafkaConsumer<String, String> consumer = new KafkaConsumer<String,String>(config);
//        //放置刚刚设置的偏移量
//        consumer.commitSync(hashMaps);//先订阅后消费consumer.subscribe(Arrays.asList("ticdc-paperfree-monitor"));//        // 批量从主题的分区拉取消息
//        //final ConsumerRecords<Integer, String> consumerRecords = consumer.poll(3_000);
//        ConsumerRecords<String, String> consumerRecords = consumer.poll(3000);while (true) {//获取数据ConsumerRecords<String, String> poll = consumer.poll(100);//解析并打印for (ConsumerRecord<String, String> stringStringConsumerRecord : poll) {System.out.println("stringStringConsumerRecord.key() = " + stringStringConsumerRecord.key() + "------" + stringStringConsumerRecord.value() + "----" + stringStringConsumerRecord.topic());}}//
//        //遍历本次从主题的分区拉取的批量消息 这里是将整个分区中的全部数据都拉出来了
//        consumerRecords.forEach(new java.util.function.Consumer<ConsumerRecord<Integer, String>>() {
//            @Override
//            public void accept(ConsumerRecord<Integer, String> consumerRecord) {
//                System.out.println(
//                        consumerRecord.topic() +"\t"
//                                +consumerRecord.offset() + "\t"
//                                +consumerRecord.key() +"\t"
//                                +consumerRecord.value()+"\t"
//                );
//            }
//        });
//        consumer.close();}
}

注意：
方式一、方式二只是写法上的不同，整体架构都是一样的，任选其一来写即可。
在这里插入图片描述至此，从kafka中读取最新数据的流程就全部结束了。

二、@KafkaListener注解实现监听kafka数据

1、导入依赖

【我这里SpringBoot版本是2.2.13】

<dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>2.3.1</version></dependency><dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId><version>2.3.7.RELEASE</version></dependency>

注意：
1、springboot +2、kafka-clients +3、spring-kafka（在下图中体现为Sprig for Apache Kafka Version）这三个要注意版本对应。具体对应情况如下图所示：
在这里插入图片描述 2、配置文件
在application.yml文件中添加如下内容：

spring:kafka:consumer:bootstrap-servers: 192.168.9.220:9092,192.168.9.221:9092,192.168.9.222:9092 #集群信息producer: #生产者retries: 0 #设置大于0的值，则客户端将发送失败的记录重新发送batch-size: 16384 #批量大小buffer-momory: 33554432 #生产端缓冲区大小acks: 1 #应答级别#指定消息key和消息体的解编码方式  序列化与反序列化key- key-serializer: org.apache.kafka.common.serialization.StringSerializervalue-serializer: org.apache.kafka.common.serialization.StringSerializconsumer:group-id: base_db_app_210325enable-auto-comnit: true #是否自动提交offsetauto-offset-reset: latest #重置为分区中最新的offset(消费者分区中新产生的数据)key-deserializer: org.apache.kafka.common.serialization.StringDeserializervalue-deserializer: org.apache.kafka.common.serialization.StringDeserializertopic-name: ticdc-paperfree-monitor #主题listener:ack-mode: manual_immediate

3、创建MyConsumer类，添加如下内容：

 @KafkaListener(id = "test1", topics = "ticdc-paperfree-monitor")//这里id是随意起的，我这里叫test1，我这里主题直接写死，取ticdc-paperfree-monitor这个主题下的数据，也可以${},动态获取主题名称，group_idpublic void listen(ConsumerRecord<String, String> record) {//从Kafka中读取到的数据System.out.println("topic:" + record.topic());System.out.println("value:" + record.value());}