Kafka学习（2）- 架构与细节

今天药忘吃喽~ 2022-03-29 03:55 220阅读 0赞

说明：windows下操作

Kafka 中文官方文档：[http://kafka.apachecn.org/documentation.html\#gettingStarted][http_kafka.apachecn.org_documentation.html_gettingStarted]

**目录：**

1.  架构
    
    1.  broker
    2.  主题
    3.  分区
    4.  生产者
    5.  消费者
2.  副本因子
3.  工作流程
4.  集群搭建
5.  代码开发
6.  复制原理
7.  ISR
8.  数据丢失与重复的解决方式
9.  kafka消息传输保障
10. kafka可视化工具
11. 深入理解

**内容：**

1. 集群架构 
 ![Cluster Architecture][] 
 
 说明：
 
 <table> 
 <tbody> 
 <tr> 
 <th>S.No</th> 
 <th>组件和说明</th> 
 </tr> 
 <tr> 
 <td>1</td> 
 <td> Broker（代理） Kafka集群通常由多个代理组成以保持负载平衡。Kafka代理是无状态的，所以他们使用ZooKeeper来维护它们的集群状态。&nbsp;一个Kafka代理实例可以每秒处理数十万次读取和写入，每个Broker可以处理TB的消息，而没有性能影响。&nbsp;Kafka经纪人领导选举可以由ZooKeeper完成。 </td> 
 </tr> 
 <tr> 
 <td>2</td> 
 <td> ZooKeeper ZooKeeper用于管理和协调Kafka代理。&nbsp;ZooKeeper服务主要用于通知生产者和消费者Kafka系统中存在任何新代理或Kafka系统中代理失败。&nbsp;根据Zookeeper接收到关于代理的存在或失败的通知，然后产品和消费者采取决定并开始与某些其他代理协调他们的任务。 </td> 
 </tr> 
 <tr> 
 <td>3</td> 
 <td> Producers（生产者） 生产者将数据推送给经纪人。&nbsp;当新代理启动时，所有生产者搜索它并自动向该新代理发送消息。&nbsp;Kafka生产者不等待来自代理的确认，并且发送消息的速度与代理可以处理的一样快。 </td> 
 </tr> 
 <tr> 
 <td>4</td> 
 <td> Consumers（消费者） 因为Kafka代理是无状态的，这意味着消费者必须通过使用分区偏移来维护已经消耗了多少消息。&nbsp;如果消费者确认特定的消息偏移，则意味着消费者已经消费了所有先前的消息。&nbsp;消费者向代理发出异步拉取请求，以具有准备好消耗的字节缓冲区。&nbsp;消费者可以简单地通过提供偏移值来快退或跳到分区中的任何点。&nbsp;消费者偏移值由ZooKeeper通知。 </td> 
 </tr> 
 </tbody> 
 </table>
2. 关于broker
 
 1. 一个borker服务下，是否可以创建多个分区？ 
 可以，broker数与分区数没有关系。
 2. 一个broker服务下，是否可以创建多个副本因子 
 不可以，副本因子应该小于broker数，会报错 replication factor:3 larger than available brokers
3. 关于主题命令
 
 1. 创建主题时，副本因子应该小于等于可用的broker数。
 2. 创建、删除、修改、查看主题结构、查看是否有某主题
 
 windows系统下
 1.创建一个主题
 创建一个名为“test”的topic，它有一个分区和一个副本：
 >bin/windows/kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
 
 replication-factor：控制有多少服务器将复制每个写入的消息。
 如果您设置了3个复制因子，那么只能最多2个相关的服务器能出问题，否则您将无法访问数据。
 我们建议您使用2或3个复制因子，以便在不中断数据消费的情况下透明的调整集群。
 partitions：参数控制 topic 将被分片到多少个日志里。partitions 会产生几个影响。
 首先，每个分区只属于一台服务器，所以如果有20个分区，那么全部数据(包含读写负载)将由不超过20个服务器（不包含副本）处理。
 最后 partitions 还会影响 consumer 的最大并行度。
 
 2.查看当前有哪些主题
 运行list（列表）命令来查看这个topic：
 >bin/windows/kafka-topics.bat --list --zookeeper localhost:2181
 
 3.查看主题结构
 >bin/windows/kafka-topics.bat --describe --zookeeper localhost:2181
 
 PartitionCount:分区数量
 ReplicationFactor：复制因子
 Leader:broker.id=0（server.properties文件中，如果有多个broker，那么broker.id不能相同）
 	因为在kafka中如果有多个副本的话，就会存在leader与follower的关系
 	表示当前这个副本为leader所在的broker是哪一个
 Replicas：所有的副本列表
 Isr：可用的副本列表，第9个知识点会对它详解
 
 4.修改topic：增加分区
 > bin/windows/kafka-topics.bat --zookeeper localhost:2181 --alter --topic test --partitions 40
 
 5.删除主题
 >bin/windows/kafka-topics.bat --zookeeper localhost:2181 --delete --topic test
 说明：
 1.当执行删除命令之后，topic不是物理删除，而是一个标记删除的操作。
 mytopic4 - marked for deletion.
 2.标记删除之后的主题是否还可以继续生产数据？
 可以生产数据，不会有影响，但是当服务器重启就会删除已经标记的topic。
 3.注意：有些版本就算重启之后也不会删除，如果需要删除topic，需要在server.properties文件中加入delete.topic.enable=true
 3. 修改配置项
 
 1. 增加一个配置项 
 >bin/windows/kafka-topics.bat --zookeeper localhost:2181 --alter --topic test --config flush.messages=1
 2. 删除一个配置项 
 >bin/windows/kafka-topics.bat --zookeeper localhost:2181 --alter --topic test --delete-config flush.messages
 4. 生产者
 
 1. 创建生产者：kafka-console-producer.bat --broker-list localhost:9092 --topic test
 5. 消费者
 
 1. 创建消费者：kafka-console-consumer.bat --zookeeper localhost:2181 --topic test
4. 关于分区
 
 1. 概念 
 是针对于某一个主题下数据的拆分。 
 一个broker服务下可以有多个分区，每个topic都是有序且不可变的消息队列。但不同的分区数据是无序的。 
 每个topic都有一个或者多个Partitions构成。 
 任何发布到此partition的消息都会被直接追加到log文件的尾部（也就是分区的尾部），每一条消息在此文件中的位置称为offset（偏移量），offset是一个long型数字，它唯一标记一条消息。消费者通过（offset、partition、topic）跟踪记录。
 2. 作用：让消费者在拿到数据的时候，速度会更快。
 3. 分区和消费组的关系 
 Topic的Partition数量可以在创建主题时配置。 
 消费组：有一个或者多个消费者组成，同一个组中的消费者对于同一条消息只消费一次。 
 某一个主题下的分区数，对于消费组来说，应该小于等于该主题下的分区。也就是说Partition数量决定了每个Consumer group中并发消费者的最大数量。 
 如：某一个主题下有4个分区，那么消费组中的消费者应该小于4，而且最好与分区数成整数倍（1，2，4）。
 4. 分区中的数据都有相应的编号（offset--偏移量） 
 偏移量：记录当前有多少记录数，而且让消费者可以知道自己消费到什么地方。 
                 可以让消费者自定义选择消费某一条消息（对于同一条消息，消费者可以消费多次） 
 确定一条唯一的数据：主题+分区+偏移量
5. 副本因子（副本因子只能增，不能减少 ）
 
 1. 概念 
 创建副本因子的时候，副本因子数应该小于等于可以用的broker数。 
 副本因子操作的单位是以分区为单位的。 
 当有多个副本数时，kafka并不是将多个副本同时对外提供读取和写入，只有leader对外提供读取和写入。 
 副本因子数包含本身，同一个副本因子不能放在同一个broker中。 
 在有多个副本的情况下，kafka会为同一个分区下的副本，设定角色关系：一个leader和N个follower。 
 leader负责与外面（生产者、消费者）的角色进行读写处理。
 2. 作用：让kafka读取和写入数据时的高可靠。
 3. 如果某一个分区中有三个副本因子。就算其中一个挂掉，那么只会在剩下的两个中选择一个leader。而不会在其它的broker中另启一个副本（因为在另外一台启动的话，存在数据传递。只要在主机之间有数据传递，就会长时间占用网络IO，但是kafka是一个高吞吐量的消息系统，这种情况不允许发生）所以不会在另一个broker中启动。
 4. 如果所有的副本都挂了，生产者如果生产的数据是定制分区的话，将写入不成功。
 5. ISR:可用的副本因子数
 6. ack机制：确认机制
 7. 在消费者拿取数据的时候，可以采用kafka的两种api：高级api和低级api
 
 1. 高级api:让用户使用时，很方便。大部分的操作都是已经封装好的。比如，当前消费到哪个位置下。但是不够灵活。推荐使用的方式
 2. 低级api:没有进行包装，所有的操作由用户决定，如自己得保存某一个分区下的记录，你当前消费到哪个位置。
 8. leader broker是之前的一种方法，现在其实不用了。 
 kafka集群中包含了很多的broker。但是在这么多的broker中也会有一个老大存在， 
 是在kafka节点中的一个临时节点中，去创建相应的数据。叫做controller broker，管理所有的broker
6. Kafka 工作流程
 
 * Kafka只是分为一个或多个分区的主题的集合。 Kafka分区是消息的线性有序序列，其中每个消息由它们的索引(称为偏移)来标识。 Kafka集群中的所有数据都是不相连的分区联合。 传入消息写在分区的末尾，消息由消费者顺序读取。 通过将消息复制到不同的代理提供持久性。
 * Kafka以快速，可靠，持久，容错和零停机的方式提供基于pub-sub和队列的消息系统。 在这两种情况下，生产者只需将消息发送到主题，消费者可以根据自己的需要选择任何一种类型的消息传递系统。
 * 发布 - 订阅消息的工作流程
 
 * 生产者定期向主题发送消息。
 * Kafka代理存储为该特定主题配置的分区中的所有消息。 它确保消息在分区之间平等共享。 如果生产者发送两个消息并且有两个分区，Kafka将在第一分区中存储一个消息，在第二分区中存储第二消息。
 * 消费者订阅特定主题。
 * 一旦消费者订阅主题，Kafka将向消费者提供主题的当前偏移，并且还将偏移保存在Zookeeper系统中。
 * 消费者将定期请求Kafka(如100 Ms)新消息。
 * 一旦Kafka收到来自生产者的消息，它将这些消息转发给消费者。消费者将收到消息并进行处理。
 * 一旦消息被处理，消费者将向Kafka代理发送确认。
 * 一旦Kafka收到确认，它将偏移更改为新值，并在Zookeeper中更新它。 由于偏移在Zookeeper中维护，消费者可以正确地读取下一封邮件，即使在服务器暴力期间。
 * 以上流程将重复，直到消费者停止请求。
 * 消费者可以随时回退/跳到所需的主题偏移量，并阅读所有后续消息。
 * 队列消息/用户组的工作流：在队列消息传递系统而不是单个消费者中，具有相同组ID 的一组消费者将订阅主题。 简单来说，订阅具有相同 Group ID 的主题的消费者被认为是单个组，并且消息在它们之间共享。 
 
 * 生产者以固定间隔向某个主题发送消息。
 * Kafka存储在为该特定主题配置的分区中的所有消息，类似于前面的方案。
 * 单个消费者订阅特定主题，假设 Topic-01 为 Group ID 为 Group-1 。
 * Kafka以与发布 - 订阅消息相同的方式与消费者交互，直到新消费者以相同的组ID 订阅相同主题 Topic-01 。
 * 一旦新消费者到达，Kafka将其操作切换到共享模式，并在两个消费者之间共享数据。 此共享将继续，直到用户数达到为该特定主题配置的分区数。
 * 一旦消费者的数量超过分区的数量，新消费者将不会接收任何进一步的消息，直到现有消费者取消订阅任何一个消费者。 出现这种情况是因为Kafka中的每个消费者将被分配至少一个分区，并且一旦所有分区被分配给现有消费者，新消费者将必须等待。
 * 此功能也称为使用者组。 同样，Kafka将以非常简单和高效的方式提供两个系统中最好的。
7. kafka集群搭建：[kafka伪分布式搭建][kafka]
 
 1. 准备好zookeeper集群
 2. 修改server.properties 
 各台主机的broker.id设置成不同 
 zk使用自己的集群 zookeeper.connect= 
 对log.dirs中进行修改
8. kafka Java开发：官方文档查看：[kafka使用Java开发-1-入门][kafka_Java_-1-]
 
 1. 生产者代码
 2. 消费者代码
 3. 手动提交
 
 1. 为什么要手动提交
 2. 手动提交的方式
 4. 自定义分区
 
 1. 为什么要自定义分区
 2. 自定义分区流程
 
 1. 创建自定义类，实现org.apache.kafka.clients.producer.Partitioner接口
 2. 重写public int partition方法
 3. 配置项中加入partitioner.class属性
 5. kafka log说明及优化
9. 复制原理 
 当producer发送一条消息到broker后，leader写入消息并复制到所有follower。消息提交之后才被成功复制到所有的同步副本中。消息复制延迟受最慢的follower限制，重要的是快速检测慢副本，如果follower“落后”太多或者失效，leader将会将它从ISR中删除。 
 kafka每个topic的partition有N个副本（replicas）。 
 kafka通过多副本机制实现故障自动转移，当kafka集群中一个broker失效情况下仍然保证服务可用。 
 kafka中发生复制是确保partition的日志能有序地写到其他节点上，N个replicas中，其中一个replicas为leader，其他都为follower，leader处理partition的所有读写请求，与此同时，follower会被动定期地去复制leader上的数据。 
 kafka提供了数据复制算法保证，如果leader发生故障或挂掉，一个新leader被选举并被接受客户端的消息成功写入。 
 leader负责维护和跟踪ISR ( In-Sync-Replicas的缩写，表示副本同步队列中所有follower滞后的状态。）
 
 1. kafka中topic的每个partition有一个预定式的日志文件，虽然partition可以继续细分为若干个segment文件，但是对于上层应用来说可以将partition看成最小的存储单元，每个partition都有一些列有序的、不可变的消息组成，这些消息被连续的追加到partition中。 
 ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lpZ3VhbmdfODIw_size_16_color_FFFFFF_t_70][]
 2. HW：HighWatermark（最高水平线）。因为用户发送一条消息过来之后，先会进入到Leader中，追加到log文件的末尾，但是follower与leader之间进行同步的时候，是有一定的时间差，有可能存在正在同步的路上时，leader挂掉了，3就不会同步成功。这时broker的处理方式是，将follow转为leader,所以说用户在消费的时候，只能消费到hw之前的数据，即HW为多个follower与leader之前已经同步的数据线。 
 LEO： 
 ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lpZ3VhbmdfODIw_size_16_color_FFFFFF_t_70 1][]
 3. timeout：生产者向kafka发送消息之后，如果在指定timeout时间内，没有给出反馈，将认为发送失败，对于生产者来说，如果已经认为发送失败，就会采取处理办法【重新发送】 
 ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lpZ3VhbmdfODIw_size_16_color_FFFFFF_t_70 2][]
10. ISR： In-Sync-Replicas（副本同步队列）说明：可用的副本列表
 
 1. ISR（In-Sync Replicas）是指副本同步队列。
 2. ISR是所有副本的一个子集，由leader维护ISR列表，follower从leader同步数据有一些延迟 
 （包括延迟时间replica.lag.time.max.ms和延迟条数replica.lag.max.message两个维度，当前最新的版本0.10.x中只支持延迟时间replica.lag.time.ms这个维度，任意一个超过阙值都会把follower踢出ISR，存入OSR（Outof-Sync Replica）列表，新加入的follower也会先存放在OSR中）
 3. leader新写入的消息，comsumer不能立即消费，leader会等待该消息被所有ISR中replicas同步后更新HW，此时消息才能被comsumer消费。这样就保证了如果leader所在的broker失效，该消息仍然可以从新选举的leader中获取。对于来自内部broker的读取请求，没有HW的限制。
 4. 思考：同步复制要求所有能工作的follower都复制完，这条消息才会被commit，这种复制方式是否极大的影响了吞吐率？ 
 那么使用异步复制方式可以吗？ follower异步的从leader复制数据，数据只要被leader写入log就会被认为已经commit，这种情况下如果follower都还没有复制完，落后于leader时，突然leader宕机，则会丢失数据。而kafka的这种使用ISR的方式则会很好的均衡了确保数据不会丢失及吞吐率。 
 
 唯一的警告是，在追赶阶段，追随者可能必须在其HW之前截断数据，由于复制是异步的，因此不能保证提交消息能够承受任何代理失败。
11. 数据丢失与重复的解决方式
 
 1. 生产方发送消息到kafka造成数据丢失
 
 1. 数据丢失和重复的情况说明
 
 1. 数据丢失：生产方向kafka写数据的时候，突然broker连接不上，会导致数据丢失，可以采用callback（回调函数）的方式进行处理（判断异常信息是否为空，如果为空表示正常发送了，否则就有异常，可进行特殊处理（解决方案））
 2. 数据丢失的解决方案：
 
 1. 单机：出现异常怎么处理？（只有单机的情况才会出现broker连接不上，broker崩溃才会连接不上，在集群模式下，会选举新的leader进行连接）
 
 1. 方法一：拿到没有发送成功的数据，等broker恢复之后再次发送
 2. 方法二：将消费的数据写入到mysql，如果消费成功就反馈一个结果，status的值为Y，异常就为N，再做其他的处理
 3. 方法三：出现异常就将正确的数据写入到redis中，开启一个消费再次消费，可以设置比较长的时间 
 ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lpZ3VhbmdfODIw_size_16_color_FFFFFF_t_70 3][]
 4. 可以设置重试次数：参数retries=3,只能解决进入错误处理逻辑次数减少。 
 ![20190114225412879.png][]
 2. kafka集群内部之间造成数据丢失 
 集群：kafka集群内部之间有ack确认机制，防止数据丢失和重复（复制原理中有对timeout 的说明）
 
 1. 在timeout内，所有的ISR列表中的副本都需要确认--all 
 注意：all有一个补充说明：可以设置参数min.insync.replicas（最小副本同步数）来保证最小个数的副本实现同步就可以，而不是所有的副本
 2. 在timeout内，只需要leader确认--1
 3. 在timeout内，不需要任何的副本进行确认--0 
 设置实例： 
 ![20190114225117582.png][]
 3. 消费者从kafka拉数据造成的数据丢失
 
 1. 数据丢失和重复的情况说明
 
 1. 数据丢失：如果消费者在处理数据的时候出现异常，采用自动提交的话，offset的值已经发生改变。但该条消息没有被真正的处理，这样就造成数据丢失。
 2. 数据重复：以下两种情况
 
 1. 当消费者成功消费从kafka中拉来的数据后，向kafka提交的过程中出现了错误，导致offset值没有发生改变，当错误解决之后，消费者再次从kafka获得数据时，是已经消费的数据，导致数据重复消费
 2. 消费端消费能力比较低,处理消息速度慢。消息处理完之后提交下一个消费的offset,而在session-time-out前,消息还没有处理完,但是已经超时被kafka视为消费失败了,导致一直重复消费
 2. 解决方案：
 
 1. 数据丢失的解决方案：
 
 1. 手动提交代替自动提交，可以解决数据丢失，[kafka使用Java开发-3-手动提交][kafka_Java_-3-]
 2. 数据重复的解决方案
 
 1. 手动提交或者延长session-time-out（消费者处理数据，给kafka反馈的时间）
 2. 开发者：消息可以使用唯一id标识作为key ，消费者处理前先根据key在redis中查询是否有此条消息，如果没有就直接处理，处理完成后将结果存储到redis中，如果有就手动提交给kafka，表明数据处理完成。 
 ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lpZ3VhbmdfODIw_size_16_color_FFFFFF_t_70 4][]
12. kafka消息传输保障 
 kakf确保消息在produce和consumer之间传输，有以下三种可能的传输保障
 
 1. At most once：消息可能会丢，但绝不会重复传输
 2. At least once：消息绝不会丢，但可能会重复传输
 3. Excetly once：每条消息肯定会被传输一次且传输一次
13. kafka可视化工具
 
 1. Kafka Manager
 
 1. 雅虎开源的Kafka集群管理工具 
 管理几个不同的集群 
 监控集群的状态（topics、broker、副本分区、分区分区） 
 产生分区分配（Generate  partition assignments）基于集群的当前状态 
 重新分配分区
 2. [kafka集群管理工具kafka-manager部署安装][kafka_kafka-manager]
 3. [https://blog.csdn.net/x541211190/article/details/81191608][https_blog.csdn.net_x541211190_article_details_81191608]
 2. KafkaOffsetMonitor 
 kafkaOffsetMonitor可以实时监控：
 
 1. kafka集群状态 
 Topic、Consumer Group列表 
 图形化展示topic和consumer之间的关系 
 图形化展示consumer的offset、lag等信息
14. 另外推荐一篇对kafka深入解读的博客：kafka数据可靠性深度解读[https://blog.csdn.net/u013256816/article/details/71091774][https_blog.csdn.net_u013256816_article_details_71091774]

[http_kafka.apachecn.org_documentation.html_gettingStarted]: http://kafka.apachecn.org/documentation.html#gettingStarted
[Cluster Architecture]: /images/20220329/6074438e60f648f3aa164ff909e330fd.png
[kafka]: https://blog.csdn.net/yiguang_820/article/details/86185186
[kafka_Java_-1-]: https://blog.csdn.net/yiguang_820/article/details/86306409
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lpZ3VhbmdfODIw_size_16_color_FFFFFF_t_70]: /images/20220329/bf0b3f52aca04d1cbcfd125399cc2113.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lpZ3VhbmdfODIw_size_16_color_FFFFFF_t_70 1]: /images/20220329/b35535a19c14445c93d1116bcc52c774.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lpZ3VhbmdfODIw_size_16_color_FFFFFF_t_70 2]: /images/20220329/414e4a1476cf49339db7ec55a48cf125.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lpZ3VhbmdfODIw_size_16_color_FFFFFF_t_70 3]: /images/20220329/1d3e73c65cdd4289a41718aed1a3d3e6.png
[20190114225412879.png]: /images/20220329/785e598336b448f4915d03b6a9bea710.png
[20190114225117582.png]: /images/20220329/ce8470cb80af448499b361a598d2b915.png
[kafka_Java_-3-]: https://blog.csdn.net/yiguang_820/article/details/86322711
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lpZ3VhbmdfODIw_size_16_color_FFFFFF_t_70 4]: /images/20220329/77f0f97a554941549ab833eda34f2c8e.png
[kafka_kafka-manager]: https://www.cnblogs.com/dadonggg/p/8205302.html
[https_blog.csdn.net_x541211190_article_details_81191608]: https://blog.csdn.net/x541211190/article/details/81191608
[https_blog.csdn.net_u013256816_article_details_71091774]: https://blog.csdn.net/u013256816/article/details/71091774