kafka 框架讲解+概念讲解+问题思考

た入场券 2021-12-18 07:37 311阅读 0赞

### 前言 ###

Kafka Kafka是最初由Linkedin公司开发，是一个分布式、支持分区的（partition）、多副本的（replica），基于zookeeper协调的发布/订阅模式的分布式消息系统，它的最大的特性就是可以实时的处理大量数据以满足各种需求场景：比如基于hadoop的批处理系统、低延迟的实时系统、storm/Spark流式处理引擎，web/nginx日志、访问日志，消息服务等等，用scala语言编写，Linkedin于2010年贡献给了Apache基金会并成为顶级开源 项目。

### kafka设计目标 ###

kafka开发的主要初衷目标是构建一个用来处理海量日志，用户行为和网站运营统计等的数据处理框架。在结合了数据挖掘，行为分析，运营监控等需求的情况下，需要能够满足各种实时在线和批量离线处理应用场合对低延迟和批量吞吐性能的要求。从需求的根本上来说，高吞吐率是第一要求，其次是实时性和持久性。

### 整体架构图 ###

![aHR0cHM6Ly9yYXcuZ2l0aHVidXNlcmNvbnRlbnQuY29tL0tsYXVzemhhby9waWN0dXJlL21hc3Rlci9waWN0dXJlL2thZmthL2thZmthJUU2JTlFJUI2JUU2JTlFJTg0JUU1JTlCJUJFLnBuZw][]

*  zookeeper作为注册中心
 *  有消息备份机制，并且有主从节点
 *  支持高并发生产和订阅，最小并发单元是分区

### kafka 概念 ###

*  #### Broke（经纪人） ####

可以简单理解为一个 Kafka 节点, 多个 Broker 节点构成整个 Kafka 集群，一个Broker可能有包含多个物理机器。一个Borker就是Kafka集群中的一个实例，或者说是一个服务单元。连接到同一个zookeeper的多个broker实例组成kafka的集群。在若干个broker中会有一个broker是leader，其余的broker为follower。leader在集群启动时候选举出来，负责和外部的通讯。当leader死掉的时候，follower们会再次通过选举，选择出新的leader，确保集群的正常工作。问题来了，leader节点做什么用的？当生产者把数据推给Kafka之后，是把消息给leader，它来接手数据，消费者来消费消息对时候也是找leader，leader是负责读写的。zookeeper中的leader是负责写，其他是负责读的，这叫读写分离，Kafka这么做是为了一致性。

*  #### Topic 主题 ####

特定类别的消息流称为主题，由用户定义并配置在Kafka服务器，用于建立生产者和消息者之间的订阅关系：生产者发送消息到指定的Topic下，消息者从这个Topic下消费消息。比如我们把发送验证码的消息流称为验证码Topic，把通知用户付款的消息通知称为PayNotify Topic，可以理解Topic为一类消息体，通过Topic来区别不同的消息体，那这里有一个疑问：一个Topic里消息体格式是一样的吗？应该说一个Topic里的消息对象都是一样的，但可能不同的group，里面消息对象字段不太一致，这个后续会讲解。 数据存储在主题中。主题被拆分成分区。 对于每个主题，Kafka保存其中一个分区的数据。 每个这样的分区包含不可变有序序列的消息。 分区被实现为具有相等大小的一组分段文件。

*  #### Partition 分区 ####

kafka采用分区（Partition）的方式，使得消费者能够做到并行消费，从而大大提高了自己的吞吐能力。同时为了实现高可用，每个分区又有若干份副本（Replica），这样在某个broker挂掉的情况下，数据不会丢失。Partition上会有Leader概念，因为partition是在broke上的，所以上面也说了broker的leader。  
![aHR0cHM6Ly9yYXcuZ2l0aHVidXNlcmNvbnRlbnQuY29tL0tsYXVzemhhby9waWN0dXJlL21hc3Rlci9waWN0dXJlL2thZmthL3RvcGljLXBhcnR0aXRpb24ucG5n][]

Topic可能有许多分区，因此它可以处理任意数量的数据，它是 Topic 在物理上的分组, 多个 Partition 会被分散地存储在不同的 Kafka 节点上; 单个 Partition 的消息是保证有序的, 但整个 Topic 的消息就不一定是有序的。主题被拆分成分区，分区被实现为具有相等大小的一组分段文件。那么每个分区数据都是一样的？ 答案是不一样的，这里分区的概念不同于主节点、子节点的概念，一个分区就是一个存储集合，partition中的每条消息都会被分配一个有序的 id（offset）。  
所有每个分区还有一个Partition offset（分区偏移）的概念，消息类似存放到一个数组里，通过下标来读取数据，Partition offset类似数组下表，记录消费者上一个时刻消费的消息下标，通过下标继续往下消费消息。

Replicas of partition（分区备份）：副本只是一个分区的备份， 表现为每个 Partition 都会有 N 个完全相同的冗余备份, 这些备份会被尽量分散存储在不同的机器上;。 副本从不读取或写入数据。 它们用于防止数据丢失。消息进来的时候会先存入leader replica，然后从leader replica复制到follower replica。只有复制全部完成时，consumer才可以消费此条消息。这是为了确保意外发生时，数据可以恢复。consumer的消费也是从leader replica读取的。

*  #### Group：消费者分组 ####

Kafka和其它消息系统有一个不一样的设计：在consumer之上加了一层group，用于归组同类消费者。在Kafka中，多个消费者可以共同消息一个Topic下的消息，每个消费者消费其中的部分消息，这些消费者就组成了一个分组，拥有同一个分组group名称，通常也被称为消费者集群。同一个group的consumer可以并行消费同一个topic的消息，比如：我有3个消费者在一个group，目前Topic里有30条消息，每个消费者消费10条，而不是每个消费者消费30条，所以同group的consumer不会重复消费。这就好比多个consumer组成了一个团队，一起干活，当然处理消息的速度就上来了。group中的consumer是如何配合协调的，其实和topic的分区相关联，后面我们会详细论述。

如果同一个topic需要被多次消费，可以通过设立多个consumer group来实现。每个group分别消费，互不影响。

*  #### 分区+ group ####

![aHR0cHM6Ly9yYXcuZ2l0aHVidXNlcmNvbnRlbnQuY29tL0tsYXVzemhhby9waWN0dXJlL21hc3Rlci9waWN0dXJlL2thZmthL3RvcGljLXBhcnR0aXRpb24tMi5wbmc][]

从上图中可以看到如下几个知识点：

*  1、一个partition只能被同Group的一个consumer消费，仅此一个。
 *  2、同一个组里的一个consumer可以消费多个partition（图中第一个consumer消费Partition 0和3），消费者与分区的关系就是1-N关系，一个consumer对应多个分区。有点像一夫多妻制，一个男人可以娶多个女人，女人只能跟着一个男人，比如比如，女孩子们也可以理解为一妻多夫，一个女人可以嫁个多个男人，男人只能跟着一个女人，这个女人就是你。
 *  3、消费效率最高的情况是partition和consumer数量相同。这样确保每个consumer专职负责一个partition。
 *  4、consumer数量不能大于partition数量。由于第一点的限制，当consumer多于partition时，就会有consumer闲置，资源浪费，应该可以配置。
 *  5、consumer group可以认为是一个订阅者的集群，其中的每个consumer负责自己所消费的分区。

这样设计就是为了提供并行消费能力，尽可能的快速消费消息，有点类似java 8 的hashmap的设计思想，还做了一点改进。

参考博客中[Apache-Kafka核心概念][Apache-Kafka]讲解了一个吃苹果的例子，一个人吃9个苹果，太慢了，找2个人一共3个人一起吃，并且把苹果分成3盘，每个人一盘，这样吃苹果就快了很多，比1个人吃一盘速度快3倍。

*  #### 生产者 - 消费者模型 ####

![aHR0cHM6Ly9yYXcuZ2l0aHVidXNlcmNvbnRlbnQuY29tL0tsYXVzemhhby9waWN0dXJlL21hc3Rlci9waWN0dXJlL2thZmthL3Byb2R1Y2VyLWNvbnN1bWVyLnBuZw][]

上图是一个生产者-消费者模型，我们分别来看：

*  生产者  
    采用round-robin算法，轮询往每个partition写入，这个也不一定非要轮询，看具体的分配算法了，毕竟分区有这么多个，解决方式也非常多。
 *  消费者  
    在消费者端，每个consumer都维护一个offset值，指向的是它所消费到的消息坐标。  
    我们先看group A的三个consumer，他们分别独立消费不同的三个partition。每个consumer维护了自己的offset。  
    我们再看group B，可以看到两个group是并行消费整个topic，同一条消息会被不同group消费到。

此处有如下知识点：  
1、每个partition都是有序的不可变的，应该是消息有序不可变，一个消息不可以从之前的位置1转移到位置2。  
2、Kafka可以保证partition的消费顺序，但不能保证topic消费顺序，不同Topic Group 消费顺序不保证。有的消费者消费的特别快，可能会赶超其他消费者Group。  
3、无论消费与否，保留周期默认两天（可配置）。如果消费慢了，会有什么情况？  
4、每个consumer维护的唯一元数据是offset，代表消费的位置，一般线性向后移动。  
5、consumer也可以重置offset到之前的位置，可以以任何顺序消费，不一定线性后移。

我还是想知道这几个问题：

*  如果消费慢了，会有什么情况？  
    会有积压的情况出现，这个时候，如果只是暂时的消息特别多，其实情况还好，如果是因为业务压力上来了，长期如此，建议还是增加分区，增加消费者数量，加速消费。有一些复杂的场景可以参考参考博客[消息队列常见问题和解决方案][Link 1]。
 *  如果生产慢了  
    没啥问题，就是浪费一些资源。

### kafka一些设计思路 ###

1、消息持久化  
是一种典型的磁盘式堆积，所有的消息都存储在磁盘中，kafka提供了两种策略删除数据：

*  基于时间，让kafka删除2天或一周的数据；
 *  基于partition文件大小：让kafka在partition文件超过1GB时删除数据

消息队列是以log文件的形式存储，消息生产者只能将消息添加到既有的文件尾部，没有任何ID信息用于消息的定位，完全依靠文件内的位移，因此消息的使用者只能依靠文件位移顺序读取消息，这样也就不需要维护复杂的支持随即读取的索引结构。

2、激进的内存管理模式  
kafka不在JVM进程内部维护消息Cache，消息直接从文件中读写，完全依赖操作系统在文件系统层面的cache，避免在JVM中管理Cache带来的额外数据结构开销和GC带来的性能代价。基于批量处理和顺序读写的应用模式，最大化利用文件系统的Cache机制和规避文件读写相对内存读写的性能代价。

3、采取各种方式最大化数据传输效率  
比如生产者和消费者可以批量读写消息减少RPC开销，使用Zero Copy方式在内核层直接将文件内容传送给网络Socket，避免应用层数据拷贝，使用合理的压缩格式等

4、可以指定分区或者利用指定key来做hash轮询分区  
这样将同类的数据放在一个分区里，可以有利用下游服务并发消费的某些需求。

5、发送消息到kafka，kafka要求全部副本都同步了才发送ack  
目前这一块主要有两个方式：

*  全部副本同步，所以如果有n+1台服务器，n台挂了，只要还有1台服务器可以，就能提供服务，但这样延迟就比较高
 *  一半副本，那这样你有n+1台，你挂了一半服务器，那么你就无法提供服务，如果你要容忍n台服务器挂了依旧可以提供服务，那么你就必须有2n+1台机器，优点：低延迟。

kafka选择了第一种方案，但实现的时候做了优化，加入了ISR（in-sync-replica set）同步队列，这是一个follower的集合，如果follower长时间没向leader的返回响应，leader就将follower踢出去，这样来保证全部副本同步内容。同时当leader挂掉，也会在ISR中选择follower来选举leader。原本是还有一个消息数的统计指标来作为是否移除的标示，但follower在这个指标上变化较大，会造成ISR频繁的变更，zk也频繁变更，于是0.9版本上移除掉了。完整的解释是：

> 假设设置replica.lag.max.messages=4，那么如果producer一次传送至broker的消息数量都小于4条时，因为在leader接受到producer发送的消息之后而follower副本开始拉取这些消息之前，follower落后leader的消息数不会超过4条消息，故此没有follower移出ISR，所以这时候replica.lag.max.message的设置似乎是合理的。但是producer发起瞬时高峰流量，producer一次发送的消息超过4条时，也就是超过replica.lag.max.messages，此时follower都会被认为是与leader副本不同步了，从而被踢出了ISR。但实际上这些follower都是存活状态的且没有性能问题。那么在之后追上leader,并被重新加入了ISR。于是就会出现它们不断地剔出ISR然后重新回归ISR，这无疑增加了无谓的性能损耗。

6、为什么 一个分区一个消费者？  
目前看到一些资料主要是因为保证分区内消息有序消费，如果分区被多个消费者消费，那么存在并发问题，而且消费者是根据index来获取消息，多个消费者就需要维护一个index中心，来协助每一个消费者下一次消费的idnex值。

7、丢消息的场景

*  生产者丢消息  
    通过带回调函数的发送消费的接口来提前发现和重发消息。
 *  消费者丢消息  
    总是先消费消息，再更新位移；这种可能带来消息重复消费的问题，但是不会出现消息丢失问题；  
    1、不丢消息：producer有个ack参数，有三个值，分别代表：不在乎是否写入成功、写入leader成功、写入leader和所有reclpica成功；要求非常可靠的话可以牺牲性能设置成最后一种。  
    2、不重复发送：正常发都不会重复，只可能丢，看你这边怎么容错重发了，参考上一条。  
    3、消息只读一次：同样，正常读不会重复，如果在上一次读的过程中发生了异常，消息可能被消费，但是offset没有及时commit；这本身是两步，存在中间crash的风险，大笨鸟所说的方法“处理消息、清空持久区”一样是两步，中间仍存在crash的风险，真要保证的话只能依靠其他逻辑判断当前消息是否被消费过。

### kafka 优点(相对于其他消息队列) ###

[rabbitMQ、activeMQ、zeroMQ、Kafka、Redis 比较][rabbitMQ_activeMQ_zeroMQ_Kafka_Redis]这篇文章里首先讲解了一些MQ功能对比，然后总结后认为在rabbitMQ、activeMQ、zeroMQ、Redis 中 RabbitMQ 综合实力最好，然后对比RabbitMQ 、Kafka，依旧是RabbitMQ最完善，里面提到的要点：

*  1、 RabbitMq比kafka成熟，在可用性上，稳定性上，可靠性上，RabbitMq超过kafka
 *  2、 Kafka设计的初衷就是处理日志的，可以看做是一个日志系统，针对性很强，所以它并没有具备一个成熟MQ应该具备的特性
 *  3、 Kafka的性能（吞吐量、tps）比RabbitMq要强，这篇文章的作者认为，两者在这方面没有可比性

RabbitMq性能其实是很强劲的，同时具备了一个成熟的MQ应该具有的特性，无需重新发明轮子。

除了kafka这种发布/订阅模式，还有1对1模式，消费者主动去拉去消息，然后消费后就去掉消息，一个消息只能被一个消费者消费，这种明显就不利于有多个消费者分组的情况，

### 为什么要用消息队列 ###

*  1、解耦  
    解耦是消息队列要解决的最本质问题。所谓解耦，简单点讲就是一个事务，只关心核心的流程。而需要依赖其他系统但不那么重要的事情，通过消息来通知其他系统即可，无需等待结果。换句话说，基于消息的模型，关心的是“通知”，而非“处理”。举一个例子，订单支付成功之后可能需要给用户发送短信积分，发送短信这件事已经不是交易系统的核心流程了，不要求这件事要立刻就去做，可以放到消息队列中，通知短信系统去发送短信。交易系统直接返回“我们支付成功了”，在后台，短信系统会去消费消息，慢慢告知用户积分的事情。
 *  2、最终一致性  
    有些分布式事务一致性用消息队列来完成，其实这个也算是扩展点。
 *  3、缓冲  
    解决生产者和消费者之间处理速度的不一致的情况，这种情况会导致下游服务被打爆。
 *  可恢复性  
    这其实是解耦带来的好处，如果下游服务中断，数据都存在消息队列中积压，等服务恢复，可继续处理业务。
 *  异步  
    异步处理，这也是解耦的好处，异步处理内容，不需要同步等待，可以提高自己应用处理能力，提高QPS。

#### 分层结构 ####

所有的架构都有一个思路，怎么隔离？做到修改A不影响B，比如我下架一台机器，怎么不影响Kafka集群？只有不停的做分层隔离，才能让上游业务不受下层业务的变动影响，同时，我们写代码也是这个思路，一个微服务工程，我们封装mapper层、service层，不允许其他层跳过service层来调用mapper层，这样如果我们要切换数据源，只需要切换mapper层，不需要改动其他任何内容，这种思想也同样在一些开源软件上有所体现，分布式环境中机器的下线、上线是最为频繁的操作，只有将这一层单独拎出来作为一层来管理机器，我们把这一层叫做A，其他任何消息流、数据流都不关心，只调用A提供的方法，让上层对机器的上线、下线无感知就是一个好的架构，这是我个人的理解，大家可以带着这样的角度来看kafka，看他是如何做到的。

### kafka 怎么保证HA(High Available)即高可用 ###

kafka 0.8以后，提供了HA机制，就是replica副本机制。每个partition的数据都会同步到其他机器上，形成自己的多个replica副本。然后所有replica会选举一个leader出来，那么生产和消费都跟这个leader打交道，然后其他replica就是follower。写的时候，leader会负责把数据同步到所有follower上去，读的时候就直接读leader上数据即可。只能读写leader？很简单，要是你可以随意读写每个follower，那么就要care数据一致性的问题，系统复杂度太高，很容易出问题。kafka会均匀的将一个partition的所有replica分布在不同的机器上，这样才可以提高容错性。

kafka的这种机制，就有所谓的高可用性了，因为如果某个broker宕机了，也没事儿，因为那个broker上面的partition在其他机器上都有副本的，那么此时会重新选举一个新的leader出来，大家继续读写那个新的leader即可。这就有所谓的高可用性了。

### leader的选举 ###

我们知道Zookeeper集群中也有选举机制，是通过Paxos算法，通过不同节点向其他节点发送信息来投票选举出leader，但是Kafka的leader的选举就没有这么复杂了。  
Kafka的Leader选举是通过在zookeeper上创建/controller临时节点来实现leader选举，并在该节点中写入当前broker的信息  
\{“version”:1,”brokerid”:1,”timestamp”:”1512018424988”\}  
利用Zookeeper的强一致性特性，一个节点只能被一个客户端创建成功，创建成功的broker即为leader，即先到先得原则，leader也就是集群中的controller，负责集群中所有大小事务。  
当leader和zookeeper失去连接时，临时节点会删除，而其他broker会监听该节点的变化，当节点删除时，其他broker会收到事件通知，重新发起leader选举。

### 参考博客 ###

[Apache-Kafka核心概念][Apache-Kafka]  
[消息队列常见问题和解决方案][Link 1]  
[rabbitMQ、activeMQ、zeroMQ、Kafka、Redis 比较][rabbitMQ_activeMQ_zeroMQ_Kafka_Redis]  
[快速理解Kafka分布式消息队列框架][Kafka]  
[kafka是如何保证高可用的][kafka]  
[深入学习Kafka：Leader Election - Kafka集群Leader选举过程分析][Kafka_Leader Election - Kafka_Leader]  
图解 kafka 的高可用机制  
[丢消息][Link 2]

[aHR0cHM6Ly9yYXcuZ2l0aHVidXNlcmNvbnRlbnQuY29tL0tsYXVzemhhby9waWN0dXJlL21hc3Rlci9waWN0dXJlL2thZmthL2thZmthJUU2JTlFJUI2JUU2JTlFJTg0JUU1JTlCJUJFLnBuZw]: /images/20211218/c14733b8afc84e61bd4fbb0af1d64594.png
[aHR0cHM6Ly9yYXcuZ2l0aHVidXNlcmNvbnRlbnQuY29tL0tsYXVzemhhby9waWN0dXJlL21hc3Rlci9waWN0dXJlL2thZmthL3RvcGljLXBhcnR0aXRpb24ucG5n]: /images/20211218/7fc88483215546c8ae67b55b0d60f55a.png
[aHR0cHM6Ly9yYXcuZ2l0aHVidXNlcmNvbnRlbnQuY29tL0tsYXVzemhhby9waWN0dXJlL21hc3Rlci9waWN0dXJlL2thZmthL3RvcGljLXBhcnR0aXRpb24tMi5wbmc]: /images/20211218/cedea3afd2404f17bd619ee6c00d7f23.png
[Apache-Kafka]: https://zhuanlan.zhihu.com/p/60110741
[aHR0cHM6Ly9yYXcuZ2l0aHVidXNlcmNvbnRlbnQuY29tL0tsYXVzemhhby9waWN0dXJlL21hc3Rlci9waWN0dXJlL2thZmthL3Byb2R1Y2VyLWNvbnN1bWVyLnBuZw]: /images/20211218/6ac529cbbae74979909eaf065a16e782.png
[Link 1]: http://xuyangyang.club/articles/2018/07/23/1532348839398.html#b3_solo_h2_11
[rabbitMQ_activeMQ_zeroMQ_Kafka_Redis]: https://yq.aliyun.com/articles/590415
[Kafka]: https://blog.csdn.net/colorant/article/details/12081909
[kafka]: https://www.cnblogs.com/756623607-zhang/p/10506360.html
[Kafka_Leader Election - Kafka_Leader]: http://blog.mozhu.org/2017/11/30/kafka/kafka_leader_election.html
[Link 2]: https://www.zhihu.com/question/34842764/answer/67014218