Kafka（四） Consumer消费者

一，基础知识

1，消费者与消费组

每个消费者都有对应的消费组，不同消费组之间互不影响。
Partition的消息只能被一个消费组中的一个消费者所消费，但Partition也可能被再平衡分配给新的消费者。
一个Topic的不同Partition会根据分配策略（消费者客户端参数partition.assignment strategy）分给不同消费者。

2，Kafka的消息模式

消息中间件一般有两种消息投递模式：点对点模式和发布／订阅模式，Kafka 同时支持两种。

如果所有的消费者都属于同一消费组，那么所有的消息都会被均衡地投递给每个消费者，即每条消息只会被一个消费者处理，这就相当于点对点模式；
如果所有的消费者都隶属于不同的消费组，那么所有的消息都会被广播给所有的消费者，即每条消息会被所有的消费者处理，这就相当于发布／订阅模式；

二，Client开发

1，消费逻辑需要具备以下几个步骤

配置消费者参数及创建消费者实例
订阅主题
拉取消息并消费
提交消费位移
关闭消费者实例

public class Consumer {
    private static final String BROKER_LIST = "localhost:9092";
    private static final String TOPIC = "TOPIC-A";
    private static final String GROUP_ID = "GROUP-A";
    private static final AtomicBoolean IS_RUNNING = new AtomicBoolean(true);

    public static Properties initConfig() {
        Properties properties = new Properties();
        // 以下3个必须
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BROKER_LIST);
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // 客户端ID
        properties.put(ConsumerConfig.CLIENT_ID_CONFIG, "eris-kafka-consumer");
        // 消费组ID
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, GROUP_ID);
        // 自动提交，默认为true
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
        return properties;
    }


    public static void main(String[] args) {
        Properties properties = initConfig();
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(properties);
        kafkaConsumer.subscribe(Arrays.asList(TOPIC));

        try {
            while (IS_RUNNING.get()) {
              // poll内部封装了消费位移提交、消费者协调器、组协调器、消费者的选举、分区分配与再均衡、心跳等
                // Duration用来控制在消费者的缓冲区里没有可用数据时阻塞等待的时间，0表示不等待直接返回
                ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(1000));
                for (ConsumerRecord<String, String> r : records) {
                    print("topic:" + r.topic() + ", patition:" + r.partition() + ", offset:" + r.offset());
                    print("key:" + r.key() + ", value:" + r.value());
                }
            }
        } catch (WakeupException e) {
            // wakeup方法是KafkaConsumer中唯一可以从其他线程里安全调用的方法，调用wakeup后可以退出poll的逻辑，并抛出WakeupException。我们也不需处理WakeupException，它只是一种跳出循环的方式。
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // maybe commit offset.
            kafkaConsumer.close();
        }
    }
}

注意：KafkaConsumer是非线程安全的，wakeup()方法是 KafkaConsume 中唯一可以从其他线程里安全调用的方法。

2，subscribe有4个重载方法

public void subscribe(Collection<String> topics, ConsumerRebalanceListener listener)
public void subscribe(Collection<String> topics)

// 在之后如果又创建了新主题，并且与正表达式相匹配，那么这个消费者也可以消费到新添加的Topic
public void subscribe (Pattern pattern, ConsumerRebalanceListener listener)
public void subscribe (Pattern pattern)

3，assign订阅指定的分区

消费者不仅可以通过 subscribe方法订阅，还可以直接订阅指定分区，如下：

consumer.assign(Arrays.asList(new TopicPartition ("topic-demo", 0))) ;

通过subscribe订阅主题具有消费者自动再均衡功能，可以根据分区分配策略来自动分配各个消费者与分区的关系；

通过assign来订阅分区时，是不具备消费者自动均衡的功能的。

4，取消订阅

以下三行代码等效。

consumer.unsubscribe();
consumer.subscribe(new ArrayList<String>()) ;
consumer.assign(new ArrayList<TopicPartition>());

5，消息消费

消费者消费到的每条消息的类型为ConsumerRecord，这个和生产者发送的ProducerRecord相对应。

注意与ConsumerRecords区别，ConsumerRecords实现了Iterable，是poll返回的对象。

public class ConsumerRecord<K, V> {
    private final String topic;
    private final int partition;
    pr vate final long offset;
    private final long timestamp;
    private final TimestampType timestampType;
    private final int serializedKeySize;
    private final int serializedValueSize;
    private final Headers headers;
    private final K key;
    private final V value;
    private volatile Long checksum;
    ／／省略若干方法
}

根据Partition或Topic来消费消息

① 根据分区对当前批次消息分类：public List<ConsumerRecord<K, V> records(TopicPart tion partition)

for (TopicPartition tp : records.partitions()) {
    for (ConsumerRecord<String, String> record : records.records(tp)) {
        System.out.println(record.partition() + ":" + record.value());
    }
}

② 根据主题对当前批次消息分类：public Iterable<ConsumerRecord<K, V> records(String topic)

// ConsumerRecords类中并没提供与partitions()类似的topics()方法来查看拉取的消息集中所含的主题列表。
for (String topic : Arrays.asList(TOPIC)) {
    for (ConsumerRecord<String, String> record : records.records(topic)) {
        System.out.println(record.topic() + ":" + record.value());
    }
}

6，反序列化

ConsumerRecord里的value，就是经过反序列化后的业务对象。

Kafka所提供的反序列器有ByteBufferDeserializer、ByteArrayDeserializer、BytesDeserializer、DoubleDeserializer、FloatDeserializer、IntegerDeserializer、LongDeserializer、ShortDeserializer、StringDeserializer。

在实际应用中，在Kafka提供的序列化器和反序列化器满足不了应用需求的前提下，推荐使用 Avro、JSON、Thrift、 ProtoBuf、Protostuff等通用的序列化工具来包装，不建议使用自定义的序列化器或反序列化器。

三，位移提交

1，消费位移提交

在每次调用poll方法时，返回的是还没有被消费过的消息集。

消费位移必须做持久化保存，否则消费者重启之后就无法知晓之前的消费位移。再者，当有新的消费者加入时，那么必然会再均衡，某个分区可能在再均衡之后分配给新的消费者，如果不持久化保存消费位移，那么这个新消费者也无法知晓之前的消费位移。

在旧消费者客户端中，消费位移存储在ZooKeeper中，而在新消费者客户端中则存储在Kafka内部的主题 __consumer_offsets中。

这里把将消费位移持久化的动作称为“提交” ，消费者在消费完消息之后需要执行消费位移的提交。

2，三个位移的关系

lastConsumedOffset：当前消费到的位置，即poll拉到的该分区最后一条消息的offset
committed offset：提交的消费位移
position：下次拉取的位置

position = committed offset = lastConsumedOffset + 1（当然position和committed offset 并不会一直相同）

TopicPartition tp = new TopicPartition("topic", 0);
kafkaConsumer.assign(Arrays.asList(tp));
long lastConsumedOffset = 0;

while (true) {
    ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofMillis(1000));
    List<ConsumerRecord<String, String>> partitionRecords = consumerRecords.records(tp);
    lastConsumedOffset = partitionRecords.get(partitionRecords.size() - 1).offset();
    // 同步提交消费位移
    kafkaConsumer.commitSync();
   
    System.out.println("consumed off set is " + lastConsumedOffset);
    OffsetAndMetadata offsetAndMetadata = kafkaConsumer.committed(tp);
    System.out.println("commited offset is " + offsetAndMetadata.offset());
    long posititon = kafkaConsumer.position(tp);
    System.out.println("he offset of t he next record is " + posititon);
}

输出结果：
consumed offset is 377
commited offset is 378
the offset of the next record is 378

3，消息丢失与重复消费

如果poll后立马提交位移，之后业务异常，再次拉取就从新位移开始，就丢失了数据。
如果poll后先处理数据，处理到一半异常了，或者最后提交位移异常，重新拉取会从之前的位移拉，就重复消费了。

4，自动提交位移原理

Kafka中默认的消费位移的提交方式是自动提交，这个由消费者客户端参数enable.auto.commit配置，默认值为true。

不是每消费一条消息就提交一次，而是定期提交，这个定期的周期时间由客户端参数auto.commit.interval.ms配置，默认值为5秒，此参数生效的前提是enable.auto.commit参数为true。

在默认的方式下，消费者每隔5秒会将拉取到的每个分区中最大的消息位移进行提交。动位移提交的动作是在poll()方法的逻辑里完成的，在每次真正向服务端发起拉取请求之前会检查是否可以进行位移提交，如果可以，那么就会提交上一次轮询的位移。

自动提交让编码更简洁，但随之而来的是重复消费和消息丢失的问题。

5，手动提交位移

① 同步提交

ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(1000));
for (ConsumerRecord<String, String> record : records) {
    //do some logical processing 
    kafkaConsumer.commitSync();
}

1）commitSync会根据poll拉取的最新位移来进行提交（注意提交的值对应于图3-6 position的位置〉。

2）可以使用带参方法，提交指定位移：commitSync(final Map<TopicPartition OffsetAndMetadata> offsets)

3）没必要每条消息提交一次，可以改为批量提交。

② 异步提交

public void commitAsync()
public void commitAsync(OffsetCommitCallback callback)
public void commitAsync(final Map<TopicPartition, OffsetAndMetadata> offsets, OffsetCommitCallback callback)

1）异步提交在执行的时候消费者线程不会被阻塞，可能在提交消费位移的结果还未返回之前就开始了新一次的poll操作。

2）提交也可能会异常，可引入重试机制。但重试可能出现问题，若第一次commitAsync失败在重试，第二次成功了，然后第一次重试也成功了，就会覆盖位移为之前。解决方案：可以提交时维护一个序号，如果发现过期的序号就不再重试。

总结：不管是自动提交还是手动提交（同步、异步），都可能出现漏消费和重复消费，一般情况提交位移这一步操作很少失败，至于业务异常如何影响提交，需要结合具体情况分析。

可以引入重试机制，重试提交或者业务处理。但重试会增加代码逻辑复杂度。

还有对于消费者异常退出，重复消费的问题就很难避免，因为这种情况下无法及时提交消费位移，需要消费者的幂等处理。

如果消费者正常退出或发生再均衡况，那么可以在退出或再均衡执行之前使用同步提交的方式做最后的把关。

try {
    while (IS_RUNNING.get()) {
        // poll records and do some logical processing .
        kafkaConsumer.commitAsync();
    }
} finally {
    try {
        kafkaConsumer.commitSync();
    } finally {
        kafkaConsumer.close();
    }
}

四，暂停或恢复消费

暂停某些分区在poll时返回数据给客户端和恢复某些分区返回数据给客户端。

public void pause(Collection<TopicPartition> partitions)
public roid resume(Collection<TopicPartition> partitions)

五，指定位移消费

有了消费位移的持久化，才使消费者在关闭、崩溃或者在遇到再均衡的时候，可以让接替的消费者能够根据存储的消费位移继续进行消费。

当新的消费组建立的时候，它根本没有可以查找的消费位移。或者消费组内的新消费者订阅了一个新的主题，它也没有可以查找的消费位移。

auto.offset.reset 配置

① 作用：

决定从何处消费；

② 何时生效：

找不到消费位移记录时；
位移越界时（seek）；

③ 配置值：

（默认值）auto.offset.reset=latest，从分区末尾开始消费；
auto.offset.reset=earliest，从分区起始开始消费;

seek方法

① 定义：指定消费的位置，可以向前跳过若干消息或回溯消息。

// partition：分区，offset：从哪个位置消费 
public void seek(TopicPartition partition, long offset)

② seek方法只能重置消费者分配到的分区的消费位置，而分区的分配是在 poll方法过程中实现的，也就是说在执行seek之前需要先执行一次poll，等到分配到分区之后才可以重置消费位置。

Set<TopicPartition> assignment = new HashSet<>();
while (assignment.size() == 0) {
    // 如采不为空，则说明已经成功分配到了分区
    kafkaConsumer.poll(Duration.ofMillis(1000));
    assignment = kafkaConsumer.assignment();
}


for (TopicPartition tp : assignment) {
    kafkaConsumer.seek(tp, 10);
}
while (true) {
    ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(1000));
    //consume the record
}

③ 如果对当前消费者未分配到的分区执行 seek方法，那么会报IllegalStateException。

④ 如果消费者启动能找到位移记录，又想从头或者尾消费，可以通过seek结合endOffsets、beginningOffsets或者直接seekToBeginning、seekToEnd实现。

注意：分区的起始位置是0，但并不代表每时每刻都为0，因为日志清理会清理旧的数据，所以分区的起始位置自然会增加。

⑤ 按时间戳指定消费位置

public Map<TopicPartition, OffsetAndTimestamp> offsetsForTimes(Map<TopicPartition, Long> timestampsToSearch)
public Map<TopicPartition, OffsetAndTimestamp> offsetsForTimes(Map<TopicPartition, Long> timestampsToSearch, Duration timeout)

给定待查分区和时间戳，返回大于等于该时间戳的第一条消息对应的offset和timestamp，对应于OffsetAndTimestamp中的offset、timestamp字段。

⑥ 将分区消费位移存储在数据库、文件等外部介质，再通过seek指定消费，可以配合再均衡监听器来实现新消费者的继续消费。

六，消费再均衡

再均衡是指分区的所属权从一个消费者转移到另一消费者的行为。它为消费组具备高可用性和伸缩性提供保障，使我们可以既方便又安全地删除消费组内的消费者或往消费组内添加消费者。

不过，在再均衡发生期间的这一小段时间，消费组会变得不可用。而且再均衡容易导致重复消费等问题，一般情况应尽量避免不必要的再均衡。

再均衡监听器 ConsumerRebalanceListener

注册：
subscribe(Collection<String> topics, ConsumerRebalanceListener listener) 和 subscribe(Patten pattern, ConsumerRebalanceListener listener)


ConsumerRebalanceListener是一个接口，有2个方法。

(1) void onPartitionsRevoked(Collection<TopicPartition> partitions)
再均衡开始之前和消费者停止读取消息之后被调用。

(2) void onPartitionsAssigned(Collection<TopicPartition> partitions)
新分配分区之后和消费者开始拉取消费之前被调用 。

示例①：消费时把位移暂存在Map，再均衡之前同步提交，避免发送再均衡的同时，异步提交还没提交上去（重复消费）。

Map<TopicPartition, OffsetAndMetadata> currentOffsets = new HashMap<>();
kafkaConsumer.subscribe(Arrays.asList("topic"), new ConsumerRebalanceListener() {
    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        kafkaConsumer.commitSync(currentOffsets);
        currentOffsets.clear();
    }
    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        //do nothing .
    }
});
ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(1000));
for (ConsumerRecord<String, String> record : records) {
    //process the record
    currentOffsets.put(
            new TopicPartition(record.topic(), record.partition()),
            new OffsetAndMetadata(record.offset() + 1));
}
kafkaConsumer.commitAsync(currentOffsets, null);

示例②：把位移存在db，再均衡后通过seek定位继续消费

kafkaConsumer.subscribe(Arrays.asList(topic), new ConsumerRebalanceListener() {
    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        // store offset in DB
    }

    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        for (TopicPartition tp : partitions) {
            // 从DB中读取消费位移
            kafkaConsumer.seek(tp, getOffsetFromDB(tp));
        }
    }
}

七，消费者拦截器

与生产者拦截器对应，消费者拦截器需要自定义实现org.apache.kafka.clients.consumer.Consumerlnterceptor接口。

Consumerlnterceptor有3个方法：

// poll方法返回之前调用，可以修改返回的消息内容、按照某种规则过滤消息等
public ConsumerRecords<K, V> onConsume(ConsumerRecords<K , V> records);

// 提交完消费位移之后调用，可以用来记录跟踪所提交的位移信息，比如当使用commitSync的无参方法时，我们不知道提交的消费位移，而onCommit方法却可以做到这一点
public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets);

public void close();

在消费者中也有拦截链的概念，和生产者的拦截链一样，也是按照interceptor.classes参数配置的拦截器的顺序来一一执行的。如果在拦截链中某个拦截器执行失败，那么下一个拦截器会接着从上一个执行成功的拦截器继续执行。