Kafka学习（2）- 架构与细节_kafka创建主题时part数量和副本数量有什么限制

作者：盐析白兔 | 2024-08-21 02:37:00

踩

kafka创建主题时part数量和副本数量有什么限制

说明：windows下操作

Kafka 中文官方文档：http://kafka.apachecn.org/documentation.html#gettingStarted

目录：

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

内容：

集群架构

说明：

S.No	组件和说明
1	Broker（代理） Kafka集群通常由多个代理组成以保持负载平衡。Kafka代理是无状态的，所以他们使用ZooKeeper来维护它们的集群状态。一个Kafka代理实例可以每秒处理数十万次读取和写入，每个Broker可以处理TB的消息，而没有性能影响。 Kafka经纪人领导选举可以由ZooKeeper完成。
2	ZooKeeper ZooKeeper用于管理和协调Kafka代理。 ZooKeeper服务主要用于通知生产者和消费者Kafka系统中存在任何新代理或Kafka系统中代理失败。根据Zookeeper接收到关于代理的存在或失败的通知，然后产品和消费者采取决定并开始与某些其他代理协调他们的任务。
3	Producers（生产者）生产者将数据推送给经纪人。当新代理启动时，所有生产者搜索它并自动向该新代理发送消息。 Kafka生产者不等待来自代理的确认，并且发送消息的速度与代理可以处理的一样快。
4	Consumers（消费者）因为Kafka代理是无状态的，这意味着消费者必须通过使用分区偏移来维护已经消耗了多少消息。如果消费者确认特定的消息偏移，则意味着消费者已经消费了所有先前的消息。消费者向代理发出异步拉取请求，以具有准备好消耗的字节缓冲区。消费者可以简单地通过提供偏移值来快退或跳到分区中的任何点。消费者偏移值由ZooKeeper通知。

关于broker
1. 一个borker服务下，是否可以创建多个分区？
  可以，broker数与分区数没有关系。
2. 一个broker服务下，是否可以创建多个副本因子
  不可以，副本因子应该小于broker数，会报错 replication factor:3 larger than available brokers

关于主题命令

创建主题时，副本因子应该小于等于可用的broker数。

创建、删除、修改、查看主题结构、查看是否有某主题


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

修改配置项
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
生产者
1. 创建生产者：kafka-console-producer.bat --broker-list localhost:9092 --topic test
消费者
1. 创建消费者：kafka-console-consumer.bat --zookeeper localhost:2181 --topic test

关于分区
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--偏移量）
  偏移量：记录当前有多少记录数，而且让消费者可以知道自己消费到什么地方。
  可以让消费者自定义选择消费某一条消息（对于同一条消息，消费者可以消费多次）
  确定一条唯一的数据：主题+分区+偏移量
副本因子（副本因子只能增，不能减少）
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
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将以非常简单和高效的方式提供两个系统中最好的。
kafka集群搭建：kafka伪分布式搭建
1. 准备好zookeeper集群
2. 修改server.properties
  各台主机的broker.id设置成不同
  zk使用自己的集群 zookeeper.connect=
  对log.dirs中进行修改
kafka Java开发：官方文档查看：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说明及优化
复制原理
当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中。
2. HW：HighWatermark（最高水平线）。因为用户发送一条消息过来之后，先会进入到Leader中，追加到log文件的末尾，但是follower与leader之间进行同步的时候，是有一定的时间差，有可能存在正在同步的路上时，leader挂掉了，3就不会同步成功。这时broker的处理方式是，将follow转为leader,所以说用户在消费的时候，只能消费到hw之前的数据，即HW为多个follower与leader之前已经同步的数据线。
  LEO：
3. timeout：生产者向kafka发送消息之后，如果在指定timeout时间内，没有给出反馈，将认为发送失败，对于生产者来说，如果已经认为发送失败，就会采取处理办法【重新发送】
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之前截断数据，由于复制是异步的，因此不能保证提交消息能够承受任何代理失败。
数据丢失与重复的解决方式
1. 生产方发送消息到kafka造成数据丢失
  1. 数据丢失和重复的情况说明
    1. 数据丢失：生产方向kafka写数据的时候，突然broker连接不上，会导致数据丢失，可以采用callback（回调函数）的方式进行处理（判断异常信息是否为空，如果为空表示正常发送了，否则就有异常，可进行特殊处理（解决方案））
  2. 数据丢失的解决方案：
    1. 单机：出现异常怎么处理？（只有单机的情况才会出现broker连接不上，broker崩溃才会连接不上，在集群模式下，会选举新的leader进行连接）
      1. 方法一：拿到没有发送成功的数据，等broker恢复之后再次发送
      2. 方法二：将消费的数据写入到mysql，如果消费成功就反馈一个结果，status的值为Y，异常就为N，再做其他的处理
      3. 方法三：出现异常就将正确的数据写入到redis中，开启一个消费再次消费，可以设置比较长的时间
      4. 可以设置重试次数：参数retries=3,只能解决进入错误处理逻辑次数减少。
2. kafka集群内部之间造成数据丢失
  集群：kafka集群内部之间有ack确认机制，防止数据丢失和重复（复制原理中有对timeout 的说明）
  1. 在timeout内，所有的ISR列表中的副本都需要确认--all
    注意：all有一个补充说明：可以设置参数min.insync.replicas（最小副本同步数）来保证最小个数的副本实现同步就可以，而不是所有的副本
  2. 在timeout内，只需要leader确认--1
  3. 在timeout内，不需要任何的副本进行确认--0
    设置实例：
3. 消费者从kafka拉数据造成的数据丢失
  1. 数据丢失和重复的情况说明
    1. 数据丢失：如果消费者在处理数据的时候出现异常，采用自动提交的话，offset的值已经发生改变。但该条消息没有被真正的处理，这样就造成数据丢失。
    2. 数据重复：以下两种情况
      1. 当消费者成功消费从kafka中拉来的数据后，向kafka提交的过程中出现了错误，导致offset值没有发生改变，当错误解决之后，消费者再次从kafka获得数据时，是已经消费的数据，导致数据重复消费
      2. 消费端消费能力比较低,处理消息速度慢。消息处理完之后提交下一个消费的offset,而在session-time-out前,消息还没有处理完,但是已经超时被kafka视为消费失败了,导致一直重复消费
  2. 解决方案：
    1. 数据丢失的解决方案：
      1. 手动提交代替自动提交，可以解决数据丢失，kafka使用Java开发-3-手动提交
    2. 数据重复的解决方案
      1. 手动提交或者延长session-time-out（消费者处理数据，给kafka反馈的时间）
      2. 开发者：消息可以使用唯一id标识作为key ，消费者处理前先根据key在redis中查询是否有此条消息，如果没有就直接处理，处理完成后将结果存储到redis中，如果有就手动提交给kafka，表明数据处理完成。
kafka消息传输保障
kakf确保消息在produce和consumer之间传输，有以下三种可能的传输保障
1. At most once：消息可能会丢，但绝不会重复传输
2. At least once：消息绝不会丢，但可能会重复传输
3. Excetly once：每条消息肯定会被传输一次且传输一次
kafka可视化工具
1. Kafka Manager
  1. 雅虎开源的Kafka集群管理工具
    管理几个不同的集群
    监控集群的状态（topics、broker、副本分区、分区分区）
    产生分区分配（Generate partition assignments）基于集群的当前状态
    重新分配分区
  2. kafka集群管理工具kafka-manager部署安装
  3. https://blog.csdn.net/x541211190/article/details/81191608
2. KafkaOffsetMonitor
  kafkaOffsetMonitor可以实时监控：
  1. kafka集群状态
    Topic、Consumer Group列表
    图形化展示topic和consumer之间的关系
    图形化展示consumer的offset、lag等信息
另外推荐一篇对kafka深入解读的博客：kafka数据可靠性深度解读https://blog.csdn.net/u013256816/article/details/71091774

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/盐析白兔/article/detail/1009609