Kafka系列3：深入理解Kafka消费者 (2)

消息轮询是消费者API的核心，消费者通过轮询 API(poll) 向服务器定时请求数据。一旦消费者订阅了主题，轮询就会处理所有的细节，包括群组协调、分区再均衡、发送心跳和获取数据，这使得开发者只需要关注从分区返回的数据，然后进行业务处理。
一个简单的消费者消费的代码样例如下：

1
2
3
4
5
6
7
8
9
10
11
12
  try {
while (true) {
// 轮询获取数据
ConsumerRecords<String, String> records = consumer.poll(Duration.of(100, ChronoUnit.MILLIS));
for (ConsumerRecord<String, String> record : records) {
System.out.printf("topic = %s,partition = %d, key = %s, value = %s, offset = %d,\n",
record.topic(), record.partition(), record.key(), record.value(), record.offset());
}
}
} finally {
consumer.close();
}
 

 

消费者配置

与生产者类似，消费者也有完整的配置列表。接下来一一介绍这些重要的属性。

fetch.min.byte

消费者从服务器获取记录的最小字节数。如果可用的数据量小于设置值，broker 会等待有足够的可用数据时才会把它返回给消费者。主要是为了降低消费者和Broker的工作负载。

fetch.max.wait.ms

broker 返回给消费者数据的等待时间，默认是 500ms。如果消费者获取最小数据量的要求得不到满足，就会在等待最多该属性所设置的时间后获取到数据。实际要看二者哪个条件先满足。

max.partition.fetch.bytes

该属性指定了服务器从每个分区返回给消费者的最大字节数，默认为 1MB。

session.timeout.ms

消费者在被认为死亡之前可以与服务器断开连接的时间，默认是 3s。

auto.offset.reset

该属性指定了消费者在读取一个没有偏移量的分区或者偏移量无效的情况下该作何处理：

latest (默认值) ：在偏移量无效的情况下，消费者将从最新的记录开始读取数据（在消费者启动之后生成的最新记录）;

earliest ：在偏移量无效的情况下，消费者将从起始位置读取分区的记录。 enable.auto.commit 是否自动提交偏移量，默认值是 true。为了避免出现重复消费和数据丢失，可以把它设置为 false。 client.id 客户端 id，服务器用来识别消息的来源。 max.poll.records 单次调用 poll() 方法能够返回的记录数量。 receive.buffer.bytes & send.buffer.byte 这两个参数分别指定 TCP socket 接收和发送数据包缓冲区的大小，-1 代表使用操作系统的默认值。 提交和偏移量 提交是指更新分区当前位置的操作，分区当前的位置，也就是所谓的偏移量。 什么是偏移量 Kafka 的每一条消息都有一个偏移量属性，记录了其在分区中的位置，偏移量是一个单调递增的整数。消费者通过往一个叫作 ＿consumer_offset 的特殊主题发送消息，消息里包含每个分区的偏移量。 如果消费者一直处于运行状态，那么偏移量就没有 什么用处。不过，如果有消费者退出或者新分区加入，此时就会触发再均衡。完成再均衡之后，每个消费者可能分配到新的分区，而不是之前处理的那个。为了能够继续之前的工作，消费者需要读取每个分区最后一次提交的偏移量，然后从偏移量指定的地方继续处理。 因为这个原因，所以如果不能正确提交偏移量，就可能会导致数据丢失或者重复出现消费，比如下面情况：

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

如果提交的偏移量大于客户端处理的最后一个消息的偏移量，那么处于两个偏移量之间的消息将会丢失。 偏移量提交 那么消费者如何提交偏移量呢？
Kafka 支持自动提交和手动提交偏移量两种方式。 自动提交： 只需要将消费者的 enable.auto.commit 属性配置为 true 即可完成自动提交的配置。 此时每隔固定的时间，消费者就会把 poll() 方法接收到的最大偏移量进行提交，提交间隔由 auto.commit.interval.ms 属性进行配置，默认值是 5s。
使用自动提交是存在隐患的，假设我们使用默认的 5s 提交时间间隔，在最近一次提交之后的 3s 发生了再均衡，再均衡之后，消费者从最后一次提交的偏移量位置开始读取消息。这个时候偏移量已经落后了 3s ，所以在这 3s 内到达的消息会被重复处理。可以通过修改提交时间间隔来更频繁地提交偏移量，减小可能出现重复消息的时间窗，不过这种情况是无法完全避免的。基于这个原因，Kafka 也提供了手动提交偏移量的 API，使得用户可以更为灵活的提交偏移量。 手动提交： 用户可以通过将 enable.auto.commit 设为 false，然后手动提交偏移量。基于用户需求手动提交偏移量可以分为两大类：
手动提交当前偏移量：即手动提交当前轮询的最大偏移量；
手动提交固定偏移量：即按照业务需求，提交某一个固定的偏移量。
而按照 Kafka API，手动提交偏移量又可以分为同步提交和异步提交。
同步提交：
通过调用 consumer.commitSync() 来进行同步提交，不传递任何参数时提交的是当前轮询的最大偏移量。 1
2
3
4
5
6
7
8
  while (true) {
ConsumerRecords<String, String> records = consumer.poll(Duration.of(100, ChronoUnit.MILLIS));
for (ConsumerRecord<String, String> record : records) {
System.out.println(record);
}
// 同步提交
consumer.commitSync();
}