Kafka（三、Kafka消费者 - 从Kafka读取数据）

刺骨的言语ヽ痛彻心扉 2023-05-30 06:11 90阅读 0赞

# Kakfa 消费者配置 #

同样由于配置太多，在此不一一列举，只显示使用率最高的几个配置，其他配置请参考官网。  
[http://kafka.apache.org/documentation/\#consumerconfigs][http_kafka.apache.org_documentation_consumerconfigs]

<table> 
 <thead> 
  <tr> 
   <th>NAME</th> 
   <th>DESCRIPTION</th> 
   <th>TYPE</th> 
   <th>DEFAULT</th> 
   <th>VALID VALUES</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>key.serializer</td> 
   <td>key序列化，可以使用Kafka默认的序列化和自定义序列化器</td> 
   <td>class</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>value.serializer</td> 
   <td>value序列化，可以使用Kafka默认的序列化和自定义序列化器</td> 
   <td></td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>acks</td> 
   <td>生产者需要leader确认请求完成之前接收的应答数。此配置控制了发送记录的持久性。允许以下设置：<strong>acks=0</strong> 如果设置为0，那么生产者将不等待任何消息确认。消息将立刻添加到socket缓冲区并考虑发送。在这种情况下不能保障消息被服务器接收到。并且重试机制不会生效（因为客户端不知道故障了没有）。每个消息返回的offset始终设置为-1。<strong>acks=1</strong>，这意味着leader写入消息到本地日志就立即响应，而不等待所有follower应答。在这种情况下，如果响应消息之后但follower还未复制之前leader立即故障，那么消息将会丢失。<strong>acks=all</strong> 这意味着leader将等待所有副本同步后应答消息。此配置保障消息不会丢失（只要至少有一个同步的副本）。这是最强壮的可用性保障。等价于acks=-1。</td> 
   <td>string</td> 
   <td>1</td> 
   <td>[all, -1, 0, 1]</td> 
  </tr> 
  <tr> 
   <td>bootstrap.servers</td> 
   <td>host/port列表，用于初始化建立和Kafka集群的连接。列表格式为host1:port1,host2:port2,…，无需添加所有的集群地址，kafka会根据提供的地址发现其他的地址（你可以多提供几个，以防提供的服务器关闭）</td> 
   <td>list</td> 
   <td>“”</td> 
   <td>non-null string</td> 
  </tr> 
  <tr> 
   <td>fetch.min.bytes</td> 
   <td>该属性指定了消费者从服务器获取记录的最小字节数，broker收到消费者的数据请求时，如果可用数据量小于设置的值，那么broker将会等待有足够可用的数据的时候才返回给消费者，这样可以降低消费者和broker的工作负载，因为当主题不是很活跃的情况下，就不需要来来回回的处理消息，如果没有很多可用数据，但消费者的CPU 使用率却很高，那么就需要把该属性的值设得比默认值大。如果消费者的数量比较多，把该属性的值设置得大一点可以降低broker 的工作负载。</td> 
   <td>int</td> 
   <td>1</td> 
   <td>[0,…]</td> 
  </tr> 
  <tr> 
   <td>heartbeat.interval.ms</td> 
   <td>使用Kafka的群组管理工具时，预期的两次心跳之间的时间会协调消费者协调员。心跳用于确保消费者的会话保持活动状态，并在新消费者加入或离开小组时促进重新平衡。该值必须设置为小于<strong>session.timeout.ms</strong>，但通常应设置为不大于该值的1/3。可以将其调整得更低，以控制正常重新平衡的预期时间。</td> 
   <td>int</td> 
   <td>3000</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>session.timeout.ms</td> 
   <td>该属性指定了当消费者被认为已经挂掉之前可以与服务器断开连接的时间。默认是3s，消费者在3s之内没有再次向服务器发送心跳，那么将会被认为已经死亡。此时，协调器将会出发再均衡，把它的分区分配给其他的消费者，该属性与heartbeat.interval.ms紧密相关，该参数定义了消费者发送心跳的时间间隔，也就是心跳频率，一般要同时修改这两个参数，heartbeat.interval.ms参数值必须要小于session.timeout.ms，一般是session.timeout.ms的三分之一，比如，session.timeout.ms设置成3min，那么heartbeat.interval.ms一般设置成1min，这样，可以更快的检测以及恢复崩溃的节点，不过长时间的轮询或垃圾收集可能导致非预期的再均衡（有一种情况就是网络延迟，本身消费者是没有挂掉的，但是网络延迟造成了心跳超时，这样本不该发生再均衡，但是因为网络原因造成了非预期的再均衡），把该属性的值设置得大一些，可以减少意外的再均衡，不过检测节点崩愤－需要更长的时间。</td> 
   <td>int</td> 
   <td>10000</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>max.partition.fetch.bytes</td> 
   <td>该属性指定了服务器从每个分区里返回给消费者的最大字节数。它的默认值是lMB ， 也就是说，kafkaConsumer.poll() 方法从每个分区里返回的记录最多不超 <strong>max.partitions.fetch.bytes</strong> 指定的字节。如果一个主题有20 个分区和5 个消费者，那么每个消费者需要至少4MB 的可用内存来接收记录。在为消费者分配内存时，可以给它们多分配一些，因为如果群组里有消费者发生崩愤，剩下的消费者需要处理更多的分区。<strong>max.partition.fetch.bytes</strong> 的值必须比broker 能够接收的最大消息的字节数（通过<strong>max.message.size</strong> 属性配置）大， 否则消费者可能无法读取这些消息，导致消费者一直挂起重试，例如，<strong>max.message.size</strong>设置为2MB，而该属性设置为1MB，那么当一个生产者可能就会生产一条大小为2MB的消息，那么就会出现问题，消费者能从分区取回的最大消息大小就只有1MB，但是数据量是2MB，所以就会导致消费者一直挂起重试。在设置该属性时，另一个需要考虑的因素是消费者处理数据的时间。消费者需要频繁调用poll（）方法来避免会话过期和发生分区再均衡，如果单次调用poll （） 返回的数据太多，消费者需要更多的时间来处理，可能无怯及时进行下一个轮询来避免会话过期。如果出现这种情况， 可以把max.partitioin.fetch.bytes 值改小，或者延长会话过期时间。</td> 
   <td>int</td> 
   <td>1048576</td> 
   <td>[0,…]</td> 
  </tr> 
  <tr> 
   <td>auto.offset.reset</td> 
   <td>该属性指定了消费者在读取一个没有偏移量后者偏移量无效（消费者长时间失效当前的偏移量已经过时并且被删除了）的分区的情况下，应该作何处理？共有四种解决办法：<strong>earliest</strong>:将偏移量自动重置为最早的偏移量；<strong>latest</strong>：自动将偏移量重置为最新偏移量；<strong>none</strong>：如果未找到消费者组的先前偏移量，则向消费者抛出异常；<strong>anything</strong>：向消费者抛出异常。其<strong>latest</strong>为默认值。</td> 
   <td>String</td> 
   <td>latest</td> 
   <td>[latest, earliest, none]</td> 
  </tr> 
  <tr> 
   <td>enable.auto.commit</td> 
   <td>指定了消费者是否自动提交偏移量，默认值是true，为了尽量避免重复数据和数据丢失，可以把它设置为false，有自己控制合适提交偏移量，如果设置为true， 可以通过设置 <strong>auto.commit.interval.ms</strong>属性来控制提交的频率</td> 
   <td>Boolean</td> 
   <td>true</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>ssl.key.password</td> 
   <td>密钥存储文件中私钥的密码。这对于客户端是可选的。</td> 
   <td>password</td> 
   <td>null</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>ssl.keystore.location</td> 
   <td>密钥存储文件的位置这对于客户端是可选的，可以用于客户端的双向身份验证。</td> 
   <td>string</td> 
   <td>null</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>ssl.keystore.password</td> 
   <td>密钥存储文件的存储密码。这对于客户端是可选的，并且仅在配置了ssl.keystore.location时才需要。</td> 
   <td>password</td> 
   <td>null</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>ssl.truststore.location</td> 
   <td>信任存储和文件的位置。</td> 
   <td>string</td> 
   <td>null</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>ssl.truststore.password</td> 
   <td>信任存储文件的密码。如果未设置密码，则仍然可以访问信任库，但是将禁用完整性检查。</td> 
   <td>password</td> 
   <td>null</td> 
   <td></td> 
  </tr> 
 </tbody> 
</table>

# 示例代码 #

import java.time.Duration;
    import java.util.*;
    
    import org.apache.kafka.clients.consumer.*;
    import org.apache.kafka.common.TopicPartition;
    import org.apache.kafka.common.errors.WakeupException;
    
    public class ConsumerDemo extends Thread { 
        private static Properties properties = new Properties();
    
        private static KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);
    
        static { 
            properties.put("bootstrap.servers", "127.0.0.1:9092");
            properties.put("group.id", "jd-group10");
            properties.put("enable.auto.commit", "true");
            properties.put("auto.commit.interval.ms", "1000");
            properties.put("auto.offset.reset", "earliest");
            properties.put("session.timeout.ms", "30000");
            properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        }
    
    
    
        public static void main(String[] args) { 
    
           while (true) { 
                ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(100));
    
                for (ConsumerRecord<String, String> record : records) { 
                    System.out.println("-----------------");
                    System.out.printf("offset = %s, key =%s, value = %s, partition = %s",
                            record.offset(), record.key(), record.value(), record.partition());
                    System.out.println();
                }
            }
        }
    }

# 提交和偏移量 #

每次调用poll()方法，他总是返回由生产者写入Kafka但是还没有被消费者读取过的记录，因此可以追踪哪些记录是被群组里的那个消费者读取的。Kafka不需要等到消费者的确认，相反的，消费者可以使用Kafka来追踪消息在分区里的位置(偏移量)。

消费者通过 `_consumer_offset` 的特殊租户提发送消息。消息里包含每个分区的偏移量。如果消费者一直处于运行状态。那么偏移量就没有什么用处。不过如果消费者发生崩溃或者有新的消费者加入群组，就会**触发再均衡**，完成再均衡之后，每个消费者可能分配到新的分区，而不是之前处理的那个，为了能够继续之前的工作，消费者需要读取每个分区最后一次提交的偏移量，然后从偏移量指定的地方继续处理。此时：

> 如果提交的偏移量小于客户端处理的最后一个消息的偏移量，那么处于两个偏移量之间的消息就会被重复处理。  
> 如果提交的偏移量大于客户端处理的最后一个消息的偏移量，那么处于两个偏移量之间的消息就会丢失。

所以，提交的便宜来那个的方式对客户端会有很大的影响。  
而KakfaConsumerAPI提供了很多种方式来提交偏移量。

## 自动提交 ##

最简单的提交方式是让悄费者自动提交偏移量。如果**enable.auto.commit**被设为 true，那么每过5s，消费者会自动把从 **poll()** 方法接收到的最大偏移量提交上去。提交时间间隔由 **auto.commit.interval.ms** 控制，默认值是 **5s**。与消费者里的其他东西 一样，自动提交也是在轮询(poll() )里进行的。消费者每次在进行轮询时会检查是否该提交偏移量了，如果是，那 么就会提交从上一次轮询返回的偏移量。

不过，在使用这种简便的方式之前，需要知道它将会带来怎样的结果。  
假设我们仍然使用默认的 5s提交时间间隔，在最近一次提交之后的 3s发生了再均衡，再 均衡之后，消费者从最后一次提交的偏移量位置开始读取消息。这个时候偏移量已经落后 了 3s，所以在这 3s 内到达的消息会被重复处理。可以通过修改提交时间间隔来更频繁地提交偏移量，减小可能出现重复消息的时间窗，不过这种情况是无也完全避免的 。

在使用自动提交时 ，每次调用轮询方怯都会把上一次调用返 回的偏移量提交上去，它并不 知道具体哪些消息已经被处理了，所以在再次调用之前最好确保所有当前调用返回 的消息 都已经处理完毕(在调用 close() 方法之前也会进行自动提交)。 一般情况下不会有什么问 题，不过在处理异常或提前退出轮询时要格外小心 。

自动提交虽然方便 ， 不过并没有为开发者留有余地来避免重复处理消息。

## 提交当前偏移量 ##

大部分开发者通过控制偏移量提交时间来消除丢失消息的可能性，井在发生再均衡时减少 重复消息的数量。消费者 API提供了另一种提交偏移量的方式 ， 开发者可以在必要的时候 提交当前偏移盘，而不是基于时间间隔。

取消自动提交，把 auto.commit.offset 设为 false，让应用程序决定何时提交 偏 移量。使用 commitSync() 提交偏移量最简单也最可靠。这个 API会提交由 poll() 方法返回 的最新偏移量，提交成 功后马上返回，如果提交失败就抛出异常。

要记住， commitSync() 将会提交由 poll() 返回的最新偏移量 ， 所以在处理完所有记录后要 确保调用了 commitSync()，否则还是会有丢失消息的风险。如果发生了再均衡，从最近一 批消息到发生再均衡之间的所有消息都将被重复处理。

/** * 提交当前偏移量 * properties.put("enable.auto.commit", "false"); */
        public void commitNowOffset() { 
            while (true) { 
                ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records) { 
                    System.out.printf("topic = %s , partition = %s , offset = %d, customer = %s ,country = %s\n"
                            , record.topic(), record.partition(), record.offset(), record.key(), record.value());
                }
                try { 
                    // 处理完当前批次的信息，在轮训更多的消息之前，调用commitSync()方法提交当前批次最新的偏移量。失败会一直重试直至成功，只记录日志，
                    kafkaConsumer.commitSync();
                } catch (CommitFailedException e) { 
                    e.printStackTrace();
                }
            }
        }

## 异步提交 ##

同步提交有一个不足之处，在 broker对提交请求作出回应之前，应用程序会一直阻塞，这样会限制应用程序的吞吐量。我们可以通过降低提交频率来提升吞吐量，但如果发生了再均衡， 会增加重复消息的数量。

这个时候可以使用异步提交 API。我们只管发送提交请求，无需等待 broker的响应。

/** * 异步提交 - no callback * properties.put("enable.auto.commit", "false"); */
        public void asynCommit() { 
            while (true) { 
                ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records) { 
                    System.out.printf("topic = %s , partition = %s , offset = %d, customer = %s ,country = %s\n"
                            , record.topic(), record.partition(), record.offset(), record.key(), record.value());
                }
                // 失败不会重试。
                kafkaConsumer.commitAsync();
            }
        }

在成功提交或碰到无怯恢复的错误之前， commitSync() 会一直重试(应用程序也一直阻塞)，但是 commitAsync() 不会，这也是 commitAsync() 不好的 一个地方。它之所以不进行重试，是因为在它收到 服务器响应的时候，可能有一个更大的偏移量已经提交成功。假设我们发出一个请求用于提交偏移量 2000，这个时候发生了短暂的通信问题 ，服务器收不到请求，自然也不会 作出任何响应。与此同时，我们处理了另外一批消息，并成功提交了偏移量 3000。如果 commitAsync() 重新尝试提交偏移量 2000，它有可能在偏移量 3000之后提交成功。这个时 候如果发生再均衡，就会出现重复消息。

我们之所以提到这个问题的复杂性和提交顺序的重要性，是因为 commitAsync()也支持回 调，在 broker 作出响应时会执行回调。回调经常被用于记录提交错误或生成度量指标， 不 过如果你要用它来进行重试， 一定要注意提交的顺序。

/** * 异步提交 - callback * properties.put("enable.auto.commit", "false"); */
        public void asynCommitCallBack() { 
            while (true) { 
                ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records) { 
                    System.out.printf("topic = %s , partition = %s , offset = %d, customer = %s ,country = %s\n"
                            , record.topic(), record.partition(), record.offset(), record.key(), record.value());
                }
                kafkaConsumer.commitAsync((offsets, exception) -> { 
                    if (exception != null) { 
                        log.error("Commit failed for offsets {}", offsets, exception);
                    }
                });
            }
        }

> **重试异步提交**  
> 我们可以使用一个单调递增的序列号来维护异步提交的顺序。在每次提交偏 移量之后或在回调里提交偏移量时递增序列号。在进行重试前，先检查回调 的序列号和即将提交的偏移量是否相等，如果相等，说明没有新的提交，那么可以安全地进行重试。如果序列号比较大，说明有一个新的提交已经发送出去了，应该停止重试。

## 同步和异步组合提交 ##

一般情况下，针对偶尔出现的提交失败，不进行重试不会有太大问题，因为如果提交失败 是 因为临时问题导致的，那么后续的提交总会有成功的。但如果这是发生在关闭消费者或 再均衡前的最后一次提交，就要确保能够提交成功。

因此，在消费者关闭前一般会组合使用 commitAsync()和 commitSync()。它们的工作原理如下(后面讲到再均衡监听器时，我们会讨论如何在发生再均衡前提交偏移量):

/** * 同步和异步提交组合 * properties.put("enable.auto.commit", "false"); */
        public void syncAndAsyncCommit() { 
            try { 
                while (true) { 
                    ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(100));
                    for (ConsumerRecord<String, String> record : records) { 
                        System.out.printf("topic = %s , partition = %s , offset = %d\n", record.topic(), record.partition(), record.offset());
                    }
                    // 如果一切正常，先用commitAsync()进行提交。这样速度更快，而且即使这次提交失败，下一次提交很可能会成功。
                    kafkaConsumer.commitAsync();
                }
            } catch (Exception e) { 
                log.error("Unexpected error", e);
            } finally { 
                try { 
                    // 如果直接关闭消费者，就没有锁为的"下一次提交"了，使用commitSync()方法会一直重试，直到提交成功或发生无法恢复的错误。
                    kafkaConsumer.commitSync();
                } finally { 
                    kafkaConsumer.close();
                }
            }
    
        }

## 提交待定的偏移量 ##

提交偏移量的频率与处理消息批次的频率是一样的。但如果想要更频繁地提交出怎么办？如果 poll() 方法返回一大批数据，为了避免因再均衡引起的重复处理整批消息，想要在批次中间提交偏移量该怎么办？这种情况无法通过调用 commitSync()或 commitAsync() 来实现，因为它们只会提交最后一个偏移量，而此时该批次里的消息还没有处理完。

幸运的是，消费者 API 允许在调用 commitSync()和 commitAsync()方法时传进去希望提交 的分区和偏移量的 map。假设你处理了半个批次的消息， 最后一个来自主题“customers” 分区 3 的消息的偏移量是 5000， 你可以调用 commitSync() 方法来提交它。不过，因为消费者可能不只读取一个分区， 你需要跟踪所有分区的偏移量，所以在这个层面上控制偏移 量 的提交会让代码变复杂。

/** * 提交特定偏移量 * properties.put("enable.auto.commit", "false"); */
        public void commitAppointOffset() { 
    
            Map<TopicPartition, OffsetAndMetadata> currentOffsets = new HashMap<>();
    
            int count = 0;
            while (true) { 
                ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records) { 
                    System.out.printf("topic = %s , partition = %s , offset = %d\n", record.topic(), record.partition(), record.offset());
    
                    currentOffsets.put(new TopicPartition(record.topic(), record.partition()), new OffsetAndMetadata(record.offset() + 1, "no metadata"));
                    // 每次处理1000条提交一次偏移量
                    if (count % 1000 == 0) { 
                        kafkaConsumer.commitAsync(currentOffsets, null);
                    }
                    count++;
                }
            }
        }

# 再均衡监听器 #

在提交偏移量一节中提到过，消费者在退出和进行分区再均衡之前，会做一些清理工作。

你会在消费者失去对一个分区的所有权之前提交最后一个已处理记录的偏移量。如果消费 者准备了 一 个缓冲区用于处理偶发的事件，那么在失去分区所有权之前， 需要处理在缓冲 区累积下来的记录。你可能还需要关闭文件句柄、数据库连接等。

在为消费者分配新分区或移除旧分区时，可以通过消费者 API执行 一 些应用程序代码，在调用 subscribe()方法时传进去一个ConsumerRebalancelistener实例就可以了。 ConsumerRebalancelistener有两个需要实现的方法。

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

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

下面的例子将演示如何在失去分区所有权之前通过 onPartitionsRevoked()方法来提交偏移量。在下一节，我们会演示另一个同时使用了 onPartitionsAssigned()方法的例子。

private static Map<TopicPartition, OffsetAndMetadata> currentOffsets = new HashMap<>();
    
        private static class HandlerRebalance implements ConsumerRebalanceListener { 
    
            @Override
            // 会在再均衡开始之前和消费者停止读取消息之后被调用。如果在这里提交偏移量，下一个接管分区的消费之就知道该成哪里开始读取了。
            public void onPartitionsRevoked(Collection<TopicPartition> partitions) { 
                System.out.println("Lost partitions in rebalance ,Committing current " +
                        "offset: " + currentOffsets);
    
                commitDbTransaction();
            }
    
            @Override
            // 方法会在重新分配分区之后和消费者开始读取消息之前被调用
            public void onPartitionsAssigned(Collection<TopicPartition> partitions) { 
    
                for (TopicPartition partition : partitions) { 
                    kafkaConsumer.seek(partition, getOffsetFromDb(partition));
                }
            }
        }
    
        public void rebalanceListener(){ 
            try { 
    
                // 把ConsumerRebalanceListener对象传给subscribe()方法，这是最重要的一步
                kafkaConsumer.subscribe(Collections.singleton("test5"), new HandlerRebalance());
    
                while (true) { 
    
                    ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(100));
    
                    for (ConsumerRecord<String, String> record : records) { 
                        System.out.println("-----------------");
                        System.out.printf("offset = %s, key =%s, value = %s, partition = %s",
                                record.offset(), record.key(), record.value(), record.partition());
                        System.out.println();
    
                        currentOffsets.put(new TopicPartition(record.topic(), record.partition()),
                                new OffsetAndMetadata(record.offset() + 1, " no metadata"));
    
                    }
    
                    // 在再均衡之前提交偏移量。
                    // 提交的是最近处理过的偏移量，而不是批次中还在处理的最后一个偏移量。
                    // 提交的是所分区的偏移量，而不是那些即将逝去所有权的分区的偏移量。以为提交的偏移量是之前处理过的，所以不会有问题。
                    kafkaConsumer.commitAsync(currentOffsets, null);
                }
    
            } catch (WakeupException e) { 
                // 忽略异常，正在关闭消费者。
            } catch (Exception e) { 
                e.printStackTrace();
            } finally { 
                try { 
                    kafkaConsumer.commitSync(currentOffsets);
                } finally { 
                    kafkaConsumer.close();
                    System.out.println("Closed consumer and we are done");
                }
            }
        }

# 从特定偏移量处开始处理记录 #

到目前为止，我们知道了如何使用 `poll()` 方法从各个分区的最新偏移量处开始处理消息。 不过，有时候我们也需要从特定的偏移量处开始读取消息。

如果你想从分区的起始位置开始读取消息，或者直接跳到分区的末尾开始读取消息， 可以使 用 `seekToBeginning(Collection<TopicPartition> tp)` 和 `seekToEnd(Collection<TopicPartition> tp)` 这两个方法。

不过， Kafka也为我们提供了用 于查找特定偏移量的 API。 它有很多用途，比如向 后回退 几个消息或者向前跳过几个消息(对时间比较敏感的应用程序在处理滞后的情况下希望能 够向前跳过若干个消息)。在使用 Kafka 以外的系统来存储偏移量时，它将给我们 带来更 大的惊喜。

试想一下这样的场景:应用程序从 Kafka读取事件(可能是网站的用户点击事件流 )，对 它们进行处理(可能是使用自动程序清理点击操作井添加会话信息)，然后把结果保 存到 数据库、 NoSQL 存储引擎或 Hadoop。假设我们真的不想丢失任何数据，也不想在数据库 里多次保存相同的结果。

这种情况下，消费者的代码可能是这样的 :

while (true) { 
                ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(100));
    
                for (ConsumerRecord<String, String> record : records) { 
                     currentOffsets.put(new TopicPartition(record.topic(), record.partition()), new OffsetAndMetadata(record.offset() + 1);
                     processRecord(record);
                     storeRecordInDB(record);
                     kafkaConsumer.commitAsync();
                }
            }

在这个例子里，每处理一条记录就提交一次偏移量。尽管如此， 在记录被保存到数据库之后以及偏移量被提交之前 ，应用程序仍然有可能发生崩溃，导致重复处理数据，数据库里就会出现重复记录。

如果保存记录和偏移量可以在一个原子操作里完成，就可以避免出现上述情况。记录和偏 移量要么 都被成功提交，要么都不提交。如果记录是保存在数据库里而偏移量是提交到 Kafka 上，那么就无法实现原子操作。

不过 ，如果在同一个事务里把记录和偏移量都写到数据库里会怎样呢？那么我们就会知道 记录和偏移量要么都成功提交，要么都没有，然后重新处理记录。

现在的问题是：如果偏移量是保存在数据库里而不是 Kafka里，那么消费者在得到新分区 时怎么知道该从哪里开始读取?这个时候可以使用 seek() 方法。在消费者启动或分配到新 分区时 ，可以使用 seek()方法查找保存在数据库里的偏移量。

下面的例子大致说明了如何使用这个 API。 使用 ConsumerRebalancelistener和 seek() 方 战确保我们是从数据库里保存的偏移量所指定的位置开始处理消息的。

/** * 将当前偏移量以及记录写进数据库，然后再即将失去分区所有权之前提交事务，确保成功保存了这些信息。 */
        private static void commitDbTransaction() { 
        }
    
        /** * 从数据库中获取偏移量 * 将保存的记录和偏移量使用原子操作，通过seek方法，让kafka获取 */
        private static long getOffsetFromDb(TopicPartition partition) { 
            return partition.partition();
        }
    
        public class SaveOffsetOnRebalance implements ConsumerRebalanceListener { 
    
            @Override
            public void onPartitionsRevoked(Collection<TopicPartition> partitions) { 
                commitDbTransaction();
            }
    
    
            @Override
            public void onPartitionsAssigned(Collection<TopicPartition> partitions) { 
                for(TopicPartition partition : partitions){ 
                    kafkaConsumer.seek(partition, getOffsetFromDb(partition));
                }
            }
        }
    
        public void startFromOffset(){ 
            // 把ConsumerRebalanceListener对象传给subscribe()方法，这是最重要的一步
            kafkaConsumer.subscribe(Collections.singleton("test5"), new HandlerRebalance());
            kafkaConsumer.poll(Duration.ofMillis(0));
    
            for (TopicPartition partition : kafkaConsumer.assignment()) { 
                // 启动消费者，调用一次poll()方法，让消费者加入到消费者群组里，并获取分配到的分区，然后马上调用seek()定位分区的偏移量。
                // 要记住，seek()方法只更新我们正在使用的位置，在下一次调用poll()时就可以获得正确的消息。如果seek()发生错误(比如偏移量不在)，poll()就会抛出异常
                kafkaConsumer.seek(partition, getOffsetFromDb(partition));
            }
    
            while (true) { 
    
                ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(100));
    
                for (ConsumerRecord<String, String> record : records) { 
                    // 以下均为虚构方法，更新数据库用于保存偏移量的表。
                    processRecord(record);
                    storeRecordInDB(record);
                    storeOffsetInDB(record.topic(), record.partition(), record.offset());
                }
    
                commitDbTransaction();
            }
    
        }

# 如何退出 #

在之前讨论轮询时就说过，不需要担心消费者会在一个无限循环里轮询消息，我们会告诉消费者如何优雅地退出循环。

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

public void quit() { 
            Runtime.getRuntime().addShutdownHook(new Thread() { 
                @Override
                public void run() { 
                    System.out.println("Starting exit....");
    
                    // ShutdownHook运行在单独的县城里，所以退出循环最安全的方式只能是调用wakeUp()方法
                    kafkaConsumer.wakeup();
                    try { 
                        this.join();
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }
            });
    
            try { 
                // 循环，直到按下Ctrl+C键，关闭的狗子会在退出时进行清理
                while (true) { 
                    ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(1000));
                    System.out.println(System.currentTimeMillis() + " - - waiting for data...");
    
                    for (ConsumerRecord<String, String> record : records) { 
                        System.out.printf("topic = %s , partition = %s , offset = %d\n", record.topic(), record.partition(), record.offset());
                    }
    
                    for (TopicPartition partition : kafkaConsumer.assignment()) { 
                        System.out.println("Commiting offset at position : " + kafkaConsumer.position(partition));
                    }
    
                    kafkaConsumer.commitSync();
    
                }
            } catch (WakeupException e) { 
                // 在拎一个二线城里调用wakeup(),导致poll()抛出WakeupException。没什么必要捕捉异常确保应用不会意外终止，可以忽略。
                // e.printStackTrace();
            } finally { 
                // 在退出之前，确保彻底关闭了消费者
                kafkaConsumer.close();
                System.out.println("Closed consumer and we are done");
            }
        }

在提交偏移量的时候，不知道你们有没有想过一个问题，我们都知道每个分区的offset是递增的，消费者读取的时候顺序不会乱。但是当触发再均衡的时候，如何保证整个topic里面的消息顺序性呢，或者理解为全局分区的顺序性？  
解决：我认为将一个表的消息存放在一个分区里面。或者根据业务，具体情况，将具有相同特性的数据放进一个分区。

[http_kafka.apache.org_documentation_consumerconfigs]: http://kafka.apache.org/documentation/#consumerconfigs