---

前言

在这里将会描述Kafka的消费者是通过什么样的方式进行消费，采用的分区策略，消费者组和消费者间的关系。以及在最后会描述通过客户端代码方式如何实现消费者。

---

提示：以下是本篇文章正文内容，下面案例可供参考

一、消费者详解

（一）消费方式

consumer采用拉取（pull）的模式从broker中读取数据，当自身有空余的消费能力的时候就会向消息队列拉取消息，这样的好处就是消费者不会因为消息数量庞大而导致接收的消息数量超过自身消费能力。

在历史上，Kafka曾考虑过采用推送（push）的模式，但是如果采用推送的模式，首先，由于消息发送速率是由broker决定的，难以适应消费速率不同的消费者；其次，很容易造成consumer来不及处理消息，典型的表现就是拒绝服务以及网络拥塞。

pull模式的不足：如果Kafka没有数据，消费者肯会陷入循环中，一直返回空数据。针对这一点，Kafka的消费者在消费数据时会传入一个时长参数timeout，如果当前没有数据可供消费，consumer会等待一段时间后在返回，这段时长即为timeout。

（二）分区分配策略

根据Kafka的架构模式我们可以知道，一个consumer group中有多个consumer，一个topic有多个partition，那么会涉及到哪个partition由哪个consumer来消费的问题。

Kafka主要有两种分区分配策略。

1、RoundRobin

将所有可用partitions和consumers展开(字典排序)，以轮询的方式将partitions依次分配给consumers。如果consuemrs订阅Topics都是相同的，那么partitions将会被均匀分配给每个consumer。最理想的状态是partitions数是consumers数的整数倍，这样每个consumer都有相同数量的partitions数。

举个栗子：

如果在Kafka中有两个topic，每个topic中含有三个partition；三个消费者

那么消费轮询的情况应该是，消费者轮询消费partition

由图可以看出，consumers0先消费了Topic0->Partition0，当consumers1再去消费的时候，就消费到的是Topic0->Partition1。

当consumers去消费Topic1的时候，再次进行轮询，这样就使得消费者均衡消费。

但是RoundRobin的分区分配策略不是完全均衡的分配策略。

当Consumers订阅Topics不相同，仍然按照轮询方式进行分配，将导致consumers之间分区分配不均衡。

还是刚刚的案例，三个Consumers和两个Topic，其中Consumers0和Consumers1不订阅Topic1，那么会造成：

此时Consumers2就进行了大量消费，而Consumers0和Consumers1就没什么事情干，导致消费不均衡。

2、Range

与RoundRobin不同，Range作用域为每个Topic。对于每一个Topic，将该Topic的所有可用partitions和订阅该Topic的所有consumers展开(字典排序)，然后将partitions数量除以consumers数量，算数除的结果分别分配给订阅该Topic的consumers，算数除的余数分配给前一个或者前几个consumers。

也就是说Range操作和RoundRobin操作都是轮询的方式分配，RoundRobin分配方式是将所有Topic看作整体参与轮询，而Range分配方式是每个Topic看作一个整体进行分配。每个主题都从Consumer0开始分配。

该方式会导致，如果该Topic的partitions数量与订阅该Topic的consumers数量不是整数倍关系，将造成前一个或者前几个consumer分配到较多的partitions，达不到consumer之间分区分配均衡的效果(不管是面向所有Topics还是单个Topic)。

举个栗子：
两个Topic，每个有三个partition，有两个consumers。
此时分配方案是尽可能的平均分配给订阅了该主题的消费者，但是如果出现了余数，那么将剩余的partition优先分配给在前面的客户。

---

二、客户端代码方式实现消费者

（一）实现过程

① 配置消费者参数
② 创建消费者实例
③ 订阅主题
④ 拉取消息进行消费
⑤ 提交位移 offset
⑥ 关闭消费者实例

（二）代码实现

import com.lmz.util.KafkaPara;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.io.*;
import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * @author zeyue
 */
public class KafkaConsumerTest {

    private static KafkaConsumer<String,String> consumer;

    /**
     * 创建历史记录保存的文件
     */
    private static File f = new File("calculator.txt");

    /**
     * 订阅主题名称
     */
    public static final String topic = "test";
    public static final AtomicBoolean isRunning = new AtomicBoolean(true);

    public static Properties initConfig() {
        Properties props = new Properties();
        props.put("key.deserializer",
                "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer",
                "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("bootstrap.servers", KafkaPara.brokerList);
        //消费者所属的消费者组
        props.put("group.id", "test-consumer-group");
        props.put("client.id", "consumer.client.id.demo");

        return props;
    }

    public static void main(String[] args) {
        consumerToPoll(initConfig());

    }

    /**
     * 消费者拉取消息
     * > ① 创建消费者实例
     * > ② 订阅主题
     * > ③ 拉取消息进行消费
     * > ④ 提交位移 offset
     * > ⑤ 关闭消费者实例
     * @param props 消费者参数
     */
    public static void consumerToPoll(Properties props) {
        consumer = new KafkaConsumer<>(props);
        //订阅主题
        consumer.subscribe(Arrays.asList(topic));
        //取消订阅
        consumer.unsubscribe();
        //再订阅回来
        consumer.subscribe(Arrays.asList(topic));
        try {
            while (isRunning.get()) {
                ConsumerRecords<String, String> records =
                        consumer.poll(Duration.ofMillis(100));
                if (records.isEmpty()){
                    System.out.println("it is null");
                }
                for (ConsumerRecord<String, String> record : records) {
                    System.out.println("topic = " + record.topic()
                            + ", partition = " + record.partition()
                            + ", offset = " + record.offset());
                    System.out.println("key = " + record.key()
                            + ", value = " + record.value());
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            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
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87

（三）代码解析

1、订阅主题

消费者可以使用集合的方式同时订阅多个主题；但是如果是多次调用订阅方法，则会以最后一次订阅的为准；

消费者还可以使用正则表达式的方式订阅主题；

在订阅方法中可以传递一个再均衡监听器；再均衡监听器用来设定发生再均衡动作前后的一些准备或收尾动作；

/**
 * KafkaConsumer订阅主题的方法
 */
public void subscribe(Collection<String> topics);
public void subscribe(Collection<String> topics, ConsumerRebalanceListener listener);

public void subscribe(Pattern pattern);
public void subscribe(Pattern pattern, ConsumerRebalanceListener listener);

1
2
3
4
5
6
7
8
9

2、订阅分区

Kafka可以同时订阅多个分区，但订阅分区就不会由Topic自动分配分区了。

订阅分区时，没有传递再均衡监听器（订阅主题时才会有再均衡）；

/**
 * KafkaConsumer订阅分区的方法
 */
public void assign(Collection<TopicPartition> partitions);

1
2
3
4
5

3、取消订阅

既可以取消通过 subscribe 订阅的主题，也可以取消通过 assign 方法指定订阅的分区。

/**
 * KafkaConsumer 取消订阅的方法
 */
 public void unsubscribe();
1
2
3
4

---

三、拦截器

生产者在发送消息时，可以使用生产者拦截器在消息发送前和发送回调逻辑前做一些定制化的需求；

消息者也可以在消息消息时，使用消费者拦截器在消费到消息时或提交消费位移之后做一些定制化需求；

生产者拦截器的接口是ProducerInterceptor，消费者拦截器的接口是ConsumerInterceptor

onConsume()方法：KafkaConsumer会在poll()方法返回之前调用onConsume()方法来对消息进行定制化操作，如果onConsume()方法中抛出异常，那么会被捕获并记录到日志中，但是异常不会再向上传递；

onCommit()方法：KafkaConsumer会在提交完消费位移之后调用拦截器的onCommit()方法；

要使自定义的消费者拦截器生效，需要在参数interceptor.classes参数中配置，该参数的默认值为""，即默认不使用消费组拦截器；

---

总结

Kafka的消费者在消费消息的时候采用的拉的方式，在拉取之前要确定拉取哪些主题的消息，也就是订阅哪些主题。

在订阅主题后要注意采用那种模式进行拉取，是通过自定义的指定位移的方式来进行拉取消息，还是默认自动的从集群中的默认位置开始拉取消息。