kafka 消费者 API 使用总结_kafka消费的接口怎么接收

前言

应用程序使用KafkaConsumer向Kafka订阅主题，并从订阅的主题中接收消息。不同于从其他消息系统读取数据，从Kafka读取数据涉及一些独特的概念和想法。如果不先理解这些概念，则难以理解如何使用消费者API。本文将先解释这些重要的概念，然后再举几个例子，演示如何使用消费者API实现不同的应用程序。

Kafka消费者相关概念

消费者和消费者群组

消费者从属于消费者群组，一个群组里的消费者订阅的是同一个主题，每个消费者负责读取这个主题的部分消息。
假设主题T1有4个分区，我们创建了消费者C1，它是群组G1中唯一的消费者，用于订阅主题T1。消费者C1将收到主题T1全部4个分区的消息，如下图所示：

如果在群组G1里新增一个消费者C2，那么每个消费者将接收到两个分区的消息。假设消费者C1接收分区0和分区2的消息，消费者C2接收分区1和分区3的消息，如下图所示：

如果群组G1有4个消费者，那么每个消费者将可以分配到一个分区：

如果向群组里添加更多的消费者，以致超过了主题的分区数量，那么就会有一部分消费者处于空闲状态，不会接收到任何消：

• 我们可以为主题创建大量的分区，当负载急剧增长时，可以加入更多的消费者。不过需要注意的是，不要让消费者的数量超过主题分区的数量，因为多余的消费者只会被闲置

kafka 的 Topic 是可以同时被多个消费者组消费的，比如说有两个程序想要同时读取数据，就可以将这两个程序放到两个不同的消费者组中：

消费者群组和分区再均衡

如上文所述，消费者群组里的消费者共享主题分区的所有权。当一个新消费者加入群组时，它将开始读取一部分原本由其他消费者读取的消息。当一个消费者被关闭或发生崩溃时，它将离开群组，原本由它读取的分区将由群组里的其他消费者读取。主题发生变化（比如管理员添加了新分区）会导致分区重分配。

分区的所有权从一个消费者转移到另一个消费者的行为称为再均衡。再均衡非常重要，它为消费者群组带来了高可用性和伸缩性（你可以放心地添加或移除消费者）。不过，在正常情况下，我们并不希望发生再均衡。

根据消费者群组所使用的分区分配策略的不同，再均衡可以分为两种类型。
主动再均衡

• 在进行主动再均衡期间，所有消费者都会停止读取消息，放弃分区所有权，重新加入消费者群组，并获得重新分配到的分区。

• 这样会导致整个消费者群组在一个很短的时间窗口内不可用。这个时间窗口的长短取决于消费者群组的大小和几个配置参数。

• 主动再均衡包含两个不同的阶段：第一个阶段，所有消费者都放弃分区所有权；第二个阶段，消费者重新加入群组，获得重新分配到的分区，并继续读取消息：

  

协作再均衡

• 协作再均衡（也称为增量再均衡）通常是指将一个消费者的部分分区重新分配给另一个消费者，其他消费者则继续读取没有被重新分配的分区。这种再均衡包含两个或多个阶段。

• 在第一个阶段，消费者群组首领会通知所有消费者，它们将失去部分分区的所有权，然后消费者会停止读取这些分区，并放弃对它们的所有权。

• 在第二个阶段，消费者群组首领会将这些没有所有权的分区分配给其他消费者。

• 虽然这种增量再均衡可能需要进行几次迭代，直到达到稳定状态，但它避免了主动再均衡中出现的“停止世界”停顿。

  
  消费者会向被指定为群组协调器的broker（不同消费者群组的协调器可能不同）发送心跳，以此来保持群组成员关系和对分区的所有权关系。心跳是由消费者的一个后台线程发送的，只要消费者能够以正常的时间间隔发送心跳，它就会被认为还“活着”。
  
  如果消费者在足够长的一段时间内没有发送心跳，那么它的会话就将超时，群组协调器会认为它已经“死亡”，进而触发再均衡。如果一个消费者发生崩溃并停止读取消息，那么群组协调器就会在几秒内收不到心跳，它会认为消费者已经“死亡”，进而触发再均衡。在这几秒时间里，“死掉”的消费者不会读取分区里的消息。在关闭消费者后，协调器会立即触发一次再均衡，尽量降低处理延迟。
  

群组固定成员

在默认情况下，消费者的群组成员身份标识是临时的。当一个消费者离开群组时，分配给它的分区所有权将被撤销；当该消费者重新加入时，将通过再均衡协议为其分配一个新的成员ID和新分区。

如果想要当消费者重启后，消费的分区不变，继续消费之前消费的分区，就可以给消费者分配一个唯一的group.instance.id，让它成为群组的固定成员。

如果两个消费者使用相同的group.instance.id加入同一个群组，则第二个消费者会收到错误，告诉它具有相同ID的消费者已存在。

固定成员不能离开群组太久，在 session.timeout.ms时间内重新加入会继续使用之前的 ID 以及消费之前的分区，当超过该时间后，该成员被认定为"死亡"， 需要以新的身份加入群组，消费新的分区。

session.timeout.ms 参数不能设置的太大，以便在出现严重停机时自动重新分配分区，避免这些分区的读取进度出现较大的滞后。又不能设置的太小，避免在进行简单的应用程序重启时触发再均衡。

创建消费者

三个必填参数：bootstrap.servers、key.deserializer和value.deserializer。

Properties props = new Properties();
props.put("bootstrap.servers", "broker1:9092,broker2:9092");
props.put("group.id", "CountryCounter");
props.put("key.deserializer",
    "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer",
    "org.apache.kafka.common.serialization.StringDeserializer");

KafkaConsumer<String, String> consumer =
    new KafkaConsumer<String,String>(props);
1
2
3
4
5
6
7
8
9
10

订阅 Topic
可以通过主题列表订阅一个或多个主题：

consumer.subscribe(Collections.singletonList("customerCountries")); ➊
1

也可以通过正则表达式匹配订阅多个主题：

consumer.subscribe(Pattern.compile("test.*"));
1

使用正则需要注意的是，消费者客户端会拉取集群的所有主题和分区元信息，可能会给网络带来很大的开销。而且为了能够使用正则表达式订阅主题，需要授予客户端获取集群全部主题元数据的权限，即全面描述整个集群的权限。

轮询消费

Duration timeout = Duration.ofMillis(100);

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(timeout);

    for (ConsumerRecord<String, String> record : records) {
        System.out.printf("topic = %s, partition = %d, offset = %d, " +
                        "customer = %s, country = %s\n",
        record.topic(), record.partition(), record.offset(),
                record.key(), record.value());
        int updatedCount = 1;
        if (custCountryMap.containsKey(record.value())) {
            updatedCount = custCountryMap.get(record.value()) + 1;
        }
        custCountryMap.put(record.value(), updatedCount);

        JSONObject json = new JSONObject(custCountryMap);
        System.out.println(json.toString());
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

1. 传给poll()的参数是一个超时时间间隔，用于控制poll()的阻塞时间（当消费者缓冲区里没有可用数据时会发生阻塞）。如果这个参数被设置为0或者有可用的数据，那么poll()就会立即返回，否则它会等待指定的毫秒数。
2. 轮询不只是获取数据那么简单。在第一次调用消费者的poll()方法时，它需要找到GroupCoordinator，加入群组，并接收分配给它的分区。如果触发了再均衡，则整个再均衡过程也会在轮询里进行，包括执行相关的回调。所以，消费者或回调里可能出现的错误最后都会转化成poll()方法抛出的异常。
3. 需要注意的是，如果超过max.poll.interval.ms没有调用poll()，则消费者将被认为已经“死亡”，并被逐出消费者群组。因此，要避免在轮询循环中做任何可能导致不可预知的阻塞的操作。
   

消费者参数配置

fetch.min.bytes

• 这个属性指定了消费者从服务器获取记录的最小字节数，默认是1字节。
• broker在收到消费者的获取数据请求时，如果可用数据量小于fetch.min.bytes指定的大小，那么它就会等到有足够可用数据时才将数据返回。
• 这样可以降低消费者和broker的负载，因为它们在主题流量不是很大的时候（或者一天里的低流量时段）不需要来来回回地传输消息。如果消费者在没有太多可用数据时CPU使用率很高，或者在有很多消费者时为了降低broker的负载，那么可以把这个属性的值设置得比默认值大。
• 但需要注意的是，在低吞吐量的情况下，加大这个值会增加延迟。
  

fetch.max.wait.ms

• 让Kafka等到有足够多的数据时才将它们返回给消费者，feth.max.wait.ms则用于指定broker等待的时间，默认是500毫秒。
• 如果要降低潜在的延迟（为了满足SLA），那么可以把这个属性的值设置得小一些。如果fetch.max.wait.ms被设置为100毫秒，fetch.min.bytes被设置为1 MB，那么Kafka在收到消费者的请求后，如果有1 MB数据，就将其返回，如果没有，就在100毫秒后返回，就看哪个条件先得到满足。
  

fetch.max.bytes

• 指定了Kafka返回的数据的最大字节数（默认为50 MB）。消费者会将服务器返回的数据放在内存中，所以这个属性被用于限制消费者用来存放数据的内存大小。
• broker端也有一个与之对应的配置属性，Kafka管理员可以用它来限制最大获取数量。broker端的这个配置属性可能很有用，因为请求的数据量越大，需要从磁盘读取的数据量就越大，通过网络发送数据的时间就越长，这可能会导致资源争用并增加broker的负载。
  

max.poll.records

• 这个属性用于控制单次调用poll()方法返回的记录条数。可以用它来控制应用程序在进行每一次轮询循环时需要处理的记录条数（不是记录的大小）。
  

max.partition.fetch.bytes

• 服务器从每个分区里返回给消费者的最大字节数（默认值是1 MB）。
• 需要注意的是，使用这个属性来控制消费者的内存使用量会让事情变得复杂，因为你无法控制broker返回的响应里包含多少个分区的数据。因此，对于这种情况，建议用fetch.max.bytes替代，除非有特殊的需求，比如要求从每个分区读取差不多的数据量。
  

session.timeout.ms和heartbeat.interval.ms

• session.timeout.ms指定了消费者可以在多长时间内不与服务器发生交互而仍然被认为还“活着”，默认是10秒。
• 如果消费者没有在session.timeout.ms指定的时间内发送心跳给群组协调器，则会被认为已“死亡”，协调器就会触发再均衡，把分区分配给群组里的其他消费者。
• session.timeout.ms与heartbeat.interval.ms紧密相关。heartbeat.interval.ms指定了消费者向协调器发送心跳的频率，session.timeout.ms指定了消费者可以多久不发送心跳。
• 把session.timeout.ms设置得比默认值小，可以更快地检测到崩溃，并从崩溃中恢复，但也会导致不必要的再均衡。
• 把session.timeout.ms设置得比默认值大，可以减少意外的再均衡，但需要更长的时间才能检测到崩溃。
  

max.poll.interval.ms

• 指定了消费者在被认为已经“死亡”之前可以在多长时间内不发起轮询。
• 这个属性的默认值为5分钟。当这个阈值被触及时，后台线程将向broker发送一个“离开群组”的请求，让broker知道这个消费者已经“死亡”，必须进行群组再均衡，然后停止发送心跳。
  

default.api.timeout.ms

• 如果在调用消费者API时没有显式地指定超时时间，那么消费者就会在调用其他API时使用这个属性指定的值。
  

request.timeout.ms

• 消费者在收到broker响应之前可以等待的最长时间。如果broker在指定时间内没有做出响应，那么客户端就会关闭连接并尝试重连。它的默认值是30秒。
  

auto.offset.reset

• 指定了消费者在读取一个没有偏移量或偏移量无效（因消费者长时间不在线，偏移量对应的记录已经过期并被删除）的分区时该做何处理。
• 它的默认值是latest，意思是说，如果没有有效的偏移量，那么消费者将从最新的记录（在消费者启动之后写入Kafka的记录）开始读取。
• 另一个值是earliest，意思是说，如果没有有效的偏移量，那么消费者将从起始位置开始读取记录。
• 如果将auto.offset.reset设置为none，并试图用一个无效的偏移量来读取记录，则消费者将抛出异常。
  

enable.auto.commit

• 指定了消费者是否自动提交偏移量，默认值是true。可以把它设置为false，选择自己控制何时提交偏移量，以尽量避免出现数据重复和丢失。
• 如果它被设置为true，那么还有另外一个属性auto.commit.interval.ms可以用来控制偏移量的提交频率。
  

partition.assignment.strategy

PartitionAssignor根据给定的消费者和它们订阅的主题来决定哪些分区应该被分配给哪个消费者。Kafka提供了几种默认的分配策略。
区间(range)

• 这个策略会把每一个主题的若干个连续分区分配给消费者。
• 假设消费者C1和消费者C2同时订阅了主题T1和主题T2，并且每个主题有3个分区。那么消费者C1有可能会被分配到这两个主题的分区0和分区1，消费者C2则会被分配到这两个主题的分区2。

轮询(roundRobin)

• 这个策略会把所有被订阅的主题的所有分区按顺序逐个分配给消费者。

黏性(sticky)

• 设计黏性分区分配器的目的有两个：一是尽可能均衡地分配分区，二是在进行再均衡时尽可能多地保留原先的分区所有权关系，减少将分区从一个消费者转移给另一个消费者所带来的开销。
• 如果同一个群组里的消费者订阅了不同的主题，那么黏性分配器的分配比例将比轮询分配器更加均衡。

协作黏性(cooperative sticky)

• 这个分配策略与黏性分配器一样，只是它支持协作（增量式）再均衡，在进行再均衡时消费者可以继续从没有被重新分配的分区读取消息。
  

client.id

• 这个属性可以是任意字符串，broker用它来标识从客户端发送过来的请求，比如获取请求。它通常被用在日志、指标和配额中。
  

client.rack

• 在默认情况下，消费者会从每个分区的首领副本那里获取消息。但是，如果集群跨越了多个数据中心或多个云区域，那么让消费者从位于同一区域的副本那里获取消息就会具有性能和成本方面的优势。要从最近的副本获取消息，需要设置client.rack这个参数，用于标识客户端所在的区域。然后，可以将broker的replica.selector.class参数值改为org.apache.kafka.common.replica.RackAwareReplicaSelector。
• 也可以实现自己的replica.selector.class，根据客户端元数据和分区元数据选择想要读取的副本。
  

group.instance.id

• 这个属性可以是任意具有唯一性的字符串，被用于消费者群组的固定名称。
  

receive.buffer.bytes和send.buffer.bytes

• 这两个属性分别指定了socket在读写数据时用到的TCP缓冲区大小。
• 如果它们被设置为–1，就使用操作系统的默认值。
• 如果生产者或消费者与broker位于不同的数据中心，则可以适当加大它们的值，因为跨数据中心网络的延迟一般都比较高，而带宽又比较低。
  

offsets.retention.minutes

• 只要消费者群组里有活跃的成员（也就是说，有成员通过发送心跳来保持其身份），群组提交的每一个分区的最后一个偏移量就会被Kafka保留下来，在进行重分配或重启之后就可以获取到这些偏移量。但是，如果一个消费者群组失去了所有成员，则Kafka只会按照这个属性指定的时间（默认为7天）保留偏移量。一旦偏移量被删除，即使消费者群组又“活”了过来，它也会像一个全新的群组一样，没有了过去的消费记忆。
  

提交和偏移量

我们把更新分区当前读取位置的操作叫作偏移量提交。与传统的消息队列不同，Kafka不会提交每一条记录。相反，消费者会将已成功处理的最后一条消息提交给Kafka，并假定该消息之前的每一条消息都已成功处理。

消费者会向一个叫作 __consumer_offset的主题发送消息，消息里包含每个分区的偏移量。如果消费者一直处于运行状态，那么偏移量就没有什么实际作用。但是，如果消费者发生崩溃或有新的消费者加入群组，则会触发再均衡。

再均衡完成之后，每个消费者可能会被分配新的分区，而不是之前读取的那个。为了能够继续之前的工作，消费者需要读取每个分区最后一次提交的偏移量，然后从偏移量指定的位置继续读取消息。

重复消费消息

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

消息丢失
如果最后一次提交的偏移量大于客户端处理的最后一条消息的偏移量，那么处于两个偏移量之间的消息就会丢失:

自动提交

• enable.auto.commit 被设置为 true，每过5秒，消费者就会自动提交poll()返回的最大偏移量。
• 提交时间间隔通过 auto.commit.interval.ms 来设定，默认是5秒。
• 如果在 5 s 内消费者奔溃，消费者提交的偏移量是上一次提交的，落后于正在处理的事件，接管分区的消费者就会出现上图重复消费的情况。
• 虽然自动提交很省事，但是用户没法对重复消费的情况进行处理，需要容忍出现重复消费的情况。
  

提交当前偏移量

• 把 enable.auto.commit 设置为false，消费者不会自动提交，应用程序可以使用 commitSync()主动提交当前偏移量。
• 需要注意的是，消费者程序在处理完所有记录前就提交了 commitSync()，那么一旦应用程序发生崩溃，就会有丢失消息的风险（消息已被提交但未被处理）。
• 如果应用程序在处理记录时发生崩溃，但commitSync()还没有被调用，那么从最近批次的开始位置到发生再均衡时的所有消息都将被再次处理。
• 主动提交可能会出现消息丢失或者重复消费，但是这是可以在用户的应用程序侧控制的，下边是在处理完最近一批消息后使用commitSync()提交偏移量的例子：

  Duration timeout = Duration.ofMillis(100);
  
  while (true) {
      ConsumerRecords<String, String> records = consumer.poll(timeout);
      for (ConsumerRecord<String, String> record : records) {
          System.out.printf("topic = %s, partition = %d, offset =
              %d, customer = %s, country = %s\n",
              record.topic(), record.partition(),
              record.offset(), record.key(), record.value());
      }
      try {
          consumer.commitSync();
      } catch (CommitFailedException e) {
          log.error("commit failed", e)
      }
  }
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16

异步提交

前边的手动提交，会阻塞等待 broker 响应，限制应用程序的吞吐量，此时可以使用consumer.commitAsync()异步提交，无需等待 broker 响应。

Duration timeout = Duration.ofMillis(100);

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(timeout);
    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());
    }
    consumer.commitAsync();
}
1
2
3
4
5
6
7
8
9
10
11
12

异步提交可能会出现的问题：
在提交成功或碰到无法恢复的错误之前，commitSync()会一直重试，但commitAsync()不会，这是commitAsync()的一个缺点。之所以不进行重试，是因为commitAsync()在收到服务器端的响应时，可能已经有一个更大的偏移量提交成功。

假设我们发出一个提交偏移量2000的请求，这个时候出现了短暂的通信问题，服务器收不到请求，自然也不会做出响应。

与此同时，我们处理了另外一批消息，并成功提交了偏移量3000。如果此时commitAsync()重新尝试提交偏移量2000，则有可能在偏移量3000之后提交成功。这个时候如果发生再均衡，就会导致消息重复。

Duration timeout = Duration.ofMillis(100);

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(timeout);
    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());
    }
    consumer.commitAsync(new OffsetCommitCallback() {
        public void onComplete(Map<TopicPartition,
        OffsetAndMetadata> offsets, Exception e) {
            if (e != null)
                log.error("Commit failed for offsets {}", offsets, e);
        }
    });
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

• commitAsync()也支持回调，回调会在broker返回响应时执行。回调经常被用于记录偏移量提交错误或生成指标，如果要用它来重试提交偏移量，那么一定要注意提交顺序。

异步提交中的重试
可以用一个单调递增的消费者序列号变量来维护异步提交的顺序。每次调用commitAsync()后增加序列号，并在回调中更新序列号变量。在准备好进行重试时，先检查回调的序列号与序列号变量是否相等。如果相等，就说明没有新的提交，可以安全地进行重试。如果序列号变量比较大，则说明已经有新的提交了，此时应该停止重试。

同步和异步组合提交

Duration timeout = Duration.ofMillis(100);

try {
    while (!closing) {
        ConsumerRecords<String, String> records = consumer.poll(timeout);
        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());
        }
        consumer.commitAsync();➊
    }
    consumer.commitSync();➋
} catch (Exception e) {
    log.error("Unexpected error", e);
} finally {
        consumer.close();
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

• ➊ 如果一切正常，就用commitAsync()提交偏移量。这样速度更快，而且即使这次提交失败，下一次提交也会成功。
• ➋ 如果直接关闭消费者，那么就没有所谓的“下一次提交”了。commitSync()会一直重试，直到提交成功或发生无法恢复的错误。
  

提交特定的偏移量

前边的提交方式都是默认提交消息批次里的最后一个偏移量，我们也可以自己记录已消费的偏移量，然后提交特定的偏移量。

private Map<TopicPartition, OffsetAndMetadata> currentOffsets =
    new HashMap<>(); ➊
int count = 0;

....
Duration timeout = Duration.ofMillis(100);

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(timeout);
    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()); ➋
        currentOffsets.put(
            new TopicPartition(record.topic(), record.partition()),
            new OffsetAndMetadata(record.offset()+1, "no metadata")); ➌
        if (count % 1000 == 0) ➍
            consumer.commitAsync(currentOffsets, null); ➎
        count++;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

❶ 用于跟踪偏移量的map。
❷ printf表示处理消息的过程。
❸ 在读取每一条记录之后，将下一条要处理的消息的偏移量更新到map中。提交的偏移量应该是应用程序要处理的下一条消息的偏移量，下一次就从这个位置开始读取。
❹ 我们决定每处理1000条记录就提交一次偏移量。在实际当中，可以基于时间或记录的内容来提交偏移量。
❺ 这里调用的是commitAsync()（没有回调，所以第二个参数是null），不过调用commitSync()也是可以的。当然，在提交特定偏移量时仍然要处理可能出现的错误，就像之前那样。

再均衡监听器

消费者API提供了一些方法，让我们可以在消费者分配到新分区或旧分区被移除时执行一些代码逻辑。我们所要做的就是在调用subscribe()方法时传进去一个ConsumerRebalanceListener对象。ConsumerRebalanceListener有3个需要实现的方法:
public void onPartitionsAssigned(Collection partitions)

• 在每次进行再均衡时都会被调用，以此来告诉消费者发生了再均衡。如果没有新的分区分配给消费者，那么它的参数就是一个空集合。

public void onPartitionsRevoked(Collection partitions)

• 会在进行正常的再均衡并且有消费者放弃分区所有权时被调用。如果它被调用，那么参数就不会是空集合。

public void onPartitionsLost(Collection partitions)

• 会在进行意外的再均衡并且参数集合中的分区已经有新的所有者的情况下被调用。

下面的例子演示了如何在失去分区所有权之前通过onPartitionsRevoked()方法来提交偏移量:

private Map<TopicPartition, OffsetAndMetadata> currentOffsets =
    new HashMap<>();
Duration timeout = Duration.ofMillis(100);

private class HandleRebalance implements ConsumerRebalanceListener { ➊
    public void onPartitionsAssigned(Collection<TopicPartition>
        partitions) { ➋
    }

    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        System.out.println("Lost partitions in rebalance. " +
            "Committing current offsets:" + currentOffsets);
        consumer.commitSync(currentOffsets); ➌
    }
}

try {
    consumer.subscribe(topics, new HandleRebalance()); ➍

    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(timeout);
        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());
            currentOffsets.put(
                new TopicPartition(record.topic(), record.partition()),
                new OffsetAndMetadata(record.offset()+1, null));
        }
        consumer.commitAsync(currentOffsets, null);
    }
} catch (WakeupException e) {
    // 忽略异常
} catch (Exception e) {
    log.error("Unexpected error", e);
} finally {
    try {
        consumer.commitSync(currentOffsets);
    } finally {
        consumer.close();
        System.out.println("Closed consumer and we are done");
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

❶ 首先，需要实现ConsumerRebalanceListener接口。
❷ 在这个例子中，在分配到新分区时我们不做任何事情，直接开始读取消息。
❸ 如果发生了再均衡，则要在即将失去分区所有权时提交偏移量。我们提交的是所有分区而不只是那些即将失去所有权的分区的偏移量——因为我们提交的是已处理过的消息的偏移量，所以不会有什么问题。况且，我们会使用commitSync()方法确保在再均衡发生之前提交偏移量。
❹ 把ConsumerRebalanceListener对象传给subscribe()方法，这样消费者才能调用它，这是非常重要的一步。

从特定偏移量位置读取记录

如果想从分区的起始位置读取所有的消息，或者直接跳到分区的末尾读取新消息，那么Kafka API分别提供了两个方法：seekToBeginning(Collection<Topic Partition>tp) 和seekToEnd(Collection<TopicPartition> tp)。

Kafka还提供了用于查找特定偏移量的API。这个API有很多用途，比如，对时间敏感的应用程序在处理速度滞后的情况下可以向前跳过几条消息，或者如果消费者写入的文件丢失了，则它可以重置偏移量，回到某个位置进行数据恢复。

下面的例子演示了如何将分区的当前偏移量定位到在指定时间点生成的记录:

Long oneHourEarlier = Instant.now().atZone(ZoneId.systemDefault())
          .minusHours(1).toEpochSecond();
Map<TopicPartition, Long> partitionTimestampMap = consumer.assignment()
        .stream()
        .collect(Collectors.toMap(tp -> tp, tp -> oneHourEarlier)); ➊
Map<TopicPartition, OffsetAndTimestamp> offsetMap
        = consumer.offsetsForTimes(partitionTimestampMap); ➋

for(Map.Entry<TopicPartition,OffsetAndTimestamp> entry: offsetMap.entrySet()) {
    consumer.seek(entry.getKey(), entry.getValue().offset()); ➌
}
1
2
3
4
5
6
7
8
9
10
11

❶ 首先，创建一个map，将所有分配给这个消费者的分区（通过调用consumer.assignment()）映射到我们想要回退到的时间戳。
❷ 然后，通过时间戳获取对应的偏移量。这个方法会向broker发送请求，通过时间戳获取对应的偏移量。
❸ 最后，将每个分区的偏移量重置成上一步返回的偏移量。

优雅退出循环

如果我们确定马上要关闭消费者（即使消费者还在等待一个poll()返回），那么可以在另一个线程中调用consumer.wakeup()。

如果轮询循环运行在主线程中，那么可以在ShutdownHook里调用这个方法。

需要注意的是，consumer.wakeup()是消费者唯一一个可以在其他线程中安全调用的方法。调用consumer.wakeup()会导致poll()抛出WakeupException，如果调用consumer.wakeup()时线程没有在轮询，那么异常将在下一次调用poll()时抛出。不一定要处理WakeupException，但在退出线程之前必须调用consumer.close()。

消费者在被关闭时会提交还没有提交的偏移量，并向消费者协调器发送消息，告知自己正在离开群组。协调器会立即触发再均衡，被关闭的消费者所拥有的分区将被重新分配给群组里其他的消费者，不需要等待会话超时。

Runtime.getRuntime().addShutdownHook(new Thread() {
    public void run() {
        System.out.println("Starting exit...");
        consumer.wakeup(); ➊
        try {
            mainThread.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
});

...
Duration timeout = Duration.ofMillis(10000); ➋

try {
    // 一直循环，直到按下Ctrl-C组合键，关闭钩子会在退出时做清理工作
    while (true) {
        ConsumerRecords<String, String> records =
            movingAvg.consumer.poll(timeout);
        System.out.println(System.currentTimeMillis() +
            "-- waiting for data...");
        for (ConsumerRecord<String, String> record : records) {
            System.out.printf("offset = %d, key = %s, value = %s\n",
                record.offset(), record.key(), record.value());
        }
        for (TopicPartition tp: consumer.assignment())
            System.out.println("Committing offset at position:" +
                consumer.position(tp));
        movingAvg.consumer.commitSync();
    }
} catch (WakeupException e) {
    // 忽略异常 ➌
} finally {
    consumer.close(); ➍
    System.out.println("Closed consumer and we are done");
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

❶ ShutdownHook运行在单独的线程中，所以退出轮询循环最安全的方式只能是调用wakeup()。
❷ 一个比较长的轮询超时时间。如果轮询的时间足够短，并且不介意在退出之前等一小会儿，那么就没有必要调用wakeup()，只需在每次轮询时检查一下原子布尔变量即可。较长的轮询超时时间在处理低吞吐量主题时比较有用，因为当broker没有新数据返回时，客户端在轮询时占用的CPU时间会更少。
❸ 因为在另一个线程中调用了wakeup()，所以poll()会抛出WakeupException。你可能想捕获异常，确保应用程序不会意外退出，但实际上在这里无须对它做任何处理。
❹ 在退出之前，确保彻底关闭了消费者。