kafka原理及常见问题总结

旧城等待， 2021-12-21 02:13 243阅读 0赞

broker：（kafka的节点，也就是服务器）1.接受来自生产者的消息，为消息设置偏移量，并提交消息到磁盘保存。2.为消费者提供服务，对读取分区的请求作出相应，返回保存到磁盘的消息（大部分工作就是处理客户端、分区副本和控制器发送给分区首领的请求）。

控制器：其实就是一个broker，除了具有一般broker功能之外，还负责分区首领的选举。集群里启动的broker会在zookeeper里创建临时节点/controller，只有一个broker可以创建成功，即为控制器。

请求的处理过程：生产请求和获取请求都必须发送给分区的首领副本。如果broker收到一个针对特定分区的请求，而该分区的首领在另外一个broker上，那么发送请求的客户端会收到一个“非分区首领的错误相应”。kafka客户端要自己负责把生产请求和获取请求发送到正确的broker上。

客户端如何知道哪个是正确的broker（首领副本所在的broker）？  
客户端使用了另一种请求类型，也就是元数据请求。这种请求包含了客户端感兴趣的主题列表，可以发送给任意的broker，broker会响应这个请求。返回给客户端这些主题所含的分区，分区里有哪些副本，以及哪个副本是首领。客户端一般会缓存这些元数据信息，通过定时的请求broker来刷新这些元数据信息。

Producer：消息生产者，向Broker发送消息的客户端。  
Consumer：消息消费者，从Broker读取消息的客户端。  
ConsumerGroup：每个Consumer属于一个特定的Consumer Group，一条消息可以发送到多个不同的Consumer Group，但是一个Consumer Group中只能有一个Consumer能够消费该消息。

Partition：物理上的概念（一个broker对应一个或多个partion），一个topic可以分为多个partition，每个partition内部是有序的  
控制器：就是一个broker，只不过它除了其它broker的功能之外，还负责分区首领的选举。

首领副本（leader）：每个分区都有一个首领副本，所有生产者请求和消费者请求都会经过这个副本。

跟随者副本（follower）：不处理任何客户端请求，唯一的责任就是从首领那里复制消息，保持与首领的一致状态，如果首领崩溃，其中一个跟随者会被选为新的首领。

ISR：Kafka在Zookeeper中动态维护一个ISR，也就是保存同步的副本列表，该列表中保存的是与Leader副本保持消息同步的所有副本对应的代理节点id。如果一个Follower宕机或者其落后太多，则该Follower副本节点将从ISR列表中移除。只要ISR列表中存在一个存活的副本那么已经提交的消息就不会丢失。（kafka集群broker挂掉后，假如正好某个分区首领在这个broker上，会触发首领选举，只有符合选举条件的副本才能有机会成为首领，这个条件就是在LSR中维护，只有LSR中与首领副本保持同步的副本，才能参加选举）

HW俗称高水位，HighWatermark的缩写，取一个partition对应的ISR中最小的LEO作为HW，consumer最多只能消费到HW所在的位置。另外每个replica都有HW,leader和follower各自负责更新自己的HW的状态。对于leader新写入的消息，consumer不能立刻消费，leader会等待该消息被所有ISR中的replicas同步后更新HW，此时消息才能被consumer消费。这样就保证了如果leader所在的broker失效，该消息仍然可以从新选举的leader中获取。对于来自内部broker的读取请求，没有HW的限制。

偏移量：发布到分区的消息会追加到日志文件的尾部，每条消息在日志文件中的位置都会对应一个按序递增的偏移量。不过偏移量不表示消息在磁盘上的位置，而且kafka几乎不允许对消息进行随机读写，消费者可以指定偏移量的的起始位置进行消费。旧版消费者将偏移量保存到Zookeeper中，新版则保存到Kafka内部的一个主题中，消费者也可以保存偏移量到其他地方。不过需要注意分区偏移量和消费者偏移量不是一个概念，一个分区对应一个日志文件某个分区的偏移量其实可以看做该分区当前日志队列的最大长度，而消费者偏移量是说消费者在某个分区上读取到哪个位置，消费者偏移量不可能超过分区偏移量。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h3MTIwMjE5_size_16_color_FFFFFF_t_70]  
**Producer**  
当producer向leader发送数据时，可以通过request.required.acks参数来设置数据可靠性的级别：

1（默认）：这意味着producer在ISR中的leader已成功收到数据并得到确认。如果leader宕机了，则会丢失数据。  
0：这意味着producer无需等待来自broker的确认而继续发送下一批消息。这种情况下数据传输效率最高，但是数据可靠性确是最低的。  
\-1：producer需要等待ISR中的所有follower都确认接收到数据后才算一次发送完成，可靠性最高。但是这样也不能保证数据不丢失，比如当ISR中只有leader时（前面ISR那一节讲到，ISR中的成员由于某些情况会增加也会减少，最少就只剩一个leader），这样就变成了acks=1的情况。

生产者发送消息的模式：  
Kafka的发送模式由producer端的配置参数producer.type来设置，这个参数指定了在后台线程中消息的发送方式是同步的还是异步的，默认是同步的方式，即producer.type=sync。如果设置成异步的模式，即producer.type=async，可以是producer以batch的形式push数据，这样会极大的提高broker的性能，但是这样会增加丢失数据的风险。如果需要确保消息的可靠性，必须要将producer.type设置为sync。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h3MTIwMjE5_size_16_color_FFFFFF_t_70 1]

**Leader选举**  
一条消息只有被ISR中的所有follower都从leader复制过去才会被认为已提交。这样就避免了部分数据被写进了leader，还没来得及被任何follower复制就宕机了，而造成数据丢失。而对于producer而言，它可以选择是否等待消息commit，这可以通过request.required.acks来设置。这种机制确保了只要ISR中有一个或者以上的follower，一条被commit的消息就不会丢失。

有一个很重要的问题是当leader宕机了，怎样在follower中选举出新的leader，因为follower可能落后很多或者直接crash了，所以必须确保选择“最新”的follower作为新的leader。一个基本的原则就是，如果leader不在了，新的leader必须拥有原来的leader commit的所有消息。这就需要做一个折中，如果leader在一个消息被commit前等待更多的follower确认，那么在它挂掉之后就有更多的follower可以成为新的leader，但这也会造成吞吐率的下降。

Kafka在Zookeeper中为每一个partition动态的维护了一个ISR，这个ISR里的所有replica都跟上了leader，只有ISR里的成员才能有被选为leader的可能（unclean.leader.election.enable=false）。在这种模式下，对于f+1个副本，一个Kafka topic能在保证不丢失已经commit消息的前提下容忍f个副本的失败，在大多数使用场景下，这种模式是十分有利的。事实上，为了容忍f个副本的失败，“少数服从多数”的方式和ISR在commit前需要等待的副本的数量是一样的，但是ISR需要的总的副本的个数几乎是“少数服从多数”的方式的一半。

上文提到，在ISR中至少有一个follower时，Kafka可以确保已经commit的数据不丢失，但如果某一个partition的所有replica都挂了，就无法保证数据不丢失了。这种情况下有两种可行的方案：

等待ISR中任意一个replica“活”过来，并且选它作为leader  
选择第一个“活”过来的replica（并不一定是在ISR中）作为leader  
这就需要在可用性和一致性当中作出一个简单的抉择。如果一定要等待ISR中的replica“活”过来，那不可用的时间就可能会相对较长。而且如果ISR中所有的replica都无法“活”过来了，或者数据丢失了，这个partition将永远不可用。选择第一个“活”过来的replica作为leader,而这个replica不是ISR中的replica,那即使它并不保障已经包含了所有已commit的消息，它也会成为leader而作为consumer的数据源。默认情况下，Kafka采用第二种策略，即unclean.leader.election.enable=true，也可以将此参数设置为false来启用第一种策略。

unclean.leader.election.enable这个参数对于leader的选举、系统的可用性以及数据的可靠性都有至关重要的影响。下面我们来分析下几种典型的场景。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h3MTIwMjE5_size_16_color_FFFFFF_t_70 2]  
如果上图所示，假设某个partition中的副本数为3，replica-0, replica-1, replica-2分别存放在broker0, broker1和broker2中。AR=(0,1,2)，ISR=(0,1)。  
设置request.required.acks=-1, min.insync.replicas=2，unclean.leader.election.enable=false。这里将broker0中的副本也称之为broker0起初broker0为leader，broker1为follower。

当ISR中的replica-0出现crash的情况时，broker1选举为新的leader\[ISR=(1)\]，因为受min.insync.replicas=2影响，write不能服务，但是read能继续正常服务。此种情况恢复方案：

尝试恢复(重启)replica-0，如果能起来，系统正常；  
如果replica-0不能恢复，需要将min.insync.replicas设置为1，恢复write功能。  
当ISR中的replica-0出现crash，紧接着replica-1也出现了crash, 此时\[ISR=(1),leader=-1\],不能对外提供服务，此种情况恢复方案：

尝试恢复replica-0和replica-1，如果都能起来，则系统恢复正常；  
如果replica-0起来，而replica-1不能起来，这时候仍然不能选出leader，因为当设置unclean.leader.election.enable=false时，leader只能从ISR中选举，当ISR中所有副本都失效之后，需要ISR中最后失效的那个副本能恢复之后才能选举leader, 即replica-0先失效，replica-1后失效，需要replica-1恢复后才能选举leader。保守的方案建议把unclean.leader.election.enable设置为true,但是这样会有丢失数据的情况发生，这样可以恢复read服务。同样需要将min.insync.replicas设置为1，恢复write功能；  
replica-1恢复，replica-0不能恢复，这个情况上面遇到过，read服务可用，需要将min.insync.replicas设置为1，恢复write功能；  
replica-0和replica-1都不能恢复，这种情况可以参考情形2.  
当ISR中的replica-0, replica-1同时宕机,此时\[ISR=(0,1)\],不能对外提供服务，此种情况恢复方案：尝试恢复replica-0和replica-1，当其中任意一个副本恢复正常时，对外可以提供read服务。直到2个副本恢复正常，write功能才能恢复，或者将将min.insync.replicas设置为1。

**消息传输保障**  
前面已经介绍了Kafka如何进行有效的存储，以及了解了producer和consumer如何工作。接下来讨论的是Kafka如何确保消息在producer和consumer之间传输。有以下三种可能的传输保障（delivery guarantee）:

At most once: 消息可能会丢，但绝不会重复传输  
At least once：消息绝不会丢，但可能会重复传输  
Exactly once：每条消息肯定会被传输一次且仅传输一次  
Kafka的消息传输保障机制非常直观。当producer向broker发送消息时，一旦这条消息被commit，由于副本机制（replication）的存在，它就不会丢失。但是如果producer发送数据给broker后，遇到的网络问题而造成通信中断，那producer就无法判断该条消息是否已经提交（commit）。虽然Kafka无法确定网络故障期间发生了什么，但是producer可以retry多次，确保消息已经正确传输到broker中，所以目前Kafka实现的是at least once。

consumer从broker中读取消息后，可以选择commit，该操作会在Zookeeper中存下该consumer在该partition下读取的消息的offset。该consumer下一次再读该partition时会从下一条开始读取。如未commit，下一次读取的开始位置会跟上一次commit之后的开始位置相同。当然也可以将consumer设置为autocommit，即consumer一旦读取到数据立即自动commit。如果只讨论这一读取消息的过程，那Kafka是确保了exactly once, 但是如果由于前面producer与broker之间的某种原因导致消息的重复，那么这里就是at least once。

考虑这样一种情况，当consumer读完消息之后先commit再处理消息，在这种模式下，如果consumer在commit后还没来得及处理消息就crash了，下次重新开始工作后就无法读到刚刚已提交而未处理的消息，这就对应于at most once了。

读完消息先处理再commit。这种模式下，如果处理完了消息在commit之前consumer crash了，下次重新开始工作时还会处理刚刚未commit的消息，实际上该消息已经被处理过了，这就对应于at least once。

要做到exactly once就需要引入消息去重机制。

**消息去重**  
消费者：将消息的唯一标识保存到外部介质中，每次消费处理时判断是否处理过。  
**在数据生产时避免数据丢失的方法：**  
1）同步模式：确认机制设置为-1，也就是让消息写入leader和所有的副本。  
2）异步模式：如果消息发出去了，但还没有收到确认的时候，缓冲池满了，在配置文件中设置成不限制阻塞超时的时间，也就说让生产端一直阻塞，这样也能保证数据不会丢失  
**消费者再均衡**  
读取分区的所有权从群组内的一个消费者转移到另一个消费者。分区的再均衡策略保证了消费者群组的高可用性和伸缩性（我们可以放心地添加或移除消费者）。  
在再均衡期间，消费者无法读取消费，造成整个群组一小段时间内的不可用。当分区被重新分配给另一个消费者时，消费者当前的读取状态会丢失，它可能需要去刷新缓存，在它重新恢复之前会拖慢应用程序。  
消费者通过向指派为群组协调器的broker（不同的群组可以有不同的协调器）发送心跳来维持它们和群组的从属关系以及它们对分区的所有权。消费者会在轮询消息或者提交偏移量时发送心跳。如果消费者停止发送心跳的时间足够长，会话就过期，群组协调器就会认为它已经死亡，就会触发一次再均衡。用户可以主动调用close()方法来关闭连接，这样可以立即触发再均衡。

**保证消息顺序消费**  
在1个topic中，有3个partition，那么如何保证数据的消费？  
1、生产者在写的时候，可以指定一个 key，比如说我们指定了某个订单 id 作为 key，那么这个订单相关的数据，一定会被分发到同一个 partition 中去，而且这个 partition 中的数据一定是有顺序的。  
2、消费者从 partition 中取出来数据的时候，也一定是有顺序的。到这里，顺序还是 ok 的，没有错乱。  
3、但是消费者里可能会有多个线程来并发来处理消息。因为如果消费者是单线程消费数据，那么这个吞吐量太低了。而多个线程并发的话，顺序可能就乱掉了。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h3MTIwMjE5_size_16_color_FFFFFF_t_70 3]  
写N个queue，将具有相同key的数据都存储在同一个queue，然后对于N个线程，每个线程分别消费一个queue即可  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h3MTIwMjE5_size_16_color_FFFFFF_t_70 4]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h3MTIwMjE5_size_16_color_FFFFFF_t_70]: /images/20211221/579a2d74aa0148f981a9dca0e395ca55.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h3MTIwMjE5_size_16_color_FFFFFF_t_70 1]: /images/20211221/9459f1376aba4185834e843354b4b549.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h3MTIwMjE5_size_16_color_FFFFFF_t_70 2]: /images/20211221/ad167f2d7aee46da89eba78f71a3c6d7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h3MTIwMjE5_size_16_color_FFFFFF_t_70 3]: /images/20211221/338aa989b2f1468eb0c42d46383e8690.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h3MTIwMjE5_size_16_color_FFFFFF_t_70 4]: /images/20211221/0a08a216063840f8aa347677844109cb.png