kafka 消费者

kafka 消费方式

1）pull（拉）模式：
consumer采用从broker中主动拉取数据。 Kafka采用这种方式。

2）push（推）模式：
Kafka没有采用这种方式，因为由broker 决定消息发送速率，很难适应所有消费者的消费速率。例如推送的速度是50m/s， Consumer1、Consumer2就来不及处理消息。

pull模式不足之处是，如果Kafka没有数据，消费者可能会陷入循环中，一直返回空数据。

Kafka 消费者工作流程

消费者总体工作流程

消费者组原理

Consumer Group（CG）：消费者组，由多个consumer组成。形成一个消费者组的条件，是所有消费者的groupid相同。

消费者组内每个消费者负责消费不同分区的数据，一个分区只能由一个组内消费者消费。
消费者组之间互不影响。所有的消费者都属于某个消费者组，即消费者组是逻辑上的一个订阅者。

消费者组初始化流程

coordinator：辅助实现消费者组的初始化和分区的分配。
coordinator节点选择 = groupid的hashcode值 % 50（ __consumer_offsets的分区数量）
例如： groupid的hashcode值 = 1，1% 50 = 1，那么__consumer_offsets 主题的1号分区，在哪个broker上，就选择这个节点的coordinator 作为这个消费者组的老大。消费者组下的所有的消费者提交offset的时候就往这个分区去提交offset。

消费者组详细消费流程

消费者重要参数

参数名称	描述
bootstrap.servers。	向 Kafka 集群建立初始连接用到的 host/port 列表
key.deserializer 和 value.deserializer	指定接收消息的 key 和 value 的反序列化类型。一定要写全类名。
group.id	标记消费者所属的消费者组。
enable.auto.commit	默认值为 true，消费者会自动周期性地向服务器提交偏移 量。
auto.commit.interval.ms	如果设置了 enable.auto.commit 的值为 true， 则该值定义了 消费者偏移量向 Kafka 提交的频率，默认 5s。
auto.offset.reset	当 Kafka 中没有初始偏移量或当前偏移量在服务器中不存在 （如，数据被删除了），该如何处理？ earliest：自动重置偏 移量到最早的偏移量。 latest：默认，自动重置偏移量为最 新的偏移量。 none：如果消费组原来的（previous）偏移量 不存在，则向消费者抛异常。 anything：向消费者抛异常。
offsets.topic.num.partitions	__consumer_offsets 的分区数，默认是 50 个分区。
heartbeat.interval.ms	Kafka 消费者和 coordinator 之间的心跳时间，默认 3s。 该条目的值必须小于 session.timeout.ms ，也不应该高于 session.timeout.ms 的 1/3。
session.timeout.ms Kafka	消费者和 coordinator 之间连接超时时间，默认 45s。 超过该值，该消费者被移除，消费者组执行再平衡。
max.poll.interval.ms	消费者处理消息的最大时长，默认是 5 分钟。超过该值，该 消费者被移除，消费者组执行再平衡。
fetch.min.bytes	默认 1 个字节。消费者获取服务器端一批消息最小的字节数。
fetch.max.wait.ms	默认 500ms。如果没有从服务器端获取到一批数据的最小字节数。该时间到，仍然会返回数据。
fetch.max.bytes	默认 Default: 52428800（50 m）。消费者获取服务器端一批 消息最大的字节数。如果服务器端一批次的数据大于该值 （50m）仍然可以拉取回来这批数据，因此，这不是一个绝 对最大值。一批次的大小受 message.max.bytes （broker config）or max.message.bytes （topic config）影响。
max.poll.records	一次 poll 拉取数据返回消息的最大条数，默认是 500 条。

消费者 API

独立消费者案例（订阅主题）

1）需求：创建一个独立消费者，消费 first 主题中数据。

注意：在消费者 API 代码中必须配置消费者组 id。命令行启动消费者不填写消费者组 id 会被自动填写随机的消费者组 id。

2）实现步骤
（1）创建包名：com.atguigu.kafka.consumer
（2）编写代码

#produce.php
<?php
$conf = new \RdKafka\Conf();
// 绑定消费者组
$conf->set('group.id', 'test');
// 绑定服务节点，多个用,分隔
$conf->set('metadata.broker.list', '192.168.172.131');
// 设置自动提交为false
$conf->set('enable.auto.commit', 'false');
// 设置当前消费者拉取数据时的偏移量， 可选参数：
// earliest: 如果消费者组是新创建的，从头开始消费，否则从消费者组当前消费位移开始。
// latest:如果消费者组是新创建的，从最新偏移量开始，否则从消费者组当前消费位移开始。
$conf->set('auto.offset.reset', 'earliest');

// 创建消费者实例
$consumer = new \RdKafka\KafkaConsumer($conf);
// 消费者订阅主题，数组形式
$consumer->subscribe(['test']);
while (true) {
// 消费数据，阻塞5秒(5秒内有数据就消费，没有数据等待5秒进入下一轮循环)
    $message = $consumer->consume(5000);
    switch ($message->err) {
        case RD_KAFKA_RESP_ERR_NO_ERROR:
            // 业务逻辑
            var_dump($message);
            $consumer->commit($message);// 提交位移
            break;
        case RD_KAFKA_RESP_ERR__PARTITION_EOF:
            echo "No more messages; will wait for more\n";
            break;
        case RD_KAFKA_RESP_ERR__TIMED_OUT:
            echo "Timed out\n";
            break;
        default:
            throw new \Exception($message->errstr(), $message->err);
            break;
    }
}

3）终端执行消费者

$ php produce.php

4）执行生产者生产数据，在第三步终端上即可看到消费情况

[atguigu@hadoop102 kafka]$ ./bin/kafka-console-producer.sh --bootstrap-server hadoop102:9092 --topic first
>hello
>hi
>today
>

独立消费者案例（订阅分区）

1）需求：创建一个独立消费者，消费 first 主题 0 号分区的数据。

2）实现：

#produce.php
<?php
$conf = new RdKafka\Conf();
$conf->set('group.id', 'myConsumerGroup');
$rk = new RdKafka\Consumer($conf);
$rk->addBrokers("192.168.172.131");
$topicConf = new RdKafka\TopicConf();
$topicConf->set('auto.commit.interval.ms', 100);//设置提交频次
$topicConf->set('offset.store.method', 'broker');//存储方式设置在broker 端
$topicConf->set('auto.offset.reset', 'earliest');//自动重置偏移量到最早的偏移量
$topic = $rk->newTopic("test", $topicConf);
$topic->consumeStart(0, RD_KAFKA_OFFSET_STORED);
while (true) {
    $message = $topic->consume(0, 5000);#指定主题分区
    switch ($message->err) {
        case RD_KAFKA_RESP_ERR_NO_ERROR:
            var_dump($message);
            #处理业务逻辑
            break;
        case RD_KAFKA_RESP_ERR__PARTITION_EOF:
            echo "No more messages; will wait for more\n";
            break;
        case RD_KAFKA_RESP_ERR__TIMED_OUT:
            echo "Timed out\n";
            break;
        default:
            throw new \Exception($message->errstr(), $message->err);
            break;
    }
}

3）测试
（1）在 IDEA 中执行消费者发送指定分区程序。
（2）在 IDEA 中执行生产者程序在控制台观察生成几个 0 和1号 分区的数据。
（3）在 IDEA 控制台，观察接收到的数据，只能消费到 0 号分区数据表示正确。

消费者组案例

1）需求：测试同一个主题的分区数据，只能由一个消费者组中的一个消费。

2）案例实操
（1）复制一份基础消费者的代码，在 IDEA 中同时启动，即可启动同一个消费者组中的两个消费者。
（2）启动代码中的生产者发送消息，在 IDEA 控制台即可看到两个消费者在消费不同 分区的数据（如果只发生到一个分区，可以在发送时增加延迟代码 sleep(2);）