深入分析kafka的消费者配置原理

叁歲伎倆 2023-10-02 18:49 1阅读 0赞

今天深入分析kafka的消费者配置原理

一、消费者入门：

1、消费者群组：

Kafka  里消费者从属于消费者群组，一个群组里的消费者订阅的都是同一个主题，每个消费者接收主题一部分分区的消息。

![d98ebe10c0ce4beaaa84a92a5a3af295.png][]

如上图，主题  T  有  4  个分区，群组中只有一个消费者，则该消费者将收到主题  T1  全部  4  个分区的消息。

![watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBAbmFuZGFvMTU4_size_8_color_FFFFFF_t_70_g_se_x_16][]

如上图，在群组中增加一个消费者  2 ，那么每个消费者将分别从两个分区接收消息，上图中就表现为消费者  1  接收分区  1  和分区  3  的消息，消费者  2 接收分区 2  和分区  4  的消息。

![acc84722e46448fb98b7a654bc42b1fd.png][]

如上图，在群组中有  4  个消费者，那么每个消费者将分别从  1  个分区接收消息。

![watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBAbmFuZGFvMTU4_size_8_color_FFFFFF_t_70_g_se_x_16 1][]

但是，当我们增加更多的消费者，超过了主题的分区数量，就会有一部分的消费者被闲置，不会接收到任何消息。 往消费者群组里增加消费者是进行横向伸缩能力的主要方式。所以我们有必要为主题设定合适规模的分区，在负载均衡的时候可以加入更多的消费者。但是要记住，一个群组里消费者数量超过了主题的分区数量，多出来的消费者是没有用处的。 如果是多个应用程序，需要从同一个主题中读取数据，只要保证每个应用程序有自己的消费者群组就行了。

具体实现如图，先建立一个  2  分区的主题：

![watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBAbmFuZGFvMTU4_size_20_color_FFFFFF_t_70_g_se_x_16][]

2、消费者配置：

消费者有很多属性可以设置，大部分都有合理的默认值，无需调整。有些参数可能对内存使用，性能和可靠性方面有较大影响。可以参考如下代码：

public static void main(String[] args) {
            //TODO 消费者三个属性必须指定(broker地址清单、key和value的反序列化器)
            Properties properties = new Properties();
            properties.put("bootstrap.servers","127.0.0.1:9092");
            properties.put("key.deserializer", StringDeserializer.class);
            properties.put("value.deserializer", StringDeserializer.class);
            //TODO 群组并非完全必须
            properties.put(ConsumerConfig.GROUP_ID_CONFIG,"test1");
    
            //TODO 更多消费者配置（重要的）
            properties.put("auto.offset.reset","latest"); //消费者在读取一个没有偏移量的分区或者偏移量无效的情况下，如何处理
            properties.put("enable.auto.commit",true); // 表明消费者是否自动提交偏移 默认值true
            properties.put("max.poll.records",500); // 控制每次poll方法返回的的记录数量 默认值500
            //分区分配给消费者的策略。系统提供两种策略。默认为Range
            properties.put("partition.assignment.strategy",Collections.singletonList(RangeAssignor.class));
    
    
    
            KafkaConsumer<String,String> consumer = new KafkaConsumer<String, String>(properties);
            try {
                //TODO 消费者订阅主题（可以多个）
                consumer.subscribe(Collections.singletonList(BusiConst.HELLO_TOPIC));
                while(true){
                    //TODO 拉取（新版本）
                    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
                    for(ConsumerRecord<String, String> record:records){
                        System.out.println(String.format("topic:%s,分区：%d,偏移量：%d," + "key:%s,value:%s",record.topic(),record.partition(),
                                record.offset(),record.key(),record.value()));
                        //do my work
                        //打包任务投入线程池
                        // ex
                    }
                }
            } finally {
                consumer.close();
            }
    
        }

auto.offset.reset

消费者在读取一个没有偏移量的分区或者偏移量无效的情况下，如何处理。默认值是  latest ，从最新的记录开始读取，另一个值是  earliest ，表示消费者从起始位置读取分区的记录。

**注意**：如果是消费者在读取一个没有偏移量的分区或者偏移量无效的情况（因消费者长时间失效，包含的偏移量记录已经过时并被删除）下，默认值是 latest 的话，消费者将从最新的记录开始读取数据（**在消费者启动之后生成的记录**），可以先启动生产者，再启动消费者，观察到这种情况。

enable .auto.commit

默认值  true ，表明消费者是否自动提交偏移。为了尽量避免重复数据和数据丢失，可以改为  false ，自行控制何时提交。

partition.assignment.strategy

分区分配给消费者的策略。系统提供两种策略。默认为  Range 。允许自定义策略。

*Range*

把主题的连续分区分配给消费者。（如果分区数量无法被消费者整除、第一个消费者会分到更多分区）

*RoundRobin*

把主题的分区循环分配给消费者。

![watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBAbmFuZGFvMTU4_size_20_color_FFFFFF_t_70_g_se_x_16 1][]

*自定义策略*

extends  类  AbstractPartitionAssignor ，然后在消费者端增加参数：

properties.put(ConsumerConfig.PARTITION\_ASSIGNMENT\_STRATEGY\_CONFIG,  类 .class.getName()); 即可。

max.poll.records

控制每次  poll  方法返回的的记录数量。

fetch.min.bytes

每次  fetch  请求时， server  应该返回的最小字节数。如果没有足够的数据返回，请求会等待，直到足够的数据才会返回。缺省为  1  个字节。多消费者下，可以设大这个值，以降低 broker  的工作负载。

fetch.wait.max.ms

如果没有足够的数据能够满足  fetch.min.bytes ，则此项配置是指在应答  fetch  请求之前， server  会阻塞的最大时间。缺省为  500  个毫秒。和上面的fetch.min.bytes 结合起来，要么满足数据的大小，要么满足时间，就看哪个条件先满足。

max.partition.fetch.bytes

指定了服务器从每个分区里返回给消费者的最大字节数，默认  1MB 。假设一个主题有  20  个分区和  5  个消费者，那么每个消费者至少要有  4MB  的可用内存来接收记录，而且一旦有消费者崩溃，这个内存还需更大。注意，这个参数要比服务器的 message.max.bytes  更大，否则消费者可能无法读取消息。

session.timeout.ms

如果  consumer  在这段时间内没有发送心跳信息，则它会被认为挂掉了。默认  3  秒。

client.id

当向  server  发出请求时，这个字符串会发送给  server 。目的是能够追踪请求源头，以此来允许  ip/port  许可列表之外的一些应用可以发送信息。这项应用可以设置任意字符串，因为没有任何功能性的目的，除了记录和跟踪。

receive.buffer.bytes  和  send.buffer.bytes

指定  TCP socket  接受和发送数据包的缓存区大小。如果它们被设置为 \-1 ，则使用操作系统的默认值。如果生产者或消费者处在不同的数据中心，那么可以适当增大这些值，因为跨数据中心的网络一般都有比较高的延迟和比较低的带宽。

二、基础概念

1、订阅

创建消费者后，使用  subscribe() 方法订阅主题，这个方法接受一个主题列表为参数，也可以接受一个正则表达式为参数；正则表达式同样也匹配多个主题。如果新创建了新主题，并且主题名字和正则表达式匹配，那么会立即触发一次再均衡，消费者就可以读取新添加的主题。比如，要订阅所有和 test 相关的主题，可以 subscribe(“tets.\*”)

2、轮询

为了不断的获取消息，我们要在循环中不断的进行轮询，也就是不停调用  poll  方法。

poll  方法的参数为超时时间，控制  poll  方法的阻塞时间，它会让消费者在指定的毫秒数内一直等待  broker  返回数据。 poll  方法将会返回一个记录（消息）列表，每一条记录都包含了记录所属的主题信息，记录所在分区信息，记录在分区里的偏移量，以及记录的键值对。

poll  方法不仅仅只是获取数据，在新消费者第一次调用时，它会负责查找群组，加入群组，接受分配的分区。如果发生了再均衡，整个过程也是在轮

询期间进行的。

3、提交偏移量

当我们调用  poll  方法的时候， broker  返回的是生产者写入  Kafka  但是还没有被消费者读取过的记录，消费者可以使用  Kafka  来追踪消息在分区里的位置，我们称之为**偏移量** 。消费者更新自己读取到哪个消息的操作，我们称之为 **提交** 。

消费者是如何提交偏移量的呢？消费者会往一个叫做 \_consumer\_offset  的特殊主题发送一个消息， 里面会包括每个分区的偏移量。

三、核心概念：

1、多线程安全：

KafkaConsumer  的实现 **不是** 线程安全的，所以我们在多线程的环境下， 使用  KafkaConsumer  的实例要小心，应该每个消费数据的线程拥有自己的 KafkaConsumer 实例。

2、群组协调：

消费者要加入群组时，会向 群组协调器 发送一个  JoinGroup  请求，第一个加入群主的消费者成为群主，群主会获得群组的成员列表，并负责给每一个消费者分配分区。分配完毕后，群主把分配情况发送给 群组协调器 ，协调器再把这些信息发送给所有的消费者，每个消费者只能看到自己的分配信息， 只有群主知道群组里所有消费者的分配信息。群组协调的工作会在消费者发生变化( 新加入或者掉线 ) ，主题中分区发生了变化（增加）时发生。

![watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBAbmFuZGFvMTU4_size_16_color_FFFFFF_t_70_g_se_x_16][]

3、分区再均衡

当消费者群组里的消费者发生变化，或者主题里的分区发生了变化，都会导致再均衡现象的发生。从前面的知识中，我们知道， Kafka  中，存在着消费者对分区所有权的关系，这样无论是消费者变化，比如增加了消费者，新消费者会读取原本由其他消费者读取的分区，消费者减少，原本由它负责的分区要由其他消费者来读取，增加了分区，哪个消费者来读取这个新增的分区，这些行为，都会导致分区所有权的变化，这种变化就被称为**再均衡** 。再均衡对 Kafka  很重要，这是消费者群组带来高可用性和伸缩性的关键所在。 不过一般情况下，尽量减少再均衡，因为再均衡期间，消费者是无法读 取消息的，会造成整个群组一小段时间的不可用 。

消费者通过向称为群组协调器的  broker （不同的群组有不同的协调器）发送心跳来维持它和群组的从属关系以及对分区的所有权关系。如果消费者长时间不发送心跳，群组协调器认为它已经死亡，就会触发一次再均衡。 在 0.10.1  及以后的版本中，心跳由单独的线程负责，相关的控制参数为  max.poll.interval.ms 。

4、消费安全问题：

一般情况下，我们调用  poll  方法的时候， broker  返回的是生产者写入  Kafka  同时  kafka  的消费者提交偏移量，这样可以确保消费者消息消费不丢失也 不重复，所以一般情况下 Kafka  提供的原生的消费者是安全的，但是事情会这么完美吗？答案显然不是的！

四、提交偏移量导致的问题

消费者是如何提交偏移量的呢？消费者会往一个叫做 \_consumer\_offset  的特殊主题发送一个消息， 里面会包括每个分区的偏移量。发生了再均衡之后，消费者可能会被分配新的分区，为了能够继续工作，消费者者需要读取每个分区最后一次提交的偏移量，然后从指定的地方，继续做处理。

1 ）如果提交的偏移量小于消费者实际处理的最后一个消息的偏移量，处于两个偏移量之间的消息会被重复处理。

2 ）如果提交的偏移量大于客户端处理的最后一个消息的偏移量 , 那么处于两个偏移量之间的消息将会丢失。

![watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBAbmFuZGFvMTU4_size_20_color_FFFFFF_t_70_g_se_x_16 2][]

所以, 处理偏移量的方式对客户端会有很大的影响 。KafkaConsumer API 提供了很多种方式来提交偏移量 。

1、自动提交

最简单的提交方式是让消费者自动提交偏移量。 如果  enable.auto.comnit  被设为  true ，消费者会自动把从  poll() 方法接收到的 **最大** 偏移量提交上去。 提交时间间隔由 auto.commit.interval.ms  控制，默认值是  5s 。自动提交是在轮询里进行的，消费者每次在进行轮询时会检査是否该提交偏移量了，如果是， 那么就会提交从上一次轮询返回的偏移量。 不过, 在使用这种简便的方式之前 , 需要知道它将会带来怎样的结果。 假设我们仍然使用默认的 5s  提交时间间隔 ,  在最近一次提交之后的  3s  发生了再均衡，再均衡之后 , 消费者从最后一次提交的偏移量位置开始读取消息。 这个时候偏移量已经落后了 3s ，所以在这  3s  内到达的消息会被重复处理。可以通过修改提交时间间隔来更频繁地提交偏移量 ,  减小可能出现重复消息的时间窗,  不过这种情况是无法完全避免的 。 在使用自动提交时, 每次调用轮询方法都会把上一次调用返回的最大偏移量提交上去 , 它并不知道具体哪些消息已经被处理了 , 所以在再次调用之前最好确保所有当前调用返回的消息都已经处理完毕(enable.auto.comnit  被设为  true  时，在调用  close() 方法之前也会进行自动提交 ) 。一般情况下不会有什么 问题, 不过在处理异常或提前退出轮询时要格外小心。

自动提交虽然方便 , 但是很明显是一种基于时间提交的方式 , 不过并没有为我们留有余地来避免重复处理消息。

2、手动提交（同步）

我们通过控制偏移量提交时间来消除丢失消息的可能性，并在发生再均衡时减少重复消息的数量。消费者 API 提供了另一种提交偏移量的方式，开发 者可以在必要的时候提交当前偏移量,而不是基于时间间隔。 把 auto.commit. offset 设为 false，自行决定何时提交偏移量。使用 commitsync()提交偏移量最简单也最可靠。这个方法会提交由 poll()方法返回的最 新偏移量，提交成功后马上返回,如果提交失败就抛出异常。

注意：  commitsync() 将会提交由  poll() 返回的最新偏移量 , 所以在处理完所有记录后要确保调用了  commitsync() ，否则还是会有丢失消息的风险。如果发生了再均衡, 从最近批消息到发生再均衡之间的所有消息都将被重复处理。 只要没有发生不可恢复的错误，commitSync （）方法会阻塞，会一直尝试直至提交成功，如果失败，也只能记录异常日志。

具体代码参考：

public static void main(String[] args) {
            /*消息消费者*/
            Properties properties = KafkaConst.consumerConfig("CommitSync",
                    StringDeserializer.class,
                    StringDeserializer.class);
            //TODO 取消自动提交
            /*取消自动提交*/
            properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,false);
    
            KafkaConsumer<String,String> consumer
                    = new KafkaConsumer<String, String>(properties);
            try {
                consumer.subscribe(Collections.singletonList(
                        BusiConst.CONSUMER_COMMIT_TOPIC));
                while(true){
                    ConsumerRecords<String, String> records
                            = consumer.poll(Duration.ofMillis(500));
                    for(ConsumerRecord<String, String> record:records){ //100个 100~ 200
                        System.out.println(String.format(
                                "主题：%s，分区：%d，偏移量：%d，key：%s，value：%s",
                                record.topic(),record.partition(),record.offset(),
                                record.key(),record.value()));
                        //do our work
    
                    }
                    //开始事务
                    //读业务写数据库-
                    //偏移量写入数据库
                    //TODO 同步提交（这个方法会阻塞）
                    consumer.commitSync(); //offset =200  max
    
                    consumer.commitSync(); //offset =200  max
                }
            } finally {
                consumer.close();
            }
        }

3、异步提交

手动提交时，在  broker  对提交请求作出回应之前，应用程序会一直阻塞。这时我们可以使用异步提交  API ，我们只管发送提交请求，无需等待  broker 的响应。

在成功提交或碰到无法恢复的错误之前 , commitsync() 会一直重试 , 但是  commitAsync  不会。它之所以不进行重试 , 是因为在它收到服务器响应的时候 , 可能有一个更大的偏移量已经提交成功。

假设我们发出一个请求用于提交偏移量  2000,, 这个时候发生了短暂的通信问题 , 服务器收不到请求 , 自然也不会作出任何响应。与此同时 , 我们处理了另外一批消息, 并成功提交了偏移量  3000 。如果  commitAsync() 重新尝试提交偏移量  2000, 它有可能在偏移量  3000  之后提交成功。这个时候如果发生再均衡 , 就会出现重复消息。

commitAsync() 也支持回调 , 在  broker  作出响应时会执行回调。回调经常被用于记录提交错误或生成度量指标。 回调具体代码参考：

public static void main(String[] args) {
            /*消息消费者*/
            Properties properties = KafkaConst.consumerConfig(
                    "CommitAsync",
                    StringDeserializer.class,
                    StringDeserializer.class);
            //TODO  取消自动提交
            /*取消自动提交*/
            properties.put("enable.auto.commit",false);
    
            KafkaConsumer<String,String> consumer
                    = new KafkaConsumer<String, String>(properties);
            try {
                consumer.subscribe(Collections.singletonList(
                        BusiConst.CONSUMER_COMMIT_TOPIC));
                while(true){
                    ConsumerRecords<String, String> records
                            = consumer.poll(Duration.ofMillis(500));
                    for(ConsumerRecord<String, String> record:records){
                        System.out.println(String.format(
                                "主题：%s，分区：%d，偏移量：%d，key：%s，value：%s",
                                record.topic(),record.partition(),record.offset(),
                                record.key(),record.value()));
                        //do our work
                    }
                    //TODO 异步提交偏移量
                    consumer.commitAsync();
                    /*允许执行回调*/
                    consumer.commitAsync(new OffsetCommitCallback() {
                        public void onComplete(
                                Map<TopicPartition, OffsetAndMetadata> offsets,
                                Exception exception) {
                            if(exception!=null){
                                System.out.print("Commmit failed for offsets ");
                                System.out.println(offsets);
                                exception.printStackTrace();
                            }
                        }
                    });
    
                }
            } finally {
                consumer.close();
            }
        }

4、同步异步组合

因为同步提交一定会成功、异步可能会失败，所以一般的场景是同步和异步一起来做。 一般情况下, 针对偶尔出现的提交失败 , 不进行重试不会有太大问题，因为如果提交失败是因为临时问题导致的 , 那么后续的提交总会有成功的。但如果 这是发生在关闭消费者或 再均衡前的最后一次提交, 就要确保能够提交成功。 因此, 在消费者关闭前一般会组合使用  commitAsync() 和  commitsync() 。具体使用，参见代码如下：

public static void main(String[] args) {
            /*消息消费者*/
            Properties properties = KafkaConst.consumerConfig("SyncAndAsync",
                    StringDeserializer.class,
                    StringDeserializer.class);
            //TODO 取消自动提交
            /*取消自动提交*/
            properties.put("enable.auto.commit",false);
    
            KafkaConsumer<String,String> consumer
                    = new KafkaConsumer<String, String>(properties);
            try {
                consumer.subscribe(Collections.singletonList(
                        BusiConst.CONSUMER_COMMIT_TOPIC));
                while(true){
                    ConsumerRecords<String, String> records
                            = consumer.poll(Duration.ofMillis(500));
                    for(ConsumerRecord<String, String> record:records){
                        System.out.println(String.format(
                                "主题：%s，分区：%d，偏移量：%d，key：%s，value：%s",
                                record.topic(),record.partition(),record.offset(),
                                record.key(),record.value()));
                        //do our work
                    }
                    //TODO 异步提交
                    consumer.commitAsync();
                }
            } catch (CommitFailedException e) {
                System.out.println("Commit failed:");
                e.printStackTrace();
            } finally {
                try {
                    //TODO 为了万不一失，需要同步提交下
                    consumer.commitSync();
                } finally {
                    consumer.close();
                }
            }
        }

5、特定提交

在我们前面的提交中，提交偏移量的频率与处理消息批次的频率是一样的。但如果想要更频繁地提交该怎么办 ?

如果  poll() 方法返回一大批数据 , 为了避免因再均衡引起的重复处理整批消息 , 想要在批次中间提交偏移量该怎么办 ? 这种情况无法通过调用 commitSync()或  commitAsync() 来实现，因为它们只会提交最后一个偏移量 , 而此时该批次里的消息还没有处理完。 消费者  API  允许在调用  commitsync() 和  commitAsync() 方法时传进去希望提交的分区和偏移量的  map 。假设我们处理了半个批次的消息 , 最后一个来自 主题“customers ”，分区  3  的消息的偏移量是  5000 ，你可以调用  commitsync() 方法来提交它。不过，因为消费者可能不只读取一个分区 , 因为我们需要跟踪所有分区的偏移量, 所以在这个层面上控制偏移量的提交会让代码变复杂。

具体使用，参见代码如下：

public static void main(String[] args) {
            /*消息消费者*/
            Properties properties = KafkaConst.consumerConfig(
                    "CommitSpecial",
                    StringDeserializer.class,
                    StringDeserializer.class);
            //TODO 必须做
            /*取消自动提交*/
            properties.put("enable.auto.commit",false);
    
            KafkaConsumer<String,String> consumer
                    = new KafkaConsumer<String, String>(properties);
            Map<TopicPartition, OffsetAndMetadata> currOffsets
                    = new HashMap<TopicPartition, OffsetAndMetadata>();
            int count = 0;
            try {
                consumer.subscribe(Collections.singletonList(
                        BusiConst.CONSUMER_COMMIT_TOPIC));
                while(true){
                    ConsumerRecords<String, String> records
                            = consumer.poll(Duration.ofMillis(500));
                    for(ConsumerRecord<String, String> record:records){
                        System.out.println(String.format(
                                "主题：%s，分区：%d，偏移量：%d，key：%s，value：%s",
                                record.topic(),record.partition(),record.offset(),
                                record.key(),record.value()));
                        currOffsets.put(new TopicPartition(record.topic(),record.partition()),
                                new OffsetAndMetadata(record.offset()+1,"no meta"));
                        if(count%11==0){
                            //TODO 这里特定提交（异步方式，加入偏移量），每11条提交一次
                            consumer.commitAsync(currOffsets,null);
                        }
                        count++;
                    }
                }
            } finally {
                //TODO 在关闭前最好同步提交一次偏移量
                consumer.commitSync();
                consumer.close();
            }
        }

五、分区再均衡

![watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBAbmFuZGFvMTU4_size_20_color_FFFFFF_t_70_g_se_x_16 3][]

1、在均衡监听器

在提交偏移量一节中提到过 , 消费者在退出和进行分区再均衡之前 , 会做一些清理工作比如，提交偏移量、关闭文件句柄、数据库连接等。 在为消费者分配新分区或移除旧分区时, 可以通过消费者  API  执行一些应用程序代码，在调用  subscribe() 方法时传进去一个  ConsumerRebalancelistener

实例就可以了。

ConsumerRebalancelistener  有两个需要实现的方法。

1) public void onPartitionsRevoked( Collection< TopicPartition> partitions) 方法会在再均衡开始之前和消费者停止读取消息之后被调用。如果在这里提交偏移量，下一个接管分区的消费者就知道该从哪里开始读取了

2) public void onPartitionsAssigned( Collection< TopicPartition> partitions) 方法会在重新分配分区之后和消费者开始读取消息之前被调用。

2、从特定偏移量开始记录

到目前为止 , 我们知道了如何使用  poll() 方法从各个分区的最新偏移量处开始处理消息。 不过, 有时候我们也需要从特定的偏移量处开始读取消息。 如果想从分区的起始位置开始读取消息, 或者直接跳到分区的末尾开始读取消息 , 可以使  seekToBeginning(Collection<TopicPartition> tp) 和

seekToEnd( Collection<TopicPartition>tp) 这两个方法。 不过,Kaka  也为我们提供了用于查找特定偏移量的  API 。它有很多用途 , 比如向后回退几个消息或者向前跳过几个消息 ( 对时间比较敏感的应用程序在处理滞后的情况下希望能够向前跳过若干个消息) 。在使用  Kafka  以外的系统来存储偏移量时 , 它将给我们带来更大的惊喜 \-- 让消息的业务处理和偏移量的提交变得一致。 试想一下这样的场景: 应用程序从  Kaka  读取事件 ( 可能是网站的用户点击事件流 ), 对它们进行处理 ( 可能是使用自动程序清理点击操作并添加会话信息 ), 然后把结果保存到数据库。假设我们真的不想丢失任何数据, 也不想在数据库里多次保存相同的结果。 我们可能会，毎处理一条记录就提交一次偏移量。尽管如此, 在记录被保存到数据库之后以及偏移量被提交之前 , 应用程序仍然有可能发生崩溃 , 导致重复处理数据, 数据库里就会出现重复记录。 如果保存记录和偏移量可以在一个原子操作里完成 , 就可以避免出现上述情况。记录和偏移量要么都被成功提交 , 要么都不提交。如果记录是保存在数 据库里而偏移量是提交到Kafka 上 , 那么就无法实现原子操作不过 , 如果在同一个事务里把记录和偏移量都写到数据库里会怎样呢 ? 那么我们就会知道记录和偏移量要么都成功提交, 要么都没有 , 然后重新处理记录。 现在的问题是: 如果偏移量是保存在数据库里而不是  Kafka  里 , 那么消费者在得到新分区时怎么知道该从哪里开始读取 ? 这个时候可以使用  seek() 方法。 在消费者启动或分配到新分区时, 可以使用  seck() 方法查找保存在数据库里的偏移量。我们可以使用使用  Consumer Rebalancelistener  和  seek() 方法确保我们是从数据库里保存的偏移量所指定的位置开始处理消息的。

六、独立消费者

到目前为止 , 我们讨论了消费者群组 , 分区被自动分配给群组里的消费者 , 在群组里新增或移除消费者时自动触发再均衡。不过有时候可能只需要一个消费者从一个主题的所有分区或者某个特定的分区读取数据。这个时候就不需要消费者群组和再均衡了, 只需要把主题或者分区分配给消费者 , 然后开始读取消息并提交偏移量。 如果是这样的话, 就不需要订阅主题 , 取而代之的是为自己分配分区。一个消费者可以订阅主题 ( 并加入消费者群组 ), 或者为自己分配分区 , 但不能同时做

这两件事情。 独立消费者相当于自己来分配分区，但是这样做的好处是自己控制，但是就没有动态的支持了，包括加入消费者（分区再均衡之类的），新增分区，这些都需要代码中去解决，所以一般情况下不推荐使用。

七、优雅退出

如果确定要退出循环 , 需要通过另一个线程调用  consumer. wakeup() 方法。如果循环运行在主线程里 , 可以在  ShutdownHook  里调用该方法。要记住 , consumer. wakeup()是消费者唯一一个可以从其他线程里安全调用的方法。调用  consumer. wakeup() 可以退出  poll(), 并抛出  WakeupException  异常。我们不需要处理 Wakeup Exception, 因为它只是用于跳出循环的一种方式。不过 , 在退出线程之前调用  consumer.close() 是很有必要的 , 它会提交任何还没有提交的东西, 并向群组协调器发送消息 , 告知自己要离开群组 , 接下来就会触发再均衡 , 而不需要等待会话超时。

八、反序列化

不过就是序列化过程的一个反向，原理和实现可以参考生产者端的实现，同样也可以自定义反序列化器。

到此，kafka消费者配置分享完毕。

[d98ebe10c0ce4beaaa84a92a5a3af295.png]: https://img-blog.csdnimg.cn/d98ebe10c0ce4beaaa84a92a5a3af295.png
[watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBAbmFuZGFvMTU4_size_8_color_FFFFFF_t_70_g_se_x_16]: https://img-blog.csdnimg.cn/a773131ee3f94523b297085b1b517768.png?x-oss-process=image/watermark,type_d3F5LXplbmhlaQ,shadow_50,text_Q1NETiBAbmFuZGFvMTU4,size_8,color_FFFFFF,t_70,g_se,x_16
[acc84722e46448fb98b7a654bc42b1fd.png]: https://img-blog.csdnimg.cn/acc84722e46448fb98b7a654bc42b1fd.png
[watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBAbmFuZGFvMTU4_size_8_color_FFFFFF_t_70_g_se_x_16 1]: https://img-blog.csdnimg.cn/ec5c958e7a8446dc9e08af965c384ef6.png?x-oss-process=image/watermark,type_d3F5LXplbmhlaQ,shadow_50,text_Q1NETiBAbmFuZGFvMTU4,size_8,color_FFFFFF,t_70,g_se,x_16
[watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBAbmFuZGFvMTU4_size_20_color_FFFFFF_t_70_g_se_x_16]: https://img-blog.csdnimg.cn/7e273761be65449a9d07903adade048f.png?x-oss-process=image/watermark,type_d3F5LXplbmhlaQ,shadow_50,text_Q1NETiBAbmFuZGFvMTU4,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBAbmFuZGFvMTU4_size_20_color_FFFFFF_t_70_g_se_x_16 1]: https://img-blog.csdnimg.cn/f2d920ec1d3f4c0c9886d229f86d19f7.png?x-oss-process=image/watermark,type_d3F5LXplbmhlaQ,shadow_50,text_Q1NETiBAbmFuZGFvMTU4,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBAbmFuZGFvMTU4_size_16_color_FFFFFF_t_70_g_se_x_16]: https://img-blog.csdnimg.cn/dff0bd892f6b41e081cb4842f3fc52b4.png?x-oss-process=image/watermark,type_d3F5LXplbmhlaQ,shadow_50,text_Q1NETiBAbmFuZGFvMTU4,size_16,color_FFFFFF,t_70,g_se,x_16
[watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBAbmFuZGFvMTU4_size_20_color_FFFFFF_t_70_g_se_x_16 2]: https://img-blog.csdnimg.cn/c54ef9a5d6364b7888472f34e4839d7d.png?x-oss-process=image/watermark,type_d3F5LXplbmhlaQ,shadow_50,text_Q1NETiBAbmFuZGFvMTU4,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBAbmFuZGFvMTU4_size_20_color_FFFFFF_t_70_g_se_x_16 3]: https://img-blog.csdnimg.cn/533dcaf3f3c84abc837b4ca9e50c91a7.png?x-oss-process=image/watermark,type_d3F5LXplbmhlaQ,shadow_50,text_Q1NETiBAbmFuZGFvMTU4,size_20,color_FFFFFF,t_70,g_se,x_16