kafka消费者理解

概念入门

消费者、消费组

消费者：
消费者从订阅的主题topic消费消息，消费消息的偏移量保存在Kafka的名字是__consumer_offsets 的主题中。消费者还可以将⾃⼰的偏移量存储到Zookeeper，需要设置offset.storage=zookeeper。
推荐使⽤Kafka存储消费者的偏移量。因为Zookeeper不适合⾼并发。
消费组：
多个从同一个主题topic消费消息的消费者，可以加入到一个消费组中。
消费组中的消费者共享group_id ，配置相同的 configs.put("group.id", "xxx");

group_id一般设置为应用的逻辑名称。比如多个订单服务订阅一个主题topic，可以设置group_id为order_process。
group_id通过消费者的配置指定：group_id=xxxxxx

消费组均衡的给消费者分配分区，每个分区只由消费组中的一个消费者消费。
消费组中的消费者和分区的关系：每个分区只能被消费组中的一个消费者消费，一个消费者可能消费多个分区。

• 消费组数量和分区数相同时，消费组中的每个消费者分别消费一个分区
• 消费组数量<分区数时，消费组中的每个消费者消费多个分区
• 消费组数量>分区数时，多余的消费者不会分配消费的分区
  通过控制分区数和消费者数量，可以通过增加消费者来横向扩展，增加消费的能力。横向扩展增加消费者不会对性能造成负面影响，但要保证每个消费者能分配到消费分区

心跳机制

kafka的心跳机制，是消费者和kafka broken之间的健康检查，用来控制消费超时。当检查不通过时，则会触发rebalance再平衡。心跳机制对消费者客户端来说是无感的，当我们启动客户端时心跳机制就开始，它是一个异步线程。心跳机制触发rebalance再平衡，可能会导致消息重复消费

• 消费者宕机活着消费消息超时，心跳机制失败退出消费组，触发再平衡，重新给消费组中的消费者分配分区
  
• kafka broken宕机，导致某个分区没有用了。心跳机制失败触发rebalance再平衡，对应的消费者没有可消费的分区闲置。
  
  Kafka 的⼼跳是 Kafka Consumer 和 Broker 之间的健康检查，只有当 Broker Coordinator 正常时，Consumer才会发送⼼跳。
  Consumer 和 Rebalance再平衡 相关的 2 个配置参数：

心跳机制原理
do something

消息接收

必要参数配置

订阅

主题和分区

• topic：kafka用于分类管理消息的逻辑单元，类似于mysql的数据库。
• partition：是kafka下存储数据的基本单元，这是物理上的概念。一个topic可以有多个partition，他们可以分布在不同的节点上，也可以在同一个节点。优势在于：可以水平扩张，避免单台机在磁盘空间和性能上的限制。同时可以通过数据备份增加容灾能力。为了做到均匀分布，通常partition数量是节点数量的蒸熟倍
• consumer group：消费组是kafka实现单播和广播的能力。同一个消费者里面，一个分区只能有一个消费者，当消息路由到分区后只能被消费组里面的一个消费者消费。
  
  cusumer采用pull模式从broken中读取消息。
  采用pull模式，消费者就能控制消息消费的速率，可以控制消费方式（逐条/批量），还可以选择不同的提交方式从而实现不同的传输语义

反序列化

在发送消息到broken之前，会先进行序列化变成字节数组。Kafka的broker中所有的消息都是字节数组，消费者获取到消息之后，需要先对消息进⾏反序列化处理，然后才能交给⽤户程序消费处理。
消费者的反序列化器包括key的和value的反序列化器。
key.deserializer、 value.deserializer
需要实现 org.apache.kafka.common.serialization.Deserializer 接⼝。

package org.apache.kafka.common.serialization;

import org.apache.kafka.common.header.Headers;

import java.io.Closeable;
import java.util.Map;

/**
 * An interface for converting bytes to objects.
 *  * A class that implements this interface is expected to have a constructor with no parameters.
 * 

 * Implement {@link org.apache.kafka.common.ClusterResourceListener} to receive cluster metadata once it's available. Please see the class documentation for ClusterResourceListener for more information.
 *  * @param  Type to be deserialized into.
 */
public interface Deserializer<T> extends Closeable {

    /**
     * Configure this class.
     * @param configs configs in key/value pairs
     * @param isKey whether is for key or value
     */
    default void configure(Map<String, ?> configs, boolean isKey) {
        // intentionally left blank
    }

    /**
     * Deserialize a record value from a byte array into a value or object.
     * @param topic topic associated with the data
     * @param data serialized bytes; may be null; implementations are recommended to handle null by returning a value or null rather than throwing an exception.
     * @return deserialized typed data; may be null
     */
    T deserialize(String topic, byte[] data);

    /**
     * Deserialize a record value from a byte array into a value or object.
     * @param topic topic associated with the data
     * @param headers headers associated with the record; may be empty.
     * @param data serialized bytes; may be null; implementations are recommended to handle null by returning a value or null rather than throwing an exception.
     * @return deserialized typed data; may be null
     */
    default T deserialize(String topic, Headers headers, byte[] data) {
        return deserialize(topic, data);
    }

    /**
     * Close this deserializer.
     * 

     * This method must be idempotent as it may be called multiple times.
     */
    @Override
    default void close() {
        // intentionally left blank
    }
}
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
45
46
47
48
49
50
51
52
53
54

系统提供了该接⼝的⼦接⼝以及实现类：

• org.apache.kafka.common.serialization.ByteArrayDeserializer
• org.apache.kafka.common.serialization.ByteBufferDeserializer
• org.apache.kafka.common.serialization.BytesDeserializer
• org.apache.kafka.common.serialization.DoubleDeserializer
• org.apache.kafka.common.serialization.FloatDeserializer
• org.apache.kafka.common.serialization.IntegerDeserializer
• org.apache.kafka.common.serialization.LongDeserializer
• org.apache.kafka.common.serialization.ShortDeserializer
• org.apache.kafka.common.serialization.StringDeserializer

⾃定义反序列化

⾃定义反序列化类，需要实现 org.apache.kafka.common.serialization.Deserializer 接⼝。

package luu.demo.kafka.consumer;

import luu.demo.kafka.model.User;
import org.apache.kafka.common.serialization.Deserializer;

import java.nio.ByteBuffer;
import java.util.Map;

public class UserDeserializer implements Deserializer<User> {
    
    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {
    }
    @Override
    public User deserialize(String topic, byte[] data) {
        ByteBuffer allocate = ByteBuffer.allocate(data.length);
        allocate.put(data);
        allocate.flip();
        int userId = allocate.getInt();
        int length = allocate.getInt();
        System.out.println(length);
        String username = new String(data, 8, length);
        return new User(userId, username);
    }
    
    @Override
    public void close() {
    }
}
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

MyConsumer

package luu.demo.kafka.consumer;

import luu.demo.kafka.model.User;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.function.Consumer;

public class MyConsumer {
    public static void main(String[] args) {
        Map<String, Object> configs = new HashMap<>();
        configs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "node1:9092");
        configs.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
                StringDeserializer.class);
        configs.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
                UserDeserializer.class);
        configs.put(ConsumerConfig.GROUP_ID_CONFIG, "consumer1");
        configs.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        configs.put(ConsumerConfig.CLIENT_ID_CONFIG, "con1");
        KafkaConsumer<String, User> consumer = new KafkaConsumer<String, User>(configs);
        consumer.subscribe(Collections.singleton("tp_user_01"));
        ConsumerRecords<String, User> records = consumer.poll(Long.MAX_VALUE);
        records.forEach(new Consumer<ConsumerRecord<String, User>>() {
            @Override
            public void accept(ConsumerRecord<String, User> record) {
                System.out.println(record.value());
            }
        });
        // 关闭消费者
        consumer.close();
    }
}
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

位移提交

• Consumer需要向Kafka记录⾃⼰的位移数据，这个汇报过程称为提交位移(Committing Offsets)
• Consumer 需要为分配给它的每个分区提交各⾃的位移数据
• 位移提交的由Consumer端负责的，Kafka只负责保管__consumer_offsets
• 位移提交分为⾃动提交和⼿动提交
• 位移提交分为同步提交和异步提交

自动提交

Kafka Consumer 后台提交

• 开启⾃动提交：enable.auto.commit=true
• 配置⾃动提交间隔：Consumer端：auto.commit.interval.ms ，默认 5s

Map<String, Object> configs = new HashMap<>();
configs.put("bootstrap.servers", "node1:9092");
configs.put("group.id", "mygrp");
// 设置偏移量⾃动提交。⾃动提交是默认值。这⾥做示例。
configs.put("enable.auto.commit", "true");
// 偏移量⾃动提交的时间间隔
configs.put("auto.commit.interval.ms", "3000");
configs.put("key.deserializer", StringDeserializer.class);
configs.put("value.deserializer", StringDeserializer.class);
KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(configs);
consumer.subscribe(Collections.singleton("tp_demo_01"));
while (true) {
ConsumerRecords<String, String> records = consumer.poll(100);
for (ConsumerRecord<String, String> record : records) {
System.out.println(record.topic()
 - "\t" + record.partition()
 - "\t" + record.offset()
 - "\t" + record.key()
 - "\t" + record.value());
 }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

• ⾃动提交位移的顺序
  配置 enable.auto.commit = true
  Kafka会保证在开始调⽤poll⽅法时，提交上次poll返回的所有消息
  因此⾃动提交不会出现消息丢失，但会 重复消费
重复消费举例:
  Consumer 每 5s 提交 offset
  假设提交 offset 后的 3s 发⽣了 Rebalance
  Rebalance 之后的所有 Consumer 从上⼀次提交的 offset 处继续消费
  因此 Rebalance 发⽣前 3s 的消息会被重复消费

同步提交

• 使⽤ KafkaConsumer#commitSync()：会提交 KafkaConsumer#poll() 返回的最新 offset
• 该⽅法为同步操作，等待直到 offset 被成功提交才返回

while (true) {
    ConsumerRecords<String, String> records =
            consumer.poll(Duration.ofSeconds(1));
    process(records); // 处理消息
    try {
        consumer.commitSync();
    } catch (CommitFailedException e) {
        handle(e); // 处理提交失败异常
    }
}
1
2
3
4
5
6
7
8
9
10

• commitSync 在处理完所有消息之后
• ⼿动同步提交可以控制offset提交的时机和频率
• ⼿动同步提交会：
  调⽤ commitSync 时，Consumer 处于阻塞状态，直到 Broker 返回结果
  会影响 TPS
  可以选择拉⻓提交间隔，但有以下问题
  会导致 Consumer 的提交频率下降
  Consumer 重启后，会有更多的消息被消费

异步提交

• KafkaConsumer#commitAsync()

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(3_000);
    process(records); // 处理消息
    consumer.commitAsync((offsets, exception) -> {
        if (exception != null) {
            handle(exception);
        }
    });
}
1
2
3
4
5
6
7
8
9

• commitAsync出现问题不会⾃动重试
• 处理⽅式：

try {
    while (true) {
        ConsumerRecords<String, String> records =
                consumer.poll(Duration.ofSeconds(1));
        process(records); // 处理消息
        commitAysnc(); // 使⽤异步提交规避阻塞
    }
} catch (Exception e) {
    handle(e); // 处理异常
} finally {
    try {
        consumer.commitSync(); // 最后⼀次提交使⽤同步阻塞式提交
    } finally {
        consumer.close();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

消费者位移管理

Kafka中，消费者根据消息的位移顺序消费消息。
消费者的位移由消费者管理，可以存储于zookeeper中，也可以存储于Kafka主题__consumer_offsets中。
Kafka提供了消费者API，让消费者可以管理⾃⼰的位移。
API如下：KafkaConsumer

再均衡

重平衡可以说是kafka为⼈诟病最多的⼀个点了。
重平衡其实就是⼀个协议，它规定了如何让消费者组下的所有消费者来分配topic中的每⼀个分区。⽐如⼀个topic有100个分区，⼀个消费者组内有20个消费者，在协调者的控制下让组内每⼀个消费者分配到5个分区，这个分配的过程就是重平衡。

触发重平衡主要的条件：

• 有新的消费者加入消费组。
• 有消费者宕机下线。消费者并不一定需要真正下线，例如遇到长时间的 GC、网络延迟、消费超时导致消费者长时间未向GroupCoordinator发送心跳等情况时，GroupCoordinator会认为消费者已经下线。
• 有消费者主动退出消费组（发送 LeaveGroupRequest 请求）。比如客户端调用了unsubscrible（）方法取消对某些主题的订阅。
• 消费组所对应的GroupCoorinator节点发生了变更。
• 消费组内所订阅的任一主题或者主题的分区数量发生变化。
  为什么说重平衡为⼈诟病呢？因为重平衡过程中，消费者⽆法从kafka消费消息，这对kafka的TPS影响极⼤，⽽如果kafka集内节点较多，⽐如数百个，那重平衡可能会耗时极多。数分钟到数⼩时都有可能，⽽这段时间kafka基本处于不可⽤状态。所以在实际环境中，应该尽量避免重平衡发⽣。

避免重平衡

要说完全避免重平衡，是不可能，因为你⽆法完全保证消费者不会故障。⽽消费者故障其实也是最常⻅的引发重平衡的地⽅，所以我们需要保证尽⼒避免消费者故障。
⽽其他⼏种触发重平衡的⽅式，增加分区，或是增加订阅的主题，抑或是增加消费者，更多的是主动控制。如果消费者真正挂掉了，就没办法了，但实际中，会有⼀些情况，kafka错误地认为⼀个正常的消费者已经挂掉了，我们要的就是避免这样的情况出现。

⾸先要知道哪些情况会出现错误判断挂掉的情况。在分布式系统中，通常是通过⼼跳来维持分布式系统的，kafka也不例外。
在分布式系统中，由于⽹络问题你不清楚没接收到⼼跳，是因为对⽅真正挂了还是只是因为负载过重没来得及发⽣⼼跳或是⽹络堵塞。所以⼀般会约定⼀个时间，超时即判定对⽅挂了。⽽在kafka消费者场景中：

• session.timout.ms控制⼼跳超时时间,就是规定这个超时时间是多少。
• heartbeat.interval.ms控制⼼跳发送频率，这个参数控制发送⼼跳的频率，频率越⾼越不容易被误判，但也会消耗更多资源
• max.poll.interval.ms控制poll消息的间隔，消费者poll数据后，需要⼀些处理，再进⾏拉取。如果两次拉取时间间隔超过这个参数设置的值，那么消费者就会被踢出消费者组。也就是说，拉取，然后处理，这个处理的时间不能超过 max.poll.interval.ms 这个参数的值。这个参数的默认值是5分钟，⽽如果消费者接收到数据后会执⾏耗时的操作，则应该将其设置得⼤⼀些

这⾥给出⼀个相对较为合理的配置，如下：

• session.timout.ms：设置为6s
• heartbeat.interval.ms：设置2s
• max.poll.interval.ms：推荐为消费者处理消息最⻓耗时再加1分钟

消费者拦截器

消费者在拉取了分区消息之后，要⾸先经过反序列化器对key和value进⾏反序列化处理。
处理完之后，如果消费端设置了拦截器，则需要经过拦截器的处理之后，才能返回给消费者应⽤程序进⾏处理。

消费端定义消息拦截器，需要实现 org.apache.kafka.clients.consumer.ConsumerInterceptor 接
⼝。

1. ⼀个可插拔接⼝，允许拦截甚⾄更改消费者接收到的消息。⾸要的⽤例在于将第三⽅组件引⼊消费者应⽤程序，⽤于定制的监控、⽇志处理等。
2. 该接⼝的实现类通过configre⽅法获取消费者配置的属性，如果消费者配置中没有指定clientID，还可以获取
   KafkaConsumer⽣成的clientId。获取的这个配置是跟其他拦截器共享的，需要保证不会在各个拦截器之间产⽣冲突。
3. ConsumerInterceptor⽅法抛出的异常会被捕获、记录，但是不会向下传播。如果⽤户配置了错误的key或value类型参数，消费者不会抛出异常，⽽仅仅是记录下来。
4. ConsumerInterceptor回调发⽣在org.apache.kafka.clients.consumer.KafkaConsumer#poll(long)⽅法同⼀个线程。

接口中有如下方法：

package org.apache.kafka.clients.consumer;

import org.apache.kafka.common.Configurable;
import org.apache.kafka.common.TopicPartition;
import java.util.Map;

public interface ConsumerInterceptor<K, V> extends Configurable {
    /**
     * 该⽅法在poll⽅法返回之前调⽤。调⽤结束后poll⽅法就返回消息了。
     * 

     * 该⽅法可以修改消费者消息，返回新的消息。拦截器可以过滤收到的消息或⽣成新的消息。
     * 如果有多个拦截器，则该⽅法按照KafkaConsumer的configs中配置的顺序调⽤。
     *
     * @param records 由上个拦截器返回的由客户端消费的消息。
     */
    public ConsumerRecords<K, V> onConsume(ConsumerRecords<K, V> records);

    /**
     * 当消费者提交偏移量时，调⽤该⽅法。
     * 该⽅法抛出的任何异常调⽤者都会忽略。
     */
    public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets);

    public void close();
}
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

自定义拦截器

package com.lagou.kafka.demo.interceptor;
import org.apache.kafka.clients.consumer.ConsumerInterceptor;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import java.util.Map;

public class OneInterceptor implements ConsumerInterceptor<String, String> {
    @Override
    public ConsumerRecords<String, String> onConsume(ConsumerRecords<String, String>
                                                             records) {
        // poll⽅法返回结果之前最后要调⽤的⽅法
        System.out.println("One -- 开始");
        // 消息不做处理，直接返回
        return records;
    }

    @Override
    public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets) {
        // 消费者提交偏移量的时候，经过该⽅法
        System.out.println("One -- 结束");
    }

    @Override
    public void close() {
        // ⽤于关闭该拦截器⽤到的资源，如打开的⽂件，连接的数据库等
    }

    @Override
    public void configure(Map<String, ?> configs) {
        // ⽤于获取消费者的设置参数
        configs.forEach((k, v) -> {
            System.out.println(k + "\t" + v);
        });
    }
}
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

配置拦截器

// One -> Two -> Three，接收消息和发送偏移量确认都是这个顺序
props.setProperty(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG,
	"com.lagou.kafka.demo.interceptor.OneInterceptor" +
	",com.lagou.kafka.demo.interceptor.TwoInterceptor" +
	",com.lagou.kafka.demo.interceptor.ThreeInterceptor"
 );
1
2
3
4
5
6

消费者参数配置补充