Kafka 之消费者（Consumer）

一. 前言

二. 消费示例

2.1. 自动提交偏移量（Automatic Offset Committing）

2.2. 手动控制偏移量（Manual Offset Control）

2.3. 订阅指定的分区（Manual Partition Assignment）

三. 消费核心说明

3.1. 跨版本兼容性

3.2. 偏移量（offset）和消费者位置

3.3. 消费者组和主题订阅

3.4. 发现消费者故障

3.5. 在 Kafka 之外存储偏移量

3.6. 控制消费的位置

3.7. 消费者流量控制

3.8. 读取事务性消息

3.9. 多线程处理

一. 前言

消息的消费一般有两种模式，推模式和拉模式。推模式是服务端主动将消息推送给消费者，而拉模式是消费者主动向服务端发起请求来拉取消息。Kakfa 采用的是拉模式，这样可以很好的控制消费速率。

Kafka 客户端从 Kafka 集群中获取消息，并透明地处理 Kafka 集群中出现故障的 Broker，透明地调节适应集群中变化的数据分区。也和 Broker交互，负载平衡消费者。

public class KafkaConsumer<K,V> extends Object implements Consumer<K,V>

消费者维护着与 Broker 的 TCP 连接来获取消息。如果在使用后没有关闭消费者，则会泄露这些连接。消费者不是线程安全的。

二. 消费示例

消费者 API 提供了灵活性，以涵盖各种消费场景，下面是一些例子来演示如何使用它们。

2.1. 自动提交偏移量（Automatic Offset Committing）

下面这个是自动提交偏移量的简单的 Kafka 消费者 API：

Properties props = new Properties();
props.setProperty("bootstrap.servers", "localhost:9092");
props.setProperty("group.id", "test");
props.setProperty("enable.auto.commit", "true");
props.setProperty("auto.commit.interval.ms", "1000");
props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("t-foo", "t-bar"));
while (true) {ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));for (ConsumerRecord<String, String> record : records)System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
}

设置 enable.auto.commit，偏移量由 auto.commit.interval.ms 控制自动提交的频率。

集群是通过配置 bootstrap.servers 指定一个或多个 Broker。不用指定全部的 Broker，它将自动发现集群中的其余的 Broker（最好指定多个，万一有服务器故障）。

在这个例子中，客户端订阅了主题 t-foo 和 t-bar。消费者组叫 test。

Broker 通过心跳机器自动检测 test 组中失败的进程，消费者会自动 ping 集群，告诉集群它还活着。只要消费者能够做到这一点，它就被认为是活着的，并保留分配给它分区的权利，如果它停止心跳的时间超过 session.timeout.ms，那么就会认为是故障的，它的分区将被分配到别的进程。

这个 deserializer 设置如何把 Byte 转成 Object 类型。例子中，通过指定 String 解析器，我们告诉获取到的消息的 key 和 value 只是简单个 String 类型。

2.2. 手动控制偏移量（Manual Offset Control）

不需要定时的提交 offset，可以自己控制 offset，当消息认为已消费过了，这个时候再去提交它们的偏移量。这个很有用的，当消费的消息结合了一些处理逻辑，这个消息就不应该认为是已经消费的，直到它完成了整个处理。

Properties props = new Properties();
props.setProperty("bootstrap.servers", "localhost:9092");
props.setProperty("group.id", "test");
props.setProperty("enable.auto.commit", "false");
props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("t-foo", "t-bar"));
final int minBatchSize = 200;
List<ConsumerRecord<String, String>> buffer = new ArrayList<>();
while (true) {ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));for (ConsumerRecord<String, String> record : records) {buffer.add(record);}if (buffer.size() >= minBatchSize) {insertIntoDb(buffer);consumer.commitSync();buffer.clear();}
}

在这个例子中，我们将消费一批消息并将它们存储在内存中。当我们积累足够多的消息后，我们再将它们批量插入到数据库中。如果我们设置 offset 自动提交（之前说的例子），消费将被认为是已消费的。这样会出现问题，我们的进程可能在批处理记录之后，但在它们被插入到数据库之前失败了。

为了避免这种情况，我们将在相应的记录插入数据库之后再手动提交偏移量。这样我们可以准确控制消息是成功消费的。提出一个相反的可能性：在插入数据库之后，但是在提交之前，这个过程可能会失败（即使这可能只是几毫秒，这是一种可能性）。在这种情况下，进程将获取到已提交的偏移量，并会重复插入的最后一批数据。这种方式就是所谓的“至少一次”保证，在故障情况下，可以重复。

如果您无法执行这些操作，可能会使已提交的偏移量超过消费的位置，从而导致缺少记录。使用手动偏移控制的优点是，您可以直接控制记录何时被视为“已消费”。

注意：使用自动提交也可以“至少一次”。但是要求你必须下次调用 poll(Duration) 之前或关闭消费者之前，处理完所有返回的数据。如果操作失败，这将会导致已提交的 offset 超过消费的位置，从而导致丢失消息。使用手动控制 offset 的优点是，你可以直接控制消息何时提交。

上面的例子使用 commitSync 表示所有收到的消息为”已提交"，在某些情况下，你可能希望更精细的控制，通过指定一个明确消息的偏移量为“已提交”。在下面，我们的例子中，我们处理完每个分区中的消息后，提交偏移量。

try {while(running) {ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(Long.MAX_VALUE));for (TopicPartition partition : records.partitions()) {List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);for (ConsumerRecord<String, String> record : partitionRecords) {System.out.println(record.offset() + ": " + record.value());}long lastOffset = partitionRecords.get(partitionRecords.size() - 1).offset();consumer.commitSync(Collections.singletonMap(partition, new OffsetAndMetadata(lastOffset + 1)));}}
} finally {consumer.close();
}

注意：已提交的 offset 应该始终是你的程序将读取的下一条消息的 offset。因此，调用commitSync(offsets) 时，你应该加1个到最后处理的消息的 offset。

2.3. 订阅指定的分区（Manual Partition Assignment）

在前面的例子中，我们订阅我们感兴趣的 Topic，让 Kafka 提供给我们平分后的 Topic 分区。但是，在有些情况下，你可能需要自己来控制分配指定分区，例如：

如果这个消费者进程与该分区保存了某种本地状态（如本地磁盘的键值存储），则它应该只能获取这个分区的消息。
如果消费者进程本身具有高可用性，并且如果它失败，会自动重新启动（可能使用集群管理框架如 YARN，Mesos，或者 AWS 设施，或作为一个流处理框架的一部分）。在这种情况下，不需要 Kafka 检测故障，重新分配分区，因为消费者进程将在另一台机器上重新启动。

要使用此模式，你只需调用 assign(collection) 消费指定的分区即可：

String topic = "t-foo";
TopicPartition partition0 = new TopicPartition(topic, 0);
TopicPartition partition1 = new TopicPartition(topic, 1);
consumer.assign(Arrays.asList(partition0, partition1));

一旦手动分配分区，你可以在循环中调用 poll（跟前面的例子一样）。消费者分组仍需要提交offset，只是现在分区的设置只能通过调用 assign 修改，因为手动分配不会进行分组协调，因此消费者故障不会引发分区重新平衡。每一个消费者是独立工作的（即使和其他的消费者共享GroupId）。为了避免 offset 提交冲突，通常你需要确认每一个 consumer 实例的 groupId 都是唯一的。

注意：手动分配分区（即 assgin）和动态分区分配的订阅 Topic模式（即 subcribe）不能混合使用。

三. 消费核心说明

3.1. 跨版本兼容性

<dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>3.6.1</version>
</dependency>

该客户端可以与 0.10.0 或更新版本的 Broker 集群进行通信。较早的版本可能不支持某些功能。例如，0.10.0 Broker 不支持 offsetsForTimes，因为此功能是在版本 0.10.1 中添加的。如果你调用Broker 版本不可用的 API 时，将报 UnsupportedVersionException 异常。

3.2. 偏移量（offset）和消费者位置

Kafka 为分区中的每条消息保存一个偏移量（offset），这个偏移量是该分区中一条消息的唯一标识。也表示消费者在分区的位置。例如，一个位置是5的消费者（说明已经消费了0到4的消息），下一个将接收消息的偏移量为5的消息。实际上这有两个与消费者相关的 “位置” 概念：

消费者的位置给出了下一条消息的偏移量。它比消费者在该分区中看到的最大偏移量要大一个。它在每次消费者调用 poll(Duration) 中接收消息时自动增长。
已提交的位置是已安全保存的最后偏移量，如果进程失败或重新启动时，消费者将恢复到这个偏移量。消费者可以选择定期自动提交偏移量，也可以选择通过调用 commit API 来手动的控制（如：commitSync 和 commitAsync）。

这个主要区别是消费者来控制一条消息什么时候才被认为是已被消费的，控制权在消费者，下面我们进一步更详细地讨论。

3.3. 消费者组和主题订阅

Kafka 的消费者组概念，通过进程池瓜分消息并处理消息。这些进程可以在同一台机器运行，也可分布到多台机器上，以增加可扩展性和容错性，相同 group.id 的消费者将视为同一个消费者组。

组中的每个消费者都通过 subscribe API 动态的订阅一个 Topic 列表。Kafka 将已订阅 Topic 的消息发送到每个消费者组中。并通过平衡分区在消费者分组中所有成员之间来达到平均。因此每个分区恰好地分配1个消费者（一个消费者组中）。所以如果一个 Topic 有4个分区，并且一个消费者分组有只有 2 个消费者。那么每个消费者将消费 2 个分区。

消费者组的成员是动态维护的。如果一个消费者故障，分配给它的分区将重新分配给同一个分组中其他的消费者。同样的，如果一个新的消费者加入到分组，将从现有消费者中分配一个给它，这被称为重新平衡分组（Rebalance），在下面更详细地讨论。当新分区添加到订阅的 Topic 时，或者当创建与订阅的正则表达式匹配的新 Topic 时，也将重新平衡。将通过定时刷新自动发现新的分区，并将其分配给分组的成员。

从概念上讲，你可以将消费者分组看作是由多个进程组成的单一逻辑订阅者。作为一个多订阅系统，Kafka 支持对于给定 Topic 任何数量的消费者组，而不重复。

这是在消息系统中常见的功能的略微概括。所有进程都将是单个消费者分组的一部分（类似传统消息传递系统中的队列的语义），因此消息传递就像队列一样，在组中平衡。与传统的消息系统不同的是，虽然你可以有多个这样的组。但每个进程都有自己的消费者组（类似于传统消息系统中pub/sub 的语义），因此每个进程都会订阅到该主题的所有消息。

此外，当分组重新分配自动发生时，可以通过 ConsumerRebalanceListener 通知消费者，这允许他们完成必要的应用程序级逻辑，例如状态清除、手动偏移提交等。有关更多详细信息，请参阅在 Kafka 之外存储偏移量。

它也允许消费者通过使用 assign(Collection) 手动分配指定分区，如果使用手动指定分配分区，那么动态分区分配和协调消费者组将失效。

3.4. 发现消费者故障

订阅一组 Topic 后，当调用 poll(long) 时，消费者将自动加入到组中。只要持续的调用 poll()，消费者将一直保持可用，并继续从分配的分区中接收消息。此外，消费者向服务器定时发送心跳。如果消费者崩溃或无法在 session.timeout.ms 配置的时间内发送心跳，则消费者将被视为死亡，并且其分区将被重新分配。

还有一种可能，消费可能遇到“活锁”的情况，它持续的发送心跳，但是没有处理。为了预防消费者在这种情况下一直持有分区，我们使用 max.poll.interval.ms 活跃检测机制。在此基础上，如果你调用的 poll() 的频率大于最大间隔，则客户端将主动地离开组，以便其他消费者接管该分区。发生这种情况时，你会看到 offset 提交失败（调用 commitSync() 引发的CommitFailedException）。这是一种安全机制，保障只有活动成员能够提交 offset。所以要留在组中，你必须持续调用 poll()。

消费者提供两个配置设置来控制 poll 循环：

max.poll.interval.ms：增大 poll 的间隔，可以为消费者提供更多的时间去处理返回的消息（调用 poll(long) 返回的消息，通常返回的消息都是一批）。缺点是此值越大将会延迟组重新平衡。
max.poll.records：此设置限制每次调用 poll 返回的消息数，这样可以更容易的预测每次 poll间隔要处理的最大值。通过调整此值，可以减少 poll 间隔，减少重新平衡分组的可能。

对于消息处理时间不可预测地的情况，这些选项是不够的。处理这种情况的推荐方法是将消息处理移到另一个线程中，让消费者继续调用 poll()。但是必须注意确保已提交的 offset 不超过实际位置。另外，你必须禁用自动提交，并只有在线程完成处理后才为记录手动提交偏移量（取决于你）。还要注意，你需要 pause 暂停分区，不会从 poll 接收到新消息，让线程处理完之前返回的消息（如果你的处理能力比拉取消息的慢，那创建新线程将导致你机器内存溢出）。

3.5. 在 Kafka 之外存储偏移量

消费者可以不使用 Kafka 内置的 offset 仓库。可以选择自己来存储 offset。要注意的是，将消费的 offset 和结果存储在同一个的系统中，用原子的方式存储结果和 offset，但这不能保证原子，要想消费是完全原子的，并提供的“正好一次”的消费保证比 Kafka 默认的“至少一次”的语义要更高。你需要使用 Kafka 的 offset 提交功能。

原子存储：

如果消费的结果存储在关系数据库中，存储在数据库的 offset，让提交结果和 offset 在单个事务中。这样，事物成功，则 offset 存储和更新。如果 offset 没有存储，那么偏移量也不会被更新。
如果 offset 和消费结果存储在本地仓库。例如，可以通过订阅一个指定的分区并将 offset 和索引数据一起存储来构建一个搜索索引。如果这是以原子的方式做的，常见的可能是，即使崩溃引起未同步的数据丢失。索引程序从它确保没有更新丢失的地方恢复，而仅仅丢失最近更新的消息。

每个消息都有自己的 offset，所以要管理自己的偏移，你只需要做到以下几点：

配置 enable.auto.commit=false。
使用提供的 ConsumerRecord 来保存你的位置。
在重启时用 seek(TopicPartition, long) 恢复消费者的位置。

当分区分配也是手动完成的，这种类型的使用是最简单的。如果分区分配是自动完成的，需要特别小心处理分区分配变更的情况。可以通过调用 subscribe(Collection，ConsumerRebalanceListener) 和 subscribe(Pattern，ConsumerRebalanceListener) 中提供的ConsumerRebalanceListener 实例来完成的。例如，当分区向消费者获取时，消费者将通过实现ConsumerRebalanceListener.onPartitionsRevoked(Collection) 来给这些分区提交它们 offset。当分区分配给消费者时，消费者通过 ConsumerRebalanceListener.onPartitionsAssigned(Collection)为新的分区正确地将消费者初始化到该位置。

ConsumerRebalanceListener 的另一个常见用法是清除应用已移动到其他位置的分区的缓存。

3.6. 控制消费的位置

大多数情况下，消费者只是简单的从头到尾的消费消息，周期性的提交偏移量（自动或手动）。Kafka 也支持消费者去手动的控制消费的偏移量，可以消费之前的消息也可以跳过最近的消息。

有几种情况，手动控制消费者的位置可能是有用的：

一种场景是对于时间敏感的消费者处理程序，对足够落后的消费者，直接跳过，从最近的消费开始消费。
另一个使用场景是本地状态存储系统（上一节说的）。在这样的系统中，消费者将要在启动时初始化它的位置（无论本地存储是否包含）。同样，如果本地状态已被破坏（假设因为磁盘丢失），则可以通过重新消费所有数据并重新创建状态（假设 Kafka 保留了足够的历史）在新的机器上重新创建。

Kafka 使用 seek(TopicPartition, long) 指定新的消费位置。用于查找服务器保留的最早和最新的offset 的特殊的方法，也可用 seekToBeginning(Collection) 和 seekToEnd(Collection)。

3.7. 消费者流量控制

如果消费者分配了多个分区，并同时消费所有的分区，这些分区具有相同的优先级。在一些情况下，消费者需要首先消费一些指定的分区，当指定的分区有少量或者已经没有可消费的数据时，则开始消费其他分区。

例如流处理，当处理器从2个 Topic 获取消息并把这两个 Topic 的消息合并，当其中一个 Topic 长时间落后另一个，则暂停消费，以便落后的赶上来。

Kafka 支持动态控制消费流量，分别在 future 的 poll(long) 中使用 pause(Collection) 和 resume(Collection) 来暂停消费指定分配的分区，重新开始消费指定暂停的分区。

3.8. 读取事务性消息

Kafka 0.11.0 中引入了事务，应用程序可以原子地写入多个主题和分区。为了使之工作，从这些分区读取的消费者应该被配置为只读取已提交的数据。这可以通过在消费者的配置中设置isolation.level=read_committed 来实现。

在 read_committed 模式下，消费者将只读取那些已经成功提交的事务性消息（像读取非事务性消息一样）。在 read_committed 模式下，没有客户端缓冲。相反，read_committed 消费者的分区的结束偏移量是分区中属于一个事务的第一个消息的偏移量。这个偏移被称为"Last Stable Offset 最后稳定偏移"（LSO）。

一个 read_committed 消费者将只读到 LSO，并过滤掉任何已经中止的事务性消息。LSO 也会影响 read_committed 消费者的 seekToEnd(Collection) 和 endOffsets(Collection) 的行为。最后，对于 read_committed 消费者来说，取数 lag（滞后指标）也被调整为相对 LSO。

带有事务性消息的分区将包括提交或中止标记，这些标记表示事务的结果。那里的标记不会返回给应用程序，但在 log 中却有一个偏移量。因此，应用程序从带有事务消息的主题中读取时，会在消耗的偏移量中看到空白。这些缺失的消息将是事务标记，它们在两个隔离级别中为消费者过滤掉。此外，使用 read_committed 消费者的应用程序也可能会看到由于中止的事务而产生的空隙，因为这些消息不会被消费者返回，但确实是有效的偏移量。

3.9. 多线程处理

Kafka 消费者不是线程安全的。所有网络 I/O 都发生在进行调用应用程序的线程中。用户的责任是确保多线程访问正确同步的。非同步访问将导致 ConcurrentModificationException。

此规则唯一的例外是 wakeup()，它可以安全地从外部线程来中断活动操作。在这种情况下，将从操作的线程阻塞并抛出一个 WakeupException。这可用于从其他线程来关闭消费者。以下代码段显示了典型模式：

public class KafkaConsumerRunner implements Runnable {private final AtomicBoolean closed = new AtomicBoolean(false);private final KafkaConsumer consumer;public KafkaConsumerRunner(KafkaConsumer consumer) {this.consumer = consumer;}@Overridepublic void run() {try {consumer.subscribe(Arrays.asList("topic"));while (!closed.get()) {ConsumerRecords records = consumer.poll(Duration.ofMillis(10000));// Handle new records}} catch (WakeupException e) {// Ignore exception if closingif (!closed.get()) throw e;} finally {consumer.close();}}// Shutdown hook which can be called from a separate threadpublic void shutdown() {closed.set(true);consumer.wakeup();}
}

在单独的线程中，可以通过设置关闭标志和唤醒消费者来关闭消费者：

closed.set(true);
consumer.wakeup();

我们没有多线程模型的例子。但以下几个操作可用来实现多线程处理消息。

1. 每个线程一个消费者

每个线程自己的消费者实例。这种方法的优点和缺点：

优点1：这是最容易实现的。
优点2：因为它不需要在线程之间协调，所以通常它是最快的。
优点3：它按顺序处理每个分区（每个线程只处理它接受的消息）。
缺点1：更多的消费者意味着更多的 TCP 连接到集群（每个线程一个）。一般 Kafka 处理连接非常的快，所以这是一个小成本。
缺点2：更多的消费者意味着更多的请求被发送到服务器，但稍微较少的数据批次可能导致 I/O 吞吐量的一些下降。
缺点3：所有进程中的线程总数受到分区总数的限制。

2. 解耦消费和处理

另一个替代方式是一个或多个消费者线程，它来消费所有数据，其消费所有数据并将ConsumerRecords 实例切换到由实际处理记录处理的处理器线程池来消费的阻塞队列。

这个选项同样有利弊：

优点1：可扩展消费者和处理进程的数量。这样单个消费者的数据可分给多个处理器线程来执行，避免对分区的任何限制。
缺点1：跨多个处理器的顺序保证需要特别注意，因为线程是独立的执行，后来的消息可能比早到的消息先处理，这仅仅是因为线程执行的运气。如果对排序没有问题，这就不是个问题。
缺点2：手动提交变得更困难，因为它需要协调所有的线程以确保处理对该分区的处理完成。

这种方法有多种玩法，例如，每个处理线程可以有自己的队列，消费者线程可以使用TopicPartition hash 到这些队列中，以确保按顺序消费，并且提交也将简化。