目录

kafka消费者

kafka消费方式

kafka消费者工作流程

消费者总体工作流程

消费者组原理

消费者重要参数

消费者API

独立消费者案例（订阅主题）

独立消费者案例（订阅分区） 

消费者组案例

生产经验—分区的分配以及再平衡

Range以及再平衡

RoundRobin以及再平衡

Sticky以及再平衡

offset位移

自动提交offset

手动提交offset

指定offset消费

指定时间消费

漏消费和重复消费

生产经验—消费者事务

生产经验—数据积压

kafka消费者

kafka消费方式

（1）pull（拉）模式：

（2）push（推）模式：

Kafka没有采用这种方式，因为由broker决定消息发送速率，很难适应所有消费者的消费速率。

例如推送的速度是50m/s，Consumer1、Consumer2就来不及处理消息。

pull模式不足之处是，如果Kafka没有数据，消费者可能会陷入循环中，一直返回空数据。

kafka消费者工作流程

消费者总体工作流程

消费者组原理

Consumer Group（CG）：消费者组，由多个consumer组成；形成一个消费者组的条件是所有消费者的groupid相同。

（1）消费者组

1）消费者组内每个消费者负责消费不同分区的数据，一个分区只能由一个组内消费者消费。

2）消费者组之间互不影响；如果所有的消费者都属于某个消费者组，即消费者组是逻辑上的一个订阅者。

3）如果向消费组中添加更多的消费者，超过主题分区数量，则有一部分消费者就会闲置，不会接收任何消息。

（2）消费者组初始化流程

coordinator：辅助实现消费者组的初始化和分区的分配。

coordinator节点选择 = groupid的hashcode值%50（ __consumer_offsets的分区数量）

例如： groupid的hashcode值 = 1，1% 50 = 1，那么__consumer_offsets 主题的1号分区，在哪个broker上，就选择这个节点的coordinator作为这个消费者组的老大。消费者组下的所有的消费者提交offset的时候就往这个分区去提交offset。

 （3）消费者组详细消费流程

fetch.min.bytes：每批次最小抓取大小，默认1字节；

fetch.max.wait.ms：一批数据最小值未达到的超时时间，默认为500ms；

fetch.max.bytes：每批次最大抓取大小，默认50m；

Max.poll.records：一次拉取数据返回消息的最大条数，默认500条；

sendFetches：发送消费请求；

FetchedRecords：从队列中抓取数据；

消费者重要参数

消费者API

独立消费者案例（订阅主题）

（1）需求：创建一个独立的消费者，消费first主题中的数据。

注意：在消费者API代码中必须配置消费者组id。命令行启动消费者不填写消费者组id会被自动填写随机的消费者组id。

（2）代码

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 org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Properties;

public class ConsumerAPITopic {
    public static void main(String[] args) {
        //1.创建消费者的配置对象
        Properties properties = new Properties();
        // 2.给消费者配置对象添加参数
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop01:9092,hadoop02:9092");
        // 配置序列化
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());
        // 配置消费者组（组名任意起名） 必须
        properties.put(ConsumerConfig.GROUP_ID_CONFIG,"test");
        //创建消费者对象
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);
        //注册要消费的主题（可以消费多个主题）
        ArrayList<String> topics = new ArrayList<>();
        topics.add("first");
        kafkaConsumer.subscribe(topics);
        //拉取数据打印
        while (true) {
            //设置 1s 中消费一批数据
            ConsumerRecords<String, String> polls = kafkaConsumer.poll(Duration.ofSeconds(1));
            //打印消费到的数据
            for (ConsumerRecord<String, String> poll:polls) {
                System.out.println(poll);
            }
        }
    }
}

（3）测试

1）执行idea的程序

2）启动kafka的first生产者并输入数据

kafka_2.11-2.4.1]# bin/kafka-console-producer.sh 
--broker-list hadoop01:9092 --topic first

3）查看idea控制台的数据

ConsumerRecord(topic = first, partition = 1, leaderEpoch = 0, offset = 16, CreateTime = 1663480002009, serialized key size = -1, serialized value size = 5, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = hello)

ConsumerRecord(topic = first, partition = 2, leaderEpoch = 2, offset = 11, CreateTime = 1663480011502, serialized key size = -1, serialized value size = 5, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = kafka)

独立消费者案例（订阅分区） 

（1）需求：创建一个独立消费者，消费first主题0号分区的数据

（2）代码

ConsumerAPIPartition

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 org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Properties;

public class ConsumerAPIPartition {
    public static void main(String[] args) {
        // 创建kafka生产者的配置对象
        Properties properties = new Properties();
        // 给kafka配置对象添加配置信息
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop01:9092");
        // 配置序列化 必须
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());
        // 配置消费者组（必须），名字可以任意起
        properties.put(ConsumerConfig.GROUP_ID_CONFIG,"test");
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);
        // 消费某个主题的某个分区数据
        ArrayList<TopicPartition> topicPartitions = new ArrayList<>();
        topicPartitions.add(new TopicPartition("first",0));
        kafkaConsumer.assign(topicPartitions);
        while (true) {
            ConsumerRecords<String, String> polls = kafkaConsumer.poll(Duration.ofSeconds(1));
            for (ConsumerRecord<String,String> poll:polls) {
                System.out.println(poll);
            }
        }
    }
}

CustomProducerCallBack

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class CustomProducerCallBack {
    public static void main(String[] args) throws InterruptedException {
        /**
         * 编写回调函数异步发送API代码
         */
        //1.创建kafka生产者环境
        Properties properties = new Properties();
        //2.给kafka配置对象添加配置信息：bootstrap.server
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop01:9092,hadoop02:9092");
        //key.value序列化：key.serializer,value.serializer
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
        //3.创建kafka生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
        //4.调用send方法，发送信息
        for (int a = 0;a < 5; a++) {
            kafkaProducer.send(new ProducerRecord<>("first", "hello mykafka" + a), new Callback() {
                //在kafkaProducer接收到ack时调用，异步调用
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    if (e == null) {
                        System.out.println("主题："+ recordMetadata.topic() + "->"+"分区:"+recordMetadata.partition());
                    } else {
                        //出现异常打印
                        e.printStackTrace();
                    }
                }
            });
            //延迟一会会看到数据发往不同的分区
            Thread.sleep(2);
        }
        //5.关闭环境
        kafkaProducer.close();
    }
}

（3）测试

1）在idea执行ConsumerAPIPartition程序；

2）在idea执行CustomProducerCallBack程序；

3）观察控制台的信息

ConsumerRecord(topic = first, partition = 0, leaderEpoch = 4, offset = 33, CreateTime = 1663497588537, serialized key size = -1, serialized value size = 14, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = hello mykafka0)

ConsumerRecord(topic = first, partition = 0, leaderEpoch = 4, offset = 34, CreateTime = 1663497588573, serialized key size = -1, serialized value size = 14, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = hello mykafka1)
 

消费者组案例

（1）需求：测试同一个主题的分区数据，只能由一个消费者组中的一个消费者消费。

（2）代码

复制一份消费者的代码，在IDEA中同时启动，即可启动同一个消费者组中的两个消费者。

ConsumerAPITopic：消费者1

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 org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Properties;

public class ConsumerAPITopic {
    public static void main(String[] args) {
        //1.创建消费者的配置对象
        Properties properties = new Properties();
        // 2.给消费者配置对象添加参数
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop01:9092,hadoop02:9092");
        // 配置序列化
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());
        // 配置消费者组（组名任意起名） 必须
        properties.put(ConsumerConfig.GROUP_ID_CONFIG,"test");
        //创建消费者对象
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);
        //注册要消费的主题（可以消费多个主题）
        ArrayList<String> topics = new ArrayList<>();
        topics.add("first");
        kafkaConsumer.subscribe(topics);
        //拉取数据打印
        while (true) {
            //设置 1s 中消费一批数据
            ConsumerRecords<String, String> polls = kafkaConsumer.poll(Duration.ofSeconds(1));
            //打印消费到的数据
            for (ConsumerRecord<String, String> poll:polls) {
                System.out.println(poll);
            }
        }
    }
}

ConsumerAPITopic2：消费者2

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 org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Properties;

public class ConsumerAPITopic2 {
    public static void main(String[] args) {
        //1.创建消费者的配置对象
        Properties properties = new Properties();
        // 2.给消费者配置对象添加参数
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop01:9092,hadoop02:9092");
        // 配置序列化
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());
        // 配置消费者组（组名任意起名） 必须
        properties.put(ConsumerConfig.GROUP_ID_CONFIG,"test");
        //创建消费者对象
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);
        //注册要消费的主题（可以消费多个主题）
        ArrayList<String> topics = new ArrayList<>();
        topics.add("first");
        kafkaConsumer.subscribe(topics);
        //拉取数据打印
        while (true) {
            //设置 1s 中消费一批数据
            ConsumerRecords<String, String> polls = kafkaConsumer.poll(Duration.ofSeconds(1));
            //打印消费到的数据
            for (ConsumerRecord<String, String> poll:polls) {
                System.out.println(poll);
            }
        }
    }
}

CustomProducerCallBack：生产者

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class CustomProducerCallBack {
    public static void main(String[] args) throws InterruptedException {
        /**
         * 编写回调函数异步发送API代码
         */
        //1.创建kafka生产者环境
        Properties properties = new Properties();
        //2.给kafka配置对象添加配置信息：bootstrap.server
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop01:9092,hadoop02:9092");
        //key.value序列化：key.serializer,value.serializer
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
        //3.创建kafka生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
        //4.调用send方法，发送信息
        for (int a = 0;a < 5; a++) {
            kafkaProducer.send(new ProducerRecord<>("first", "kafka group" + a), new Callback() {
                //在kafkaProducer接收到ack时调用，异步调用
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    if (e == null) {
                        System.out.println("主题："+ recordMetadata.topic() + "->"+"分区:"+recordMetadata.partition());
                    } else {
                        //出现异常打印
                        e.printStackTrace();
                    }
                }
            });
            //延迟一会会看到数据发往不同的分区
            Thread.sleep(2);
        }
        //5.关闭环境
        kafkaProducer.close();
    }
}

（3）测试

1）启动执行消费者代码

2）启动执行生产者代码，在IDEA控制台即可看到两个消费者在消费不同分区的数据（如果只发生到一个分区，可以在发送时增加延迟代码 Thread.sleep(2);）

3）观察控制台信息

生产经验—分区的分配以及再平衡

一个consumer group中有多个consumer组成，一个topic有多个partition组成，现在的问题是到底由哪个consumer来消费哪个partition的数据。

Kafka有四种主流的分区分配策略： Range、RoundRobin、Sticky、CooperativeSticky。

可以通过配置参数partition.assignment.strategy修改分区的分配策略。默认策略是Range + CooperativeSticky。Kafka可以同时使用多个分区分配策略。

Range以及再平衡

（1）Range分区策略原理

Range是对每个topic而言的。

首先对同一个topic里面的分区按照序号进行排序，并对消费者按照字母顺序进行排序。

例：

假如现在有7个分区，3个消费者，排序后的分区将会是0,1,2,3,4,5,6；消费者排序完之后将会是C0,C1,C2。 7/3 = 2余1 ，除不尽，那么 消费者C0便会多消费1个分区。 若8/3=2余2，除不尽，那么C0和C1分别多消费一个。

通过partitions数/consumer数来决定每个消费者应该消费几个分区。如果除不尽，那么前面几个消费者将会多消费1个分区。

注：如果只是针对1个topic而言，C0消费者多消费1个分区影响不是很大。但是如果有N多个topic，那么针对每个topic，消费者C0都将多消费1个分区，topic越多，C0消费的分区会比其他消费者明显多消费N个分区，容易产生数据倾斜。

（2）Range分区分配策略案例

1）修改主题first为7个分区

bin/kafka-topics.sh --bootstrap-server hadoop01:9092 
--alter --topic first --partitions 7

2）复制几份消费者的代码，在IDEA中同时启动，即可启动同一个消费者组中的几个消费者。

3）启动生产者，发送50条消息，会随机发送到不同的分区。

4）观察数据

（3）Range分区分配再平衡案例

1）停止掉0号消费者，快速重新发送消息观看结果（45s以内，越快越好）。

1号消费者：消费到3、4号分区数据。

2号消费者：消费到5、6号分区数据。

0号消费者的任务会整体被分配到1号消费者或者2号消费者。

说明：0号消费者挂掉后，消费者组需要按照超时时间45s来判断它是否退出，所以需要等待，时间到了45s后，判断它真的退出就会把任务分配给其他broker执行。

2）再次重新发送消息观看结果（45s以后）。

1号消费者：消费到0、1、2、3号分区数据。

2号消费者：消费到4、5、6号分区数据。

说明：消费者0已经被踢出消费者组，所以重新按照range方式分配。

RoundRobin以及再平衡

（1）RoundRobin分区策略原理

RoundRobin针对集群中所有Topic而言。

RoundRobin轮询分区策略，是把所有的partition和所有的consumer都列出来，然后按照hashcode 进行排序，最后通过轮询算法来分配partition给到各个consumer。

（2）RoundRobin分区分配策略案例

1）复制几份消费者的代码，在IDEA中同时启动，即可启动同一个消费者组中的几个消费者。

2）修改分区

properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFI
G, "org.apache.kafka.clients.consumer.RoundRobinAssignor");

3）启动生产者，发送50条消息，会随机发送到不同的分区。

4）观察数据

（3）RoundRobin分区分配再平衡案例

1）停止掉0号消费者，快速重新发送消息观看结果（45s以内，越快越好）。

1 号消费者：消费到2、5号分区数据

2 号消费者：消费到4、1号分区数据

0号消费者的任务会按照RoundRobin的方式，把数据轮询分成0、6和3号分区数据，分别由1号消费者或者2号消费者消费。

说明：0号消费者挂掉后，消费者组需要按照超时时间45s来判断它是否退出，所以需要等待，时间到了45s后，判断它真的退出就会把任务分配给其他broker执行。

2）再次重新发送消息观看结果（45s以后）。

1号消费者：消费到0、2、4、6号分区数据

2号消费者：消费到1、3、5号分区数据

说明：消费者0已经被踢出消费者组，所以重新按照RoundRobin方式分配。

Sticky以及再平衡

粘性分区定义：可以理解为分配的结果带有“粘性的”。即在执行一次新的分配之前，考虑上一次分配的结果，尽量少的调整分配的变动，可以节省大量的开销。

粘性分区是Kafka从0.11.x版本开始引入这种分配策略，首先会尽量均衡的放置分区到消费者上面，在出现同一消费者组内消费者出现问题的时候，会尽量保持原有分配的分区不变化。

修改分区分配策略

ArrayList<String> startegys = new ArrayList<>();
startegys.add("org.apache.kafka.clients.consumer.StickyAssignor");
properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, 
startegys);

Kafka0.9版本之前consumer默认将offset保存在Zookeeper中

从0.9版本开始，consumer默认将offset保存在Kafka一个内置的topic中，该topic为__consumer_offsets

__consumer_offsets主题里面采用key和value的方式存储数据。key是group.id+topic+分区号，value就是当前offset的值。每隔一段时间，kafka内部会对这个topic进行compact，也就是每个group.id+topic+分区号就保留最新数据。

（1）消费offset案例

思想：__consumer_offsets为Kafka中的topic，那就可以通过消费者进行消费。

1）在配置文件config/consumer.properties中添加配置exclude.internal.topics=false，默认是true，表示不能消费系统主题。为了查看该系统主题数据，所以该参数修改为false。

2）采用命令行方式，创建一个新的topic。

bin/kafka-topics.sh --bootstrap-server hadoop01:9092 
--create --topic test2 --partitions 2 --replication-factor 2

3）启动生产者往test2生产数据。

bin/kafka-console-producer.sh --topic test2
--bootstrap-server hadoop01:9092

4）启动消费者消费 atguigu 数据。

bin/kafka-console-consumer.sh --bootstrap-server hadoop01:9092 
--topic test2 --group test

注意：指定消费者组名称，更好观察数据存储位置（key是group.id+topic+分区号）。

5）查看消费者消费主题__consumer_offsets。

bin/kafka-console-consumer.sh --topic __consumer_offsets 
--bootstrap-server hadoop01:9092 
--consumer.config config/consumer.properties 
--formatter "kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageFormatter" 
--from-beginning

自动提交offset

 为了使我们能够专注于自己的业务逻辑，Kafka提供了自动提交offset的功能。 

自动提交offset的相关参数：

enable.auto.commit：是否开启自动提交offset功能，默认是true，消费者会自动周期性地向服务器提交偏移量。

auto.commit.interval.ms：自动提交offset的时间间隔，默认是5s；

（1）消费者自动提交offset 

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 org.apache.kafka.common.serialization.StringSerializer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

public class CondumerAutoOffset {
    public static void main(String[] args) {
        //创建kafka消费者配置类
        Properties properties = new Properties();
        //添加配置参数
        //添加连接
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop01:9092");
        //配置序列化 必须
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
        //配置消费者组
        properties.put(ConsumerConfig.GROUP_ID_CONFIG,"test");
        //提交offset的时间周期1000ms，默认5s
        properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,1000);
        //创建kafka消费者
        KafkaConsumer<String, String> stringStringKafkaConsumer = new KafkaConsumer<>(properties);
        //设置消费主题 形参是列表
        stringStringKafkaConsumer.subscribe(Arrays.asList("first"));
        //消费数据
        while (true) {
            //读取消息
            ConsumerRecords<String, String> polls = stringStringKafkaConsumer.poll(Duration.ofSeconds(1));
            //输出消息
            for (ConsumerRecord<String, String> polls1 :polls) {
                System.out.println(polls1);
            }
        }
    }
}

手动提交offset

虽然自动提交offset十分简单便利，但由于其是基于时间提交的，开发人员难以把握offset提交的时机。因此Kafka还提供了手动提交offset的API。

手动提交offset的方法有两种：分别是commitSync（同步提交）和commitAsync（异步提交）。两者的相同点是，都会将本次提交的一批数据最高的偏移量提交；

不同点是，同步提交阻塞当前线程，一直到提交成功，并且会自动失败重试；而异步提交则没有失败重试机制，故有可能提交失败。

commitSync（同步提交）：必须等待offset提交完毕，再去消费下一批数据。

commitAsync（异步提交） ：发送完提交offset请求后，就开始消费下一批数据了。

（1）同步提交

同步提交offset有失败重试机制，更加可靠，但是由于一直等待提交结果，提交的效率比较低。以下为同步提交offset的示例。

代码：

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 org.apache.kafka.common.serialization.StringSerializer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

public class ConsumerHandOffset {
    public static void main(String[] args) {
        //
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop01:9092");
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
        properties.put(ConsumerConfig.GROUP_ID_CONFIG,"test");
        //是否自动提交 offset
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,false);
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
        consumer.subscribe(Arrays.asList("first"));
        while (true) {
            ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofSeconds(1));
            for (ConsumerRecord<String, String> consumerRecord :consumerRecords) {
                System.out.println(consumerRecord.value());
            }
            //同步提交 offset
            consumer.commitSync();
        }
    }
}

（2）异步提交

虽然同步提交offset更可靠一些，但是由于其会阻塞当前线程，直到提交成功。因此吞吐量会受到很大的影响。因此更多的情况下，会选用异步提交offset的方式。

代码：

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 org.apache.kafka.common.serialization.StringSerializer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

public class ConsumerHandOffsetAsync {
    public static void main(String[] args) {
        //
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop01:9092");
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
        properties.put(ConsumerConfig.GROUP_ID_CONFIG,"test");
        //是否自动提交 offset
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,false);
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
        consumer.subscribe(Arrays.asList("first"));
        //消费数据
        while (true) {
            //读取消息
            ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofSeconds(1));
            //输出消息
            for (ConsumerRecord<String, String> consumerRecord :consumerRecords) {
                System.out.println(consumerRecord.value());
            }
            //异步提交 offset
            consumer.commitAsync();
        }
    }
}

指定offset消费

auto.offset.reset = earliest | latest | none；默认是latest。

当Kafka中没有初始偏移量（消费者组第一次消费）或服务器上不再存在当前偏移量时

1）earliest：自动将偏移量重置为最早的偏移量，--from-beginning。

2）latest（默认值）：自动将偏移量重置为最新偏移量。

3）none：如果未找到消费者组的先前偏移量，则向消费者抛出异常。

4）任意指定offset位移开始消费

代码：

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 org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringSerializer;

import java.time.Duration;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.Properties;

public class ConsumerSeek {
    public static void main(String[] args) {
        //
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop01:9092");
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
        properties.put(ConsumerConfig.GROUP_ID_CONFIG,"test2");
        KafkaConsumer<String, String> stringStringKafkaConsumer = new KafkaConsumer<>(properties);
        //订阅主题
        ArrayList<String> topic = new ArrayList<>();
        topic.add("first");
        stringStringKafkaConsumer.subscribe(topic);

        HashSet<TopicPartition> topicPartitions = new HashSet<>();

        while (topicPartitions.size() == 0) {
            stringStringKafkaConsumer.poll(Duration.ofSeconds(1));
            //获取消费者分区分配信息（有了分区分配信息才能开始消费）
            topicPartitions = (HashSet<TopicPartition>) stringStringKafkaConsumer.assignment();
        }
        //遍历所有分区，并指定offset从1600的位置开始消费
        for (TopicPartition tp:topicPartitions) {
            stringStringKafkaConsumer.seek(tp,1600);
        }
        //消费该主题数据
        while (true) {
            ConsumerRecords<String, String> poll = stringStringKafkaConsumer.poll(Duration.ofSeconds(1));
            for (ConsumerRecord<String, String> polls : poll) {
                System.out.println(polls);
            }
        }
    }
}

注：每次执行完需要修改消费者组名。

指定时间消费

（1）需求：在生产环境中，会遇到最近消费的几个小时数据异常，想重新按照时间消费。

例如：要求按照时间消费前一天的数据。

代码：

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndTimestamp;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringSerializer;

import java.time.Duration;
import java.util.*;

public class ConsumerForTime {
    public static void main(String[] args) {
        //
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"HADOOP01:9092");
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
        properties.put(ConsumerConfig.GROUP_ID_CONFIG,"test2");
        KafkaConsumer<String, String> stringStringKafkaConsumer = new KafkaConsumer<>(properties);
        //订阅主题
        ArrayList<String> topic = new ArrayList<>();
        topic.add("first");
        stringStringKafkaConsumer.subscribe(topic);

        HashSet<TopicPartition> topicPartitions = new HashSet<>();
        while (topicPartitions.size() == 0) {
            stringStringKafkaConsumer.poll(Duration.ofSeconds(1));
            //获取消费者分区分配信息（有了分区分配信息才能开始消费）
            topicPartitions = (HashSet<TopicPartition>) stringStringKafkaConsumer.assignment();
        }
        HashMap<TopicPartition, Long> topicPartitionLongHashMap = new HashMap<>();
        //封装集合存储，每个分区对应一天前的数据
        for (TopicPartition topicPartition : topicPartitions) {
            topicPartitionLongHashMap.put(topicPartition, System.currentTimeMillis() - 1 * 24 * 3600 * 1000);
        }
        //获取从1天前开始消费的每个分区的offset
        Map<TopicPartition, OffsetAndTimestamp> topicPartitionOffsetAndTimestampMap = stringStringKafkaConsumer.offsetsForTimes(topicPartitionLongHashMap);
        //遍历每个分区，对每个分区设置消费时间
        for (TopicPartition topicPartition : topicPartitions) {
            OffsetAndTimestamp offsetAndTimestamp = topicPartitionOffsetAndTimestampMap.get(topicPartition);
            //根据时间指定开始消费的位置
            if (offsetAndTimestamp != null) {
                stringStringKafkaConsumer.seek(topicPartition,offsetAndTimestamp.offset());
            }
        }
        //消费主题数据
        while (true) {
            ConsumerRecords<String, String> poll = stringStringKafkaConsumer.poll(Duration.ofSeconds(1));
            for (ConsumerRecords<String, String> poll1 = poll) {
                System.out.println(poll1);
            }
        }
    }
}

漏消费和重复消费

重复消费：已经消费了数据，但是offset没提交，自动提交offset引起。

1） Consumer每5s提交offset；

2）如果提交offset后的2s，consumer挂了；

漏消费：先提交offset后消费，有可能会造成数据的漏消费。 设置offset为手动提交，当offset被提交时，数据还在内存中未落盘，此时刚好消费者线程被kill掉，那么offset已经提交，但是数据未处理，导致这部分内存中的数据丢失。

1） 提交offset

2） 消费者消费的数据还在内存中，消费者挂掉，导致漏消费