凌云时刻 · 技术
导读：了解完Producer，接下来介绍Kafka中的Consumer的概念，以及在消费Message时有什么样的策略。
作者 | 计缘
来源 | 凌云时刻（微信号：linuxpk）
Consumer
Consumer负责从Topic中读取数据，我们已经知道了Topic是通过名称确定唯一的，所以指定Consumer从哪个Topic中读数据，同样使用Topic名称指定。Kafka中的Consumer有以下几点需要我们注意：
我们只需要指定需要从哪个Topic中读取数据即可。不需要关心Consumer是从哪个Broker中的哪个Partition中读数据，这些工作由Kafka帮我们处理好了。
当持有Topic的Broker挂掉，重新恢复后，Consumer可以自动重新从该Broker中读数据。
在一个Partition中，Consumer是按Offset的顺序读取数据的。
一个Consumer可以同时读取多个Broker中的不同Partition，但是Partition之间无法保证读取数据的顺序，因为是并行执行的。
Consumer Group
Consumer有组的概念，对于Consumer Group有以下几点需要我们注意：
不同的Consumer Group之间可以读取相同的Partition中的数据。
Consumer Group里的Consumer之间不能读取相同的Partition中的数据，他们读取数据的Partition是专享的。
所以基于上面的知识点，如果Consumer数量多于Partition数量，那么就会有Consumer处于空闲的状态。
上图的示例中，如果Consumer Group 2中再增加一个Consumer 4，那么Consumer 4就会处于空闲状态，因为没有多余的Partition分给它了。
Consumer Offset
一个优秀的MQ系统，必定会有一个能力，那就是断点续传的能力。既当Consumer挂掉再恢复后，需要从挂掉的前一时刻读数据的点开始接着往后读。那么如何做到这一点呢，那就是通过Consumer Offset来实现的。
每当一个活跃的Consumer正在从Partition中读取数据时，Kafka都会根据给定的策略记住该Consumer读取数据的Offset。这个策略就是Consumer提交Offset的策略。目前有三个策略：
At most once:
这种策略下，只要Consumer读到了Message，就立即提交Offset，不考虑Message有没有被正确处理。如果Message刚读过来，还没有处理的时候，Consumer挂掉了，重新恢复后对上一次读取的Message不会重新读取，所以这种模式比较容易丢失数据。整个过程如下图所示：
At least once:
这种策略下，Consumer需要读到Message，并且正确处理了Message后，才会提交Offset。如果Consumer挂掉，再恢复后，可以重新读取上一次的Message继续处理。这里就需要我们处理Message的逻辑必须是幂等的，否则会造成Message重复执行导致错误的业务结果。整个过程如下图所示：
Exactly once:
这个策略想做到的是不丢数据，又可以不用幂等的处理逻辑。这里通常需要Kafka和外部系统配合使用。后面再做具体介绍。
Consumer Poll Options
在Consumer订阅Topic拉取Message的行为中，会涉及到四个参数：
fetch.min.bytes：该参数表示每次拉取Message的最小量，默认是1Bytes。fetch.max.bytes：该参数表示每次拉取Message的最大量，默认是50MB。max.poll.records：该参数表示每次拉取Message的条数，默认是500条。max.partitions.fetch.bytes：该参数表示每个Partition在一次拉取中，可以被拉取到Message的最大量，默认是1MB。
这些参数可以让Consumer控制拉取Message的速率，以及可以监控Consumer每次拉取Message的具体信息。
Consumer Offset Reset Behevior
在实际应用中，Consumer是很有可能在运行过程中挂掉的，那么当Consumer重新恢复后，拉取什么范围的Message，是有策略可以设置的，可以通过设置auto.offset.reset属性，常用的值有两个：
earliest：从Message文件的最开始进行拉取，既将Topic中的所有数据重新拉取过来。latest：从Message文件的最后开始拉取，既不考虑Topic之前的所有数据，只拉取最新的数据。
在后面讲到CLI的时候，这部分再做详细阐述。
Consumer Internal Thread
为保证Consumer的稳定性和高可用性。Kafka有心跳机制，所以Consumer不光和Broker交互，也要和心跳监控节点交互：
这里引出了两个参数：
seesion.timeout.ms：该参数的作用是Broker认为Consumer挂掉的持续时间。默认为10秒。也就是说Broker在10秒内没有收到Consumer的心跳信号，那么认为该Consumer已经挂掉了。heartbeat.interval.ms：该参数决定了Consumer发送心跳信号的间隔时间。默认为3秒。
总结一下前文介绍的Kafka核心概念。先上一张图总体概括：
Producer
在Producer层面，我们了解了以下知识点：
Producer发送Message到Broker默认采用轮询方式，除非显示的将Message带着Key。
如果希望Message根据某个字段发送至相同的Partition中，可以将Message带着Key发送。
Producer有acks机制，关系到Message的完整性，以及整体MQ系统的整体性能（Message吞吐量）。
Producer发送Message有重试机制。
在实际使用时，我们通常需要考虑幂等Producer，以确保不会有业务上的错误。
Message压缩和批量发送有助于提高Message传输性能。
Broker
在Broker层面，我们了解了以下知识点：
Partition是以文件夹的形式存储在Broker中的。
Partition有Replication的概念，可以确保Message的完整性。
Partition有Leader和ISR的概念。
Partition中Message存储的方式。
Partition中清理Message的策略。
Consumer
在Consumer层面，我们了解了以下知识点：
Consumer有组的概念，Consumer Group和Consumer获取Topic中数据的方式。
Consumer提交Offset的策略，关系到Consumer断点续传的方式。
Consumer如何控制获取Topic中Message的速率。
Consumer如何重制Offset。
最后我们再明确一下有哪些是Kafka提供的保障，或者说是我们不能，也不应该违背的原则：
Message写入Topic-Partition的顺序严格按照Producer发送Message的顺序。
Consumer从Topic-Partition读Message的顺序严格按照Partition中Message的Offset顺序。
如果Partition的Replication Factor是N，那么可以允许有N-1个Broker挂掉，而且Kafka可以正常运转。
只要Topic的Partition的数量恒定，那么带有指定Key的Message会始终写入该Key对应的Partition。
如果你想给Kafka集群中的某个Topic发送数据，你只需要连接Kafka集群中的一个Broker以及给定Topic名称既可。不用考虑Partition、Replication等等的问题。
总结
前面五个章节阐述了什么是MQ系统，然后基于这个大的框架介绍了Kafka系统中的核心模块，以及这些核心模块中的核心知识点。之后的章节主要就是实践部分，包括如何使用Kafka CLI、搭建Zookeeper、单机Kakfa、集群Kafka等。希望能给小伙伴们带来帮助。
END
往期精彩文章回顾
Kafka从上手到实践 - 庖丁解牛：Producer
Kafka从上手到实践 - 庖丁解牛：Partition
Kafka从上手到实践 - 庖丁解牛：Topic & Broker | 凌云时刻
Kafka从上手到实践 - 初步认知：MQ系统
进阶之路：深入解读 Java 堆外内存
干货：一文看懂Apache Ranger
吴翰清：有变革的需求，才有技术的诞生
云原生时代，消息中间件的演进路线
长按扫描二维码关注凌云时刻
每日收获前沿技术与科技洞见
