kafka快速入门---思考与实现

Posted 2022-05-20 YaoYong_BigData

一、为什么需要消息引擎？

由于工作中经常用到kafka，但是对kafka的一些内部机制不是很熟悉，所以最近在看kafka相关的知识，我们知道kafka非常经典的消息引擎，它以高性能、高可用著称。那么问题来了，它是怎么做到高性能、高可用的？它的消息是以什么样的形式持久化的？既然写了磁盘，为何速度还那么快？它是如何保证消息不丢失的...？带着这一系列的问题，我们来扒开kafka的面纱。

首先我们思考这样一个问题：为什么需要消息引擎？为什么不能直接走rpc？ 以一个订单系统为例：当我们下了一个订单的时候，应该是要先减商品库存，然后用户支付扣钱，商家账户加钱...，最后可能还要发推送或者短信告诉用户下单成功，告诉商家来订单了。

这整个下单过程，如果全部同步阻塞，那么耗时会增加，用户等待的时间会加长，体验不太好，同时下单过程依赖的链路越长，风险越大。为了加快响应，减少风险，我们可以把一些非必须卡在主链路中的业务拆解出去，让它们和主业务解耦。下单的最关键核心就是要保证库存、用户支付、商家打款的一致性，消息的通知完全可以走异步。这样整个下单过程不会因为通知商家或者通知用户阻塞而阻塞，也不会因为它们失败而提示订单失败。

接下来就是如何设计一个消息引擎了，宏观来看一个消息引擎支持发送、存储、接收就行了：

那么如上图一个简易消息队列模型出现了，Engine把发送方的消息存储起来，这样当接收方来找Engine要数据的时候，Engine再从存储中把数据响应给接收放就ok了。既然涉及到持久化的存储，那么缓慢的磁盘IO是要考虑的问题。还有接收方可能不止一个，以上述订单为例，下单完成之后，通过消息把完成事件发出去，这时候负责用户侧推送的开发需要消费这条消息，负责商户侧推送的开发也需要消费这条消息，能想到的最简单的做法就是copy出两套消息，但是这样是不是显得有点浪费？高可用也是一个需要考虑的点，那么我们的engine是不是得副本，有了副本之后，如果一个engine节点挂掉，我们可以选举出一个新副本来工作。光有副本也不行，发送方可能也是多个，这时候如果所有的发送方都把数据打到一个Leader（主）节点上似乎也不合理，单个节点的压力太大。可能你会说：不是有副本吗？让接收方直接从副本读取消息。这样的话又带来另一个问题：副本复制Leader的消息延迟了咋办？读不到消息再读一次Leader？如果这样的话，引擎的设计的貌似更加复杂了，似乎不太合理。那就得想一种既能不通过副本又能分散单节点压力就行了，答案就是分片技术，既然单个Leader节点压力太大，那么就分成多个Leader节点，我们只需要一个好的负载均衡算法，通过负载均衡把消息平均分配到各个分片节点就好了，于是我们可以设计出一套大概长这样的生产者-消费者模型。

但是这些只是简单的想法，具体如何实现还是很复杂的，带着这一系列问题和想法，我们来看看kafka是如何实现的。

二、一条消息该怎么设计？

消息是服务的源头，一切的设计都是为了将消息从一端送到另一端，这里面涉及到消息的结构，消息体不能太大，太大容易造成存储成本上升，网络传输开销变大，所以消息体只需要包含必要的信息，最好不要冗余。消息最好也支持压缩，通过压缩可以在消息体本身就精简的情况下变的更小，那么存储和网络开销可以进一步降低。消息是要持久化的，被消费掉的消息不能一直存储，或者说非常老的消息被再次消费的可能性不大，需要一套机制来清理老的消息，释放磁盘空间，如何找出老的消息是关键，所以每个消息最好带个消息生产时的时间戳，通过时间戳计算出老的消息，在合适的时候进行删除。消息也是需要编号的，编号一方面代表了消息的位置，另一方面消费者可以通过编号找到对应的消息。大量的消息如何存储也是个问题，全部存储在一个文件中，查询效率低且不利于清理老数据，所以采用分段，通过分段的方式把大的日志文件切割成多个相对小的日志文件来提升维护性，这样当插入消息的时候只要追加在段的最后就行，但是在查找消息的时候如果把整个段加载到内存中一条一条找，似乎也需要很大的内存开销，所以需要一套索引机制，通过索引来加速访问对应的Message。

总结：一条kafka的消息包含创造时间、消息的序号、支持消息压缩，存储消息的日志是分段存储，并且是有索引的。

三、为什么需要Topic？

宏观来看消息引擎就是一发一收，有个问题：生产者A要给消费者B发送消息，同时也要给消费者C发送消息。那么消费者B和消费者C如何只消费到自己需要的数据？能想到的简单的做法就是在消息中加Tag，消费者根据Tag来获取自己的消息，不是自己的消息直接跳过，但是这样似乎不太优雅，而且存在cpu资源浪费在消息的过滤上。所以最有效的办法就是对于给B消息不会给C，给C的消息不会给B，这就是Topic。通过Topic来区分不同的业务，每个消费者只需要订阅自己关注的Topic即可，生产者把消费者需要的消息通过约定好的Topic发过去，那么简单的理解就是消息按照Topic分类了。

总结：Topic是个逻辑的概念，Topic可以很好的做业务划分，每个消费者只需要关注自己的Topic即可。

四、分区如何保证顺序？

通过上文我们知道分区的目的就是分散单节点的压力，再结合Topic和Message，那么消息的大概分层就是Topic（主题）->Partition（分区）->Message（消息）。也许你会问，既然分区是为了降低单节点的压力，那么干嘛不用多个topic代替多个分区，在多个机器节点的情况下，我们可以把多个topic部署在多个节点上，似乎也能实现分布式，简单一想似乎可行，仔细一想，还是不对。我们最终还要服务业务的，这样的话，本来一个topic的业务，要拆解成多个topic，反而把业务的定义打散了。

好吧，既然有多个分区了，那么消息的分配是个问题，如果topic下面的数据过于集中在某个分区上，又会造成分布不均匀，解决这个问题，一套好的分配算法是很有必要的。

kafka支持轮询法，即在多分区的情况下，通过轮询可以均匀地把消息分给每个分区，这里需要注意的是，每个分区里的数据是有序的，但是整体的数据是无法保证顺序的，如果你的业务强依赖消息的顺序，那么就要慎重考虑这种方案，比如生产者依次发了A、B、C三个消息，它们分别分布在3个分区中，那么有可能出现的消费顺序是B、A、C。

那么如何保证消息的顺序性？从整体的角度来看，只要分区数大于1，就永远无法保证消息的顺序性，除非你把分区数设置成1，但是这样的话吞吐就是问题。从实际的业务场景来说，一般我们可能需要某个用户的消息、或者某个商品的消息有序就可以了，用户A和用户B的消息谁先谁后没关系，因为它们之间没什么关联，但是用户A的消息我们可能要保持有序，比如消息描述的是用户的行为，行为的先后顺序是不能乱的。这时候我们可以考虑用key hash的方式，同一个用户id，通过hash始终能保持分到一个分区上，我们知道分区内部是有序的，所以这样的话，同一个用户的消息一定是有序的，同时不同的用户可以分配到不同的分区上，这样也利用到了多分区的特性。

总结：kafka整体消息是无法保证有序的，但是单个分区的消息是可以保证有序的。

五、如何设计一个合理的消费者模型？

既然是设计消息模型，那么消费者必不可少，实现消费者最简单的方式就是起一个进程或者线程直接去broker里面拉取消息即可，这很合理，但是如果生产的速度大于当前的消费速度怎么办？第一时间想到的就是再起一个消费者，通过多个消费者来提升消费速度，这里似乎又有个问题，两个消费者都消费到了同一条消息怎么办？加锁是个解决方案，但是效率会降低，也许你会说消费的本质就是读，读是可以共享的，只要保证业务幂等，重复消费消息也没关系。这样的话，如果10个消费者都争抢到了同样的消息，结果有9个消费者都是白白浪费资源的。因此在需要多个消费者提升消费能力的同时，还要保证每个消费者都消费到没被处理的消息，这就是消费者组，消费者组下面可以有多个消费者，我们知道topic是分区的，因此只要消费者组内的每个消费者订阅不同的分区就可以了。理想的情况下是每个消费者都分配到相同数据量分区，如果某个消费者获得的分区数不平均（较多或者较少），出现数据倾斜状态，那么就会导致某些消费者非常繁忙或者轻松，这样就不合理，这就需要一套均衡的分配策略。

kafka消费者分区分配策略主要有3种：

1.Range：这种策略是针对topic的，会把topic的分区数和消费者数进行一个相除，如果有余数，那就说明多余的分区不够平均分了，此时排在前面的消费者会多分得1个分区，乍看其实挺合理，毕竟本来数量就不均衡。但是如果消费者订阅了多个topic，并且每个topic平均算下来都多几个个分区，那么对于排在前面的消费者就会多消费很多分区。

Range范围分配策略是Kafka默认的分配策略，它可以确保每个消费者消费的分区数量是均衡的。

注意：Rangle范围分配策略是针对每个Topic的。

配置

配置消费者的partition.assignment.strategy为org.apache.kafka.clients.consumer.RangeAssignor。

算法公式

n = 分区数量 / 消费者数量

m = 分区数量 % 消费者数量

前m个消费者消费n+1个

剩余消费者消费n个

由于是按照topic维度来划分的，所以最终:

• c1消费 Topic0-p0、Topic0-p1、Topic1-p0、Topic1-p1
• c2消费 Topic0-p2、Topic1-p2

最终可以发现消费者c1比消费者c2整整多两个分区，完全可以把c1的分区分一个给c2，这样就可以均衡了。

2.RoundRobin：这种策略的原理是将消费组内所有消费者以及消费者所订阅的所有topic的partition按照字典序排序，然后通过轮询算法逐个将分区以此分配给每个消费者。假设现在有两个topic，每个topic3个分区，并且有3个消费者。那么大致消费状况是这样的：

• c0消费 Topic0-p0、Topic1-p0
• c1消费 Topic0-p1、Topic1-p1
• c2消费 Topic0-p2、Topic1-p2

看似很完美，但是如果现在有3个topic，并且每个topic分区数是不一致的，比如topic0只有一个分区，topic1有两个分区，topic2有三个分区，而且消费者c0订阅了topic0，消费者c1订阅了topic0和topic1，消费者c2订阅了topic0、topic1、topic2，那么大致消费状况是这样的：

• c0消费 Topic0-p0
• c1消费 Topic1-p0
• c2消费 Topic1-p1、Topic2-p0、Topic2-p1、Topic2-p2

这么看来RoundRobin并不是最完美的，在不考虑每个topic分区吞吐能力的差异，可以看到c2的消费负担明显很大，完全可以将Topic1-p1分区分给消费者c1。

3.Sticky：

从Kafka 0.11.x开始，引入此类分配策略。主要目的：

1. 分区分配尽可能均匀；
2. 在发生rebalance的时候，分区的分配尽可能与上一次分配保持相同。

没有发生rebalance时，Striky粘性分配策略和RoundRobin分配策略类似。

引入目的之一就是：分区的分配要尽可能均匀。以上面RoundRobin 3个topic分别对应1、2、3个分区的case来说，因为c1完全可以消费Topic1-p1，但是它没有。针对这种情况，在Sticky模式下，就可以做到把Topic1-p1分给c1。

Sticky引入目的之二就是：分区的分配尽可能与上次分配的保持相同。这里主要解决就是rebalance后分区重新分配的问题，假设现在有3个消费者c0、c1、c2，他们都订阅了topic0、topic1、topic2、topic3，并且每个topic都有两个分区，此时消费的状况大概是这样：

 这种分配方式目前看RoundRobin没什么区别，但是如果此时消费者c1退出，消费者组内只剩c0、c2。那么就需要把c1的分区重新分给c0和c2，我们先来看看RoundRobin是如何rebalance的：

可以发现原来c0的topic1-p1分给了c2，原来c2的topic1-p0分给了c0。这种情况可能会造成重复消费问题，在消费者还没来得及提交的时候，发现分区已经被分给了一个新的消费者，那么新的消费者就会产生重复消费。但是从理论的角度来说，在c1退出之后，可以没必要去动c0和c2的分区，只需要把原本c1的分区瓜分给c0和c2即可，这就是sticky的做法：

需要注意的是Sticky策略中，如果分区的分配要尽可能均匀和分区的分配尽可能与上次分配的保持相同发生冲突，那么会优先实现第一个。

总结：kafka默认支持以上3种分区分配策略，也支持自定义分区分配，自定义的方式需要自己去实现，从效果来看RoundRobin要好于Range的，Sticky是要好于RoundRobin的，推荐大家使用版本支持的最好的策略。