分布式 - 消息队列Kafka：Kafka 消费者消息消费与参数配置_kafka消费消息

1. Kafka 消费者消费消息

public class CustomConsumer {
    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"10.65.132.2:9093");
        properties.put(ConsumerConfig.GROUP_ID_CONFIG,"test-group-hh");

        // 创建消费者
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);
        // 订阅主题 test
        consumer.subscribe(Collections.singletonList("test"));
        // 消费数据
        while (true){
            ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofSeconds(1));
            for (ConsumerRecord<String, String> record : consumerRecords) {
                System.out.printf("主题 = %s, 分区 = %d, 位移 = %d, " + "消息键 = %s, 消息值 = %s\n",
                        record.topic(), record.partition(), record.offset(), record.key(), record.value());
            }
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

01. 创建消费者

在读取消息之前，需要先创建一个KafkaConsumer对象。创建KafkaConsumer对象与创建KafkaProducer对象非常相似——把想要传给消费者的属性放在Properties对象里。

Properties properties = new Properties();
properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());
properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"10.65.132.2:9093");
properties.put(ConsumerConfig.GROUP_ID_CONFIG,"test-group-hh");

// 创建消费者
KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);
1
2
3
4
5
6
7
8

为简单起见，这里只提供4个必要的属性：bootstrap.servers、key.deserializer 和 value.deserializer。

① bootstrap.servers 指定了连接Kafka集群的字符串。

② key.deserializer 和 value.deserialize 是为了把字节数组转成Java对象。

③ group.id 指定了一个消费者属于哪一个消费者群组 ，默认值为“”。如果设置为空，则会报出异常：Exception in thread "main"org.apache.kafka.common.errors.InvalidGroupIdException：The configured groupId is invalid。一般而言，这个参数需要设置成具有一定的业务意义的名称。

02. 订阅主题

① 在创建好消费者之后，下一步就可以开始订阅主题了。subscribe()方法会接收一个主题列表作为参数。

// 订阅单个主题 test
consumer.subscribe(Collections.singletonList("test"));
// 订阅多个主题
consumer.subscribe(Arrays.asList("test","test1"));
1
2
3
4

② 也可以在调用subscribe()方法时传入一个正则表达式。正则表达式可以匹配多个主题，如果有人创建了新主题，并且主题的名字与正则表达式匹配，那么就会立即触发一次再均衡，然后消费者就可以读取新主题里的消息。如果应用程序需要读取多个主题，并且可以处理不同类型的数据，那么这种订阅方式就很有用。

consumer.subscribe(Pattern.compile("test.*"));
1

subscribe 的重载方法中有一个参数类型是ConsumerRebalance-Listener，这个是用来设置相应的再均衡监听器的。

③ 消费者不仅可以通过KafkaConsumer.subscribe（）方法订阅主题，还可以直接订阅某些主题的特定分区，在KafkaConsumer中还提供了一个assign（）方法来实现这些功能，这个方法只接受一个参数partitions，用来指定需要订阅的分区集合。

public class KafkaConsumer<K, V> implements Consumer<K, V> {
    @Override
    public void assign(Collection<TopicPartition> partitions) {
        // ...
    }

    public final class TopicPartition implements Serializable {
        private static final long serialVersionUID = -613627415771699627L;

        private int hash = 0;
        private final int partition;
        private final String topic;

        public TopicPartition(String topic, int partition) {
            this.partition = partition;
            this.topic = topic;
        }

        public int partition() {
            return partition;
        }

        public String topic() {
            return topic;
        }
        // ...
    }

    @Override
    public List<PartitionInfo> partitionsFor(String topic) {
        return partitionsFor(topic, Duration.ofMillis(defaultApiTimeoutMs));
    }
}
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

TopicPartition类只有2个属性：topic和partition，分别代表分区所属的主题和自身的分区编号，这个类可以和我们通常所说的主题—分区的概念映射起来。

// 订阅主题 test 和分区2
consumer.assign(Collections.singletonList(new TopicPartition("test",2)));
1
2

如果我们事先并不知道主题中有多少个分区怎么办？KafkaConsumer 中的partitionsFor（）方法可以用来查询指定主题的元数据信息。PartitionInfo类中的属性topic表示主题名称，partition代表分区编号，leader代表分区的leader副本所在的位置，replicas代表分区的AR集合，inSyncReplicas代表分区的ISR集合，offlineReplicas代表分区的OSR集合。

public class PartitionInfo {
    private final String topic;
    private final int partition;
    private final Node leader;
    private final Node[] replicas;
    private final Node[] inSyncReplicas;
    private final Node[] offlineReplicas;
}
1
2
3
4
5
6
7
8

通过 subscribe（）方法订阅主题具有消费者自动再均衡的功能，在多个消费者的情况下可以根据分区分配策略来自动分配各个消费者与分区的关系。当消费组内的消费者增加或减少时，分区分配关系会自动调整，以实现消费负载均衡及故障自动转移。而通过assign（）方法订阅分区时，是不具备消费者自动均衡的功能的，其实这一点从assign（）方法的参数中就可以看出端倪，两种类型的subscribe（）都有ConsumerRebalanceListener类型参数的方法，而assign（）方法却没有。

03. 轮询拉取数据

消费者API最核心的东西是通过一个简单的轮询向服务器请求数据。

// 消费数据
while (true){
    ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofSeconds(1));
    for (ConsumerRecord<String, String> record : consumerRecords) {
        System.out.printf("主题 = %s, 分区 = %d, 位移 = %d, " + "消息键 = %s, 消息值 = %s\n",
                          record.topic(), record.partition(), record.offset(), record.key(), record.value());
    }
}
1
2
3
4
5
6
7
8

这是一个无限循环。消费者实际上是一个长时间运行的应用程序，它通过持续轮询来向Kafka请求数据。消费者必须持续对Kafka进行轮询，否则会被认为已经“死亡”，它所消费的分区将被移交给群组里其他的消费者。传给poll()的参数是一个超时时间间隔，用于控制poll()的阻塞时间（当消费者缓冲区里没有可用数据时会发生阻塞）。如果这个参数被设置为0或者有可用的数据，那么poll()就会立即返回，否则它会等待指定的毫秒数。poll()方法会返回一个记录列表。列表中的每一条记录都包含了主题和分区的信息、记录在分区里的偏移量，以及记录的键–值对。我们一般会遍历这个列表，逐条处理记录。

轮询不只是获取数据那么简单。在第一次调用消费者的poll()方法时，它需要找到GroupCoordinator，加入群组，并接收分配给它的分区。如果触发了再均衡，则整个再均衡过程也会在轮询里进行，包括执行相关的回调。所以，消费者或回调里可能出现的错误最后都会转化成poll()方法抛出的异常。

需要注意的是，如果超过max.poll.interval.ms没有调用poll()，则消费者将被认为已经“死亡”，并被逐出消费者群组。因此，要避免在轮询循环中做任何可能导致不可预知的阻塞的操作。

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

public class ConsumerRecord<K, V> {
    private final String topic;
    private final int partition;
    private 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 final Optional<Integer> leaderEpoch;
}
1
2
3
4
5
6
7
8
9
10
11
12
13

其中，topic 和 partition 这两个字段分别代表消息所属主题的名称和所在分区的编号。offset 表示消息在所属分区的偏移量。timestamp 表示时间戳，与此对应的timestampType 表示时间戳的类型。timestampType 有两种类型：CreateTime 和LogAppendTime，分别代表消息创建的时间戳和消息追加到日志的时间戳。headers表示消息的头部内容。key 和 value 分别表示消息的键和消息的值，一般业务应用要读取的就是value。

2. Kafka 消费者参数配置

01. fetch.min.bytes

这个属性指定了消费者从服务器获取记录的最小字节数，默认是1字节。broker在收到消费者的获取数据请求时，如果可用数据量小于fetch.min.bytes指定的大小，那么它就会等到有足够可用数据时才将数据返回。这样可以降低消费者和broker的负载，因为它们在主题流量不是很大的时候（或者一天里的低流量时段）不需要来来回回地传输消息。如果消费者在没有太多可用数据时CPU使用率很高，或者在有很多消费者时为了降低broker的负载，那么可以把这个属性的值设置得比默认值大。但需要注意的是，在低吞吐量的情况下，加大这个值会增加延迟。

02. fetch.max.wait.ms

通过设置fetch.min.bytes，可以让Kafka等到有足够多的数据时才将它们返回给消费者，feth.max.wait.ms则用于指定broker等待的时间，默认是500毫秒。如果没有足够多的数据流入Kafka，那么消费者获取数据的请求就得不到满足，最多会导致500毫秒的延迟。如果要降低潜在的延迟，那么可以把这个属性的值设置得小一些。如果fetch.max.wait.ms被设置为100毫秒，fetch.min.bytes被设置为1 MB，那么Kafka在收到消费者的请求后，如果有1MB数据，就将其返回，如果没有，就在100毫秒后返回，就看哪个条件先得到满足。

03. fetch.max.bytes

这个属性指定了Kafka返回的数据的最大字节数（默认为50 MB）。消费者会将服务器返回的数据放在内存中，所以这个属性被用于限制消费者用来存放数据的内存大小。需要注意的是，记录是分批发送给客户端的，如果broker要发送的批次超过了这个属性指定的大小，那么这个限制将被忽略。这样可以保证消费者能够继续处理消息。值得注意的是，broker端也有一个与之对应的配置属性，Kafka管理员可以用它来限制最大获取数量。broker端的这个配置属性可能很有用，因为请求的数据量越大，需要从磁盘读取的数据量就越大，通过网络发送数据的时间就越长，这可能会导致资源争用并增加broker的负载。

04. max.poll.records

这个属性用于控制单次调用poll()方法返回的记录条数。可以用它来控制应用程序在进行每一次轮询循环时需要处理的记录条数（不是记录的大小）。

05. max.partition.fetch.bytes

这个属性指定了服务器从每个分区里返回给消费者的最大字节数（默认值是1MB）。当KafkaConsumer.poll()方法返回ConsumerRecords时，从每个分区里返回的记录最多不超过max.partition.fetch.bytes指定的字节。需要注意的是，使用这个属性来控制消费者的内存使用量会让事情变得复杂，因为你无法控制broker返回的响应里包含多少个分区的数据。因此，对于这种情况，建议用fetch.max.bytes替代，除非有特殊的需求，比如要求从每个分区读取差不多的数据量。

06. session.timeout.ms 和 heartbeat.interval.ms

session.timeout.ms指定了消费者可以在多长时间内不与服务器发生交互而仍然被认为还“活着”，默认是10秒。如果消费者没有在session.timeout.ms指定的时间内发送心跳给群组协调器，则会被认为已“死亡”，协调器就会触发再均衡，把分区分配给群组里的其他消费者。session.timeout.ms与heartbeat.interval.ms紧密相关。

heartbeat.interval.ms指定了消费者向协调器发送心跳的频率，session.timeout.ms指定了消费者可以多久不发送心跳。因此，我们一般会同时设置这两个属性，heartbeat.interval.ms必须比session.timeout.ms小，通常前者是后者的1/3。如果session.timeout.ms是3秒，那么heartbeat.interval.ms就应该是1秒。把session.timeout.ms设置得比默认值小，可以更快地检测到崩溃，并从崩溃中恢复，但也会导致不必要的再均衡。把session.timeout.ms设置得比默认值大，可以减少意外的再均衡，但需要更长的时间才能检测到崩溃。

07. max.poll.interval.ms

这个属性指定了消费者在被认为已经“死亡”之前可以在多长时间内不发起轮询。前面提到过，心跳和会话超时是Kafka检测已“死亡”的消费者并撤销其分区的主要机制。我们也提到了心跳是通过后台线程发送的，而后台线程有可能在消费者主线程发生死锁的情况下继续发送心跳，但这个消费者并没有在读取分区里的数据。要想知道消费者是否还在处理消息，最简单的方法是检查它是否还在请求数据。但是，请求之间的时间间隔是很难预测的，它不仅取决于可用的数据量、消费者处理数据的方式，有时还取决于其他服务的延迟。在需要耗费时间来处理每个记录的应用程序中，可以通过max.poll.records来限制返回的数据量，从而限制应用程序在再次调用poll()之前的等待时长。但是，即使设置了max.poll.records，调用poll()的时间间隔仍然很难预测。于是，设置max.poll.interval.ms就成了一种保险措施。它必须被设置得足够大，让正常的消费者尽量不触及这个阈值，但又要足够小，避免有问题的消费者给应用程序造成严重影响。这个属性的默认值为5分钟。

08. default.api.timeout.ms

如果在调用消费者API时没有显式地指定超时时间，那么消费者就会在调用其他API时使用这个属性指定的值。默认值是1分钟，因为它比请求超时时间的默认值大，所以可以将重试时间包含在内。poll()方法是一个例外，因为它需要显式地指定超时时间。

09. request.timeout.ms

这个属性指定了消费者在收到broker响应之前可以等待的最长时间。如果broker在指定时间内没有做出响应，那么客户端就会关闭连接并尝试重连。它的默认值是30秒。不建议把它设置得比默认值小。在放弃请求之前要给broker留有足够长的时间来处理其他请求，因为向已经过载的broker发送请求几乎没有什么好处，况且断开并重连只会造成更大的开销。

10. auto.offset.reset

这个属性指定了消费者在读取一个没有偏移量或偏移量无效（因消费者长时间不在线，偏移量对应的记录已经过期并被删除）的分区时该做何处理。它的默认值是latest，意思是说，如果没有有效的偏移量，那么消费者将从最新的记录（在消费者启动之后写入Kafka的记录）开始读取。另一个值是earliest，意思是说，如果没有有效的偏移量，那么消费者将从起始位置开始读取记录。如果将auto.offset.reset设置为none，并试图用一个无效的偏移量来读取记录，则消费者将抛出异常。

11. partition.assignment.strategy

我们知道，分区会被分配给群组里的消费者。PartitionAssignor根据给定的消费者和它们订阅的主题来决定哪些分区应该被分配给哪个消费者。Kafka提供了几种默认的分配策略。

① 区间(range)

这个策略会把每一个主题的若干个连续分区分配给消费者。假设消费者C1和消费者C2同时订阅了主题T1和主题T2，并且每个主题有3个分区。那么消费者C1有可能会被分配到这两个主题的分区0和分区1，消费者C2则会被分配到这两个主题的分区2。因为每个主题拥有奇数个分区，并且都遵循一样的分配策略，所以第一个消费者会分配到比第二个消费者更多的分区。只要使用了这个策略，并且分区数量无法被消费者数量整除，就会出现这种情况。

② 轮询 (roundRobin)

这个策略会把所有被订阅的主题的所有分区按顺序逐个分配给消费者。如果使用轮询策略为消费者C1和消费者C2分配分区，那么消费者C1将分配到主题T1的分区0和分区2以及主题T2的分区1，消费者C2将分配到主题T1的分区1以及主题T2的分区0和分区2。一般来说，如果所有消费者都订阅了相同的主题（这种情况很常见），那么轮询策略会给所有消费者都分配相同数量（或最多就差一个）的分区。

③ 黏性(sticky)

设计黏性分区分配器的目的有两个：一是尽可能均衡地分配分区，二是在进行再均衡时尽可能多地保留原先的分区所有权关系，减少将分区从一个消费者转移给另一个消费者所带来的开销。如果所有消费者都订阅了相同的主题，那么黏性分配器初始的分配比例将与轮询分配器一样均衡。后续的重新分配将同样保持均衡，但减少了需要移动的分区的数量。如果同一个群组里的消费者订阅了不同的主题，那么黏性分配器的分配比例将比轮询分配器更加均衡。

④ 协作黏性(cooperative sticky)

这个分配策略与黏性分配器一样，只是它支持协作（增量式）再均衡，在进行再均衡时消费者可以继续从没有被重新分配的分区读取消息。

可以通过partition.assignment.strategy来配置分区策略，默认值是org.apache.kafka.clients.consumer.RangeAssignor，它实现了区间策略。你也可以把它改成org.apache.kafka.clients.consumer.RoundRobinAssignor、org.apache.kafka.clients.consumer.StickyAssignor或org.apache.kafka.clients.consumer.CooperativeStickyAssignor。还可以使用自定义分配策略，如果是这样，则需要把partition.assignment.strategy设置成自定义类的名字。

12. client.id

这个属性可以是任意字符串，broker用它来标识从客户端发送过来的请求，比如获取请求。它通常被用在日志、指标和配额中。

13. group.instance.id

这个属性可以是任意具有唯一性的字符串，被用于消费者群组的固定名称。

14. receive.buffer.bytes和send.buffer.bytes

这两个属性分别指定了socket在读写数据时用到的TCP缓冲区大小。如果它们被设置为–1，就使用操作系统的默认值。如果生产者或消费者与broker位于不同的数据中心，则可以适当加大它们的值，因为跨数据中心网络的延迟一般都比较高，而带宽又比较低。

15. offsets.retention.minutes

这是broker端的一个配置属性，需要注意的是，它也会影响消费者的行为。只要消费者群组里有活跃的成员（也就是说，有成员通过发送心跳来保持其身份），群组提交的每一个分区的最后一个偏移量就会被Kafka保留下来，在进行重分配或重启之后就可以获取到这些偏移量。但是，如果一个消费者群组失去了所有成员，则Kafka只会按照这个属性指定的时间（默认为7天）保留偏移量。一旦偏移量被删除，即使消费者群组又“活”了过来，它也会像一个全新的群组一样，没有了过去的消费记忆。