初始 Kafka Consumer 消费者

舒适提示：整个 Kafka 专栏基于 kafka-2.2.1 版本。

一、KafkaConsumer 概述

根据 KafkaConsumer 类上的注释上来看 KafkaConsumer 具备以下特征：

• 在 Kafka 中 KafkaConsumer 是线程不安全的。
• 2.2.1 版本的KafkaConsumer 兼容 kafka 0.10.0 和 0.11.0 等低版本。
• 消息偏移量与消费偏移量(消息消费进度)
  Kafka 为分区中的每一条消息维护一个偏移量，即消息偏移量。这个偏移量充当该分区内记录的惟一标识符。消费偏移量(消息消费进度)存储的是消费组当前的处理进度。消息消费进度的提交在 kafka 中能够定时自动提交也能够手动提交。手动提交能够调用 ommitSync() 或 commitAsync 方法。
• 消费组 与 订阅关系
  多个消费这能够同属于一个消费组，消费组内的全部消费者共同消费主题下的全部消息。一个消费组能够订阅多个主题。

• 队列负载机制
  既然同一个消费组内的消费者共同承担主题下全部队列的消费，那他们如何进行分工呢？默认状况下采起平均分配，例如一个消费组有两个消费者c一、c2，一个 topic 的分区数为6，那 c1 会负责3个分区的消费，一样 c2 会负责另外3个分区的分配。

  那若是其中一个消费者宕机或新增一个消费者，那队列能动态调整吗？

  答案是会从新再次平衡，例如若是新增一个消费者 c3，则c1,c2,c3都会负责2个分区的消息消费，分区重平衡会在后续文章中重点介绍。消费者也能够经过 assign 方法手动指定分区，此时会禁用默认的自动分配机制。

• 消费者故障检测机制
  当经过 subscribe 方法订阅某些主题时，此时该消费者还未真正加入到订阅组，只有当 consumeer#poll 方法被调用后，而且会向 broker 定时发送心跳包，若是 broker 在 session.timeout.ms 时间内未收到心跳包，则 broker 会任务该消费者已宕机，会将其剔除，并触发消费端的分区重平衡。

  消费者也有可能遇到“活体锁”的状况，即它继续发送心跳，但没有任何进展。在这种状况下，为了防止消费者无限期地占用它的分区，可使用max.poll.interval.ms 设置提供了一个活性检测机制。基本上，若是您调用轮询的频率低于配置的最大间隔，那么客户机将主动离开组，以便另外一个消费者能够接管它的分区。当这种状况发生时,您可能会看到一个偏移提交失败(由调用{@link #commitSync()}抛出的{@link CommitFailedException}表示)。

• kafka 对 poll loop 行为的控制参数
  Kafka 提供了以下两个参数来控制 poll 的行为：
  • max.poll.interval.ms
    容许 两次调用 poll 方法的最大间隔，即设置每一批任务最大的处理时间。
  • max.poll.records
    每一次 poll 最大拉取的消息条数。

  对于消息处理时间不可预测的状况下上述两个参数可能不够用，那将如何是好呢？

  一般的建议将消息拉取与消息消费分开，一个线程负责 poll 消息，处理这些消息使用另外的线程，这里就须要手动提交消费进度。为了控制消息拉起的过快，您可能会须要用到 Consumer#pause(Collection) 方法，暂时中止向该分区拉起消息。RocketMQ 的推模式就是采用了这种策略。若是你们有兴趣的话，能够从笔者所著的《RocketMQ技术内幕》一书中详细了解。

二、KafkaConsume 使用示例

2.1 自动提交消费进度

public static void testConsumer1() {
    Properties props = new Properties();
    props.setProperty("bootstrap.servers", "localhost:9092,localhost:9082,localhost:9072");
    props.setProperty("group.id", "C_ODS_ORDERCONSUME_01");
    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("TOPIC_ORDER"));
    while (true) {
        ConsumerRecords<String, String>  records = consumer.poll(Duration.ofMillis(100));
        for (ConsumerRecord<String, String> record : records) {
            System.out.println("消息消费中");
            System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
        }
    }
}

2.2 手动提交消费进度

public static void testConsumer2() {
        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("foo", "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();
            }
        }
    }

三、认识 Consumer 接口

要认识 Kafka 的消费者，我的认为最好的办法就是从它的类图着手，下面给出 Consumer 接口的类图。

接下来对起重点方法进行一个初步的介绍，从下篇文章开始将对其进行详细设计。

• Set< TopicPartition> assignment()
  获取该消费者的队列分配列表。
• Set< String> subscription()
  获取该消费者的订阅信息。
• void subscribe(Collection< String> topics)
  订阅主题。
• void subscribe(Collection< String> topics, ConsumerRebalanceListener callback)
  订阅主题，并指定队列重平衡的监听器。
• void assign(Collection< TopicPartition> partitions)
  取代 subscription，手动指定消费哪些队列。
• void unsubscribe()
  取消订阅关系。
• ConsumerRecords<K, V> poll(Duration timeout)
  拉取消息，是 KafkaConsumer 的核心方法，将在下文详细介绍。
• void commitSync()
  同步提交消费进度，为本批次的消费提交，将在后续文章中详细介绍。
• void commitSync(Duration timeout)
  同步提交消费进度，可设置超时时间。
• void commitSync(Map<TopicPartition, OffsetAndMetadata> offsets)
  显示同步提交消费进度， offsets 指明须要提交消费进度的信息。
• void commitSync(final Map<TopicPartition, OffsetAndMetadata> offsets, final Duration timeout)
  显示同步提交消费进度，带超时间。
• void seek(TopicPartition partition, long offset)
  重置 consumer#poll 方法下一次拉消息的偏移量。
• void seek(TopicPartition partition, OffsetAndMetadata offsetAndMetadata)
  seek 方法重载方法。
• void seekToBeginning(Collection< TopicPartition> partitions)
  将 poll 方法下一次的拉取偏移量设置为队列的初始偏移量。
• void seekToEnd(Collection< TopicPartition> partitions)
  将 poll 方法下一次的拉取偏移量设置为队列的最大偏移量。
• long position(TopicPartition partition)
  获取将被拉取的偏移量。
• long position(TopicPartition partition, final Duration timeout)
  同上。
• OffsetAndMetadata committed(TopicPartition partition)
  获取指定分区已提交的偏移量。
• OffsetAndMetadata committed(TopicPartition partition, final Duration timeout)
  同上。
• Map<MetricName, ? extends Metric> metrics()
  统计指标。
• List< PartitionInfo> partitionsFor(String topic)
  获取主题的路由信息。
• List< PartitionInfo> partitionsFor(String topic, Duration timeout)
  同上。
• Map<String, List< PartitionInfo>> listTopics()
  获取全部 topic 的路由信息。
• Map<String, List< PartitionInfo>> listTopics(Duration timeout)
  同上。
• Set< TopicPartition> paused()
  获取已挂起的分区信息。
• void pause(Collection< TopicPartition> partitions)
  挂起分区，下一次 poll 方法将不会返回这些分区的消息。
• void resume(Collection< TopicPartition> partitions)
  恢复挂起的分区。
• Map<TopicPartition, OffsetAndTimestamp> offsetsForTimes(Map<TopicPartition, Long> timestampsToSearch)
  根据时间戳查找最近的一条消息的偏移量。
• Map<TopicPartition, OffsetAndTimestamp> offsetsForTimes(Map<TopicPartition, Long> timestampsToSearch, Duration timeout)
  同上。
• Map<TopicPartition, Long> beginningOffsets(Collection< TopicPartition> partitions)
  查询指定分区当前最小的偏移量。
• Map<TopicPartition, Long> beginningOffsets(Collection< TopicPartition> partitions, Duration timeout)
  同上。
• Map<TopicPartition, Long> endOffsets(Collection< TopicPartition> partitions)
  查询指定分区当前最大的偏移量。
• Map<TopicPartition, Long> endOffsets(Collection< TopicPartition> partitions, Duration timeout)
  同上。
• void close()
  关闭消费者。
• void close(Duration timeout)
  关闭消费者。
• void wakeup()
  唤醒消费者。

四、初始 KafkaConsumer

接下来笔者根据其构造函数，对一一介绍其核心属性的含义，为接下来说解其核心方法打下基础。

• String groupId
  消费组ID。同一个消费组内的多个消费者共同消费一个主题下的消息。
• String clientId
  发出请求时传递给服务器的id字符串。设置该值的目的是方便在服务器端请求日志中包含逻辑应用程序名称，从而可以跟踪ip/端口以外的请求源。该值能够设置为应用名称。
• ConsumerCoordinator coordinator
  消费协调器，后续会详细介绍。
• Deserializer< K> keyDeserializer
  key 序列化器。
• Deserializer< V> valueDeserializer
  值序列化器。
• ConsumerNetworkClient client
  网络通信客户端。
• SubscriptionState subscriptions
  用于管理订阅状态的类，用于跟踪 topics, partitions, offsets 等信息。后续会详细介绍。
• ConsumerMetadata metadata
  消费者元数据信息，包含路由信息。
• long retryBackoffMs
  若是向 broker 发送请求失败后，发起重试以前须要等待的间隔时间，经过属性 retry.backoff.ms　指定。
• long requestTimeoutMs
  一次请求的超时时间。
• int defaultApiTimeoutMs
  为全部可能阻塞的API设置一个默认的超时时间。
• List< PartitionAssignor> assignors
  分区分配算法（分区负载算法）。

Kafka Consumer 消费者就介绍到这里了，从下篇文章开始将开始详细介绍 Kafka 关于消息消费的方方面面。

---

做者介绍：
丁威，《RocketMQ技术内幕》做者，RocketMQ 社区布道师，公众号：中间件兴趣圈 维护者，目前已陆续发表源码分析Java集合、Java 并发包(JUC)、Netty、Mycat、Dubbo、RocketMQ、Mybatis等源码专栏。欢迎加入个人知识星球，构建一个高质量的技术交流社群。