Kafka-消费者概述
一、消费方式
1.pull（ 拉）模式
consumer采用从broker中主动拉取数据。
Kafka采用这种方式。
2.push（推）模式
Kafka没有采用这种方式，因为由broker决定消息发送速率， 很难适应所有消费者的消费速率。 例如推送的速度是50m/s，Consumer1、 Consumer2就来不及处理消息。
pull模式不足之处是， 如果Kafka没有数据， 消费者可能会陷入循环中， 一直返回空数据。
image.png
二、消费者组
Consumer Group 是 Kafka 提供的可扩展且具有容错性的消费者机制。
组内可以有多个消费者或消费者实例（Consumer Instance），它们共享一个公共的 ID，这个 ID 被称为 Group ID。
组内的所有消费者协调在一起来消费订阅主题（Subscribed Topics）的所有分区（Partition）。每个分区只能由同一个消费者组内的一个 Consumer 实例来消费。
Consumer Group三个特性：
- Consumer Group 下可以有一个或多个 Consumer 实例。这里的实例可以是一个单独的进程，也可以是同一进程下的线程。在实际场景中，使用进程更为常见一些。
- Group ID 是一个字符串，在一个 Kafka 集群中，它标识唯一的一个 Consumer Group。
- Consumer Group 下所有实例订阅的主题的单个分区，只能分配给组内的某个 Consumer 实例消费。这个分区当然也可以被其他的 Group 消费。
当 Consumer Group 订阅了多个主题后，组内的每个实例不要求一定要订阅主题的所有分区，它只会消费部分分区中的消息。
Consumer Group 之间彼此独立，互不影响，它们能够订阅相同的一组主题而互不干涉。再加上 Broker 端的消息留存机制，Kafka 的 Consumer Group 完美地规避了上面提到的伸缩性差的问题。
Kafka 仅仅使用 Consumer Group 这一种机制，却同时实现了传统消息引擎系统的两大模型：
- 如果所有实例都属于同一个 Group，那么它实现的就是消息队列模型；
- 如果所有实例分别属于不同的 Group，那么它实现的就是发布 / 订阅模型。
理想情况下，Consumer 实例的数量应该等于该 Group 订阅主题的分区总数。
消费者在消费的过程中需要记录自己消费了多少数据，即消费位置信息。在 Kafka 中，这个位置信息叫位移（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 （ brokerconfig） or max.message.bytes （topic config） 影响。 |
| max.poll.records | 一次 poll 拉取数据返回消息的最大条数， 默认是 500 条 |
极客时间《Kafka 核心技术与实战》学习笔记Day14 - http://gk.link/a/11UOV
