kafka3.x原理详解看这篇就够了

　　　　　　　　1）kafka传统的定义：kafka是一个分布式的基于发布/订阅模式的消息队列，主要用于大数据实时处理领域

　　　　　　　　2）kafka最新的定义：kafka是一个开源的分布式事件流平台（event stream platform），主要用高性能数据管道，流分析，数据集成和关键任务等领域

　　　　　　　　目前市面上大部分公司采用的消息队列主要有kafka，activeMQ，rabbitMQ，rocketMQ等。kafka作为消息队列，主要应用于大数据场景下，而在Javaee开发中更多采用的是activeMQ，rabbitMQ,rockectMQ等。

　　　　　　　　传统的消息队列的主要应用场景包括：缓冲/削峰，解耦和异步通信

　　　　　　　　缓冲/削峰：在实际的应用系统中，如果数据生产端（比如其前端）的数据产生的速率与数据处理端（服务端）的数据处理速率相当或小于时，整合系统运行就不会有很大的压力。但是当系统上了个秒杀活动或者双11活动到来，前端用户猛增，数据率随之也会增加数倍，甚至是数十倍。但是服务端需要对数据进行处理，持久化等操作，处理速率必然跟不上数据产生的速度，久而久之系统就会产生数据积压，最终就有可能导致系统的崩溃。在数据生产端和处理端之间使用消息队列就可以解决这种问题。此时，消息队列就发挥了不同系统之间数据的缓冲和削峰的作用。数据生产端将数据发送到消息队列，然后随即返回响应，这个过程相对来说是非常快的。数据处理端则根据自己的处理速度从消息队列中拉取数据。示意图如下：

 不使用消息队列的情况

 使用消息队列的情况

　　　　　　　　解耦：允许独立的拓展和修改两边的处理过程，但两边需要确保使用相同的接口约束。

　　　　　　　　异步通信：将处理的用户数据写入到消息队列中，并立即返回处理结果，队列中数据由另一个线程拉取出来做响应的处理。下面是用户注册，并把注册成功的消息发送到用户手机上的同步处理和异步处理的流程。

　　　　　　　　1）点对点模式

　　　　　　　　2）发布/订阅模式

 　　　   1）producer：消息生产者，就是向broker发送消息的客户端

　　　　2）consumer：消息消费者，就是从broker拉取数据的客户端

　　　　3）consumer group：消费者组，由多个消费者consumer组成。消费者组内每个消费者负责消费不同的分区，一个分区只能由同一个消费者组内的一个消费者消费；消费者组之间相互独立，互不影响。所有的消费者都属于某个消费者组，即消费者组是一个逻辑上的订阅者。

　　　　4）broker：一台服务器就是一个broker，一个集群由多个broker组成，一个broker可以有多个topic。

　　　　5）topic：可以理解为一个队列，所有的生产者和消费者都是面向topic的。

　　　　6）partition：分区，kafka中的topic为了提高拓展性和实现高可用而将它分布到不同的broker中，一个topic可以分为多个partition，每个partition都是有序的，即消息发送到队列的顺序跟消费时拉取到的顺序是一致的。

　　　　7）replication：副本。一个topic对应的分区partition可以有多个副本，多个副本中只有一个为leader，其余的为follower。为了保证数据的高可用性，leader和follower会尽量均匀的分布在各个broker中，避免了leader所在的服务器宕机而导致topic不可用的问题。

　　　　8）leader：多个副本的主副本，生产者发送的数据和消费者消费的数据都是通过leader进行处理的。

　　　　9）follower：多个副本中除了leader副本，其余的均为follower副本，也即从副本。从副本不会和生产者和消费者提供服务，而是实时同步主副本的数据。当主副本宕机后，通过一定算法选举出新的从副本成为主副本，继续为生产者和消费者提供服务。

参数名称

bootstrap.servers 

key.serializer 和 value.serializer

buffer.memory

batch.size

linger.ms

acks

max.in.flight.requests.per.connection

允许最多没有返回 ack 的次数，默认为 5，开启幂等性要保证该值是 1-5 的数字。

retries

当消息发送出现错误的时候，系统会重发消息。retries表示重试次数。默认是 int 最大值，2147483647。如果设置了重试，还想保证消息的有序性，需要设置MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION=1否则在重试此失败消息的时候，其他的消息可能发送成功了。

retry.backoff.ms

两次重试之间的时间间隔，默认是 100ms。

enable.idempotence

 是否开启幂等性，默认 true，开启幂等性。

compression.type

 生产者发送的所有数据的压缩方式。默认是 none，也就是不压缩。支持压缩类型：none、gzip、snappy、lz4 和 zstd。

　　　　　　　　在实际的企业应用中，可能会有不用的场景，默认的分区器无法满足需求，那么就需要自定也分区器来满足需求，比如某个分区的服务器性能比较好，另一个是比较久的服务器，性能相对差一点，那么就需要通过自定义分区器让更多的数据向性能更好的分区倾斜。

　　　　　　　　自定义分区器的实现：第一，定义类实现 Partitioner 接口。第二，重写partition()方法。第三，在生产者配置中指定自定义的分区器。

　　　　　　要想更好的提高生产者的吞吐量，则必须先了解生产者发送数据的流程，具体流程可以看靠（一）的介绍。以下是提高生产者的吞吐量的建议。

修改compression.type的数据压缩类型，默认snappy。根据不同的业务场景选择不同的数据压缩方法，提高数据压缩率。kafka提供的压缩类型有：gzip,snappy,lz4,zstd。

修改RecordAccumulator缓冲区的大小。默认32m，增加缓冲区大小可以那么每个batch.size的值就更大，每次发送的数据更多。

　　　　　　数据的可靠性是指producer发送数据到kafka收到应答后，该数据都能成功落盘，那么这次发送是可靠的。但是，为了适应不用的应用场景以及实现高可用，kafka会将数据备份到不同的副本当中，在数据同步的过程中如果出现的故障，那么就有可能出现数据丢失，重复情况。想要弄清楚kafka数据可靠性，就必须先要了解kafka中ACK的应答原理。

　　　　　　　　ACK应答是指在leader分区接收到生产者的数据后，何时对生产者做出应答的策略。ACK可选的值有0，1，-1三个，可以在生产者的配置项 acks 中设置，ACK设置不同，对生产者做出应答的时机也不同。

　　　　　　　　ACK=0：可靠性级别最低。leader收到生产者数据后不需要等数据落盘，立即对生产者做出应答。生产者收到应答后认为leader已成功接收数据，因此不需要再发当前数据了。但是，如果leader在将内存中的数据落盘时突然出现故障，那么这条数据因为没有保存到磁盘中而导致数据的丢失。

　　　　　　　　ACK=1：可靠性级别较高。leader收到生产者的数据并将数据落盘后，对生产者做出应答。生产者收到应答后继续发送其他数据。如果leader做出应答并且follower未同步到该数据时，leader出现故障。kafka会重新在follower中选出新的leader，而新的leader心有同步到数据，生产者也不会再发该数据，因此导致该数据的丢失。

　　　　　　　　ACK=-1（all）：可靠性级别最高，kafka的acks默认值。leader收到数据并落盘，并且确认所有follower收到数据后再给生产者应答。此时，所有分区副本都有该数据了，即使任意分区出现故障数据仍然是完整的。

　　　　　　　　在使用kafka时为了保证数据的高可靠性，我们一般都会将应答级别设置为-1（all），即leader的ISR列表的follower均收到数据后再应答。非常不幸的是，leader在收到所有的follower的确认后发生故障，所有的分区均已保存到磁盘中，但是生产者没有收到应答，认为leader没有收到生产者发送请求，于是尝试重新发送请求。由于leader发生故障，kafka重新选举leader，生产者将数据再一次发送到新的leader上，所以造成的数据重复。

　　　　　　　　kafka 0.11之后引入了幂等性和事务两大特性。利用这两个特性可解决数据重复的问题。

　　　　　　　　　　 　　At Least Once可以保证数据不丢失，但不能保证数据重复。

　　　　　　　　　　　　At Most Once 可以保证数据不重复，但不能保证数据不丢失。

　　　　　　　　　　　　　　幂等性：是指无论producer发送多少条重复的数据，broker端都只会持久化一条数据，保证了数据不重复。

　　　　　　　　　　　　　　数据重复的判断依据：具有相同主键的消息提交时，Broker只会持久化一条。其中PID是Kafka每次重启都会分配一个新的；Partition 表示分区号；Sequence Number是单调自增的。

　　　　　　　　　　　　　　从数据重复判断依据来看，幂等性只能保证单分区会话内数据不重复。

　　　　　　　　　　　　       开启幂等性：enable.idempotence = true,默认为true。

　　　　　　　　kafka的事务需要跟幂等性配合起来使用。开启事务就必须开始幂等性。

　　　　　　kafka的producer客户端在向broker发送数据时并不是直接将数据发送出去，而是将数据先缓存到本地的双端缓存队列中，sender线程会不断地检测缓存队列中地数据，若队列中地数据达到了设置的（batch.size）值地容量或者达到一定的时间（linger.ms），sender就会创建一个InFlightRequests线程，该线程负责将数据发送到broker上。kafka 默认的InFlightRequests是5个，InFlightRequests线程发送数据后不需要broker的应答就可以发下一个数据，因此最多可以一起发5个请求。比如，需要发送0，1，2，3，4这五个数据依次被发送出去，但是2这个数据没有接收成功，客户端则重新2的数据，此时broker接收到的数据的顺序就变为了01342，而非01234。

　　　　　　　　1）kafka在1.x之前保证数据单分区有序，需要将InFlightRequests线程数设置为1个（max.in.flight.requests.per.connection=1）。当线程数为1时就能保证broker收到数据确认后再发下一条数据。

　　　　　　　　2）kafka在1.x以后在未开启幂等性的情况下，处理流程跟1.x以前的版本一样。

　　　　　　　　3）1.x以后在开启幂等性的情况下，可以将max.in.flight.requests.per.connection设置为小于等于5。原因是，Kafka服务端会缓存最近发过来的元数据，等缓存满了5个后就会对这些元数据进行排序，这样就可以保证数据有序了。

　　　　　　① broker启动，向zookeeper注册broker节点信息

　　　　　　② 向zookeeper注册controller，若注册成功则该controller负责选举leader，监听broker节点等工作

　　　　　　③ 注册成功的controller开始监听brokers节点变化

　　　　　　④ 注册成功的controller举行leader选举。按照AR列表优先选取排在前面的节点，若该节点在ISR中存活则成为leader，否则继续轮询。例如：AR[0,1,2],ISR[1,2]，controller会从broker 0 开始轮询，发现ISR中并没有0，那么继续轮询到1，发现ISR有1则1成功leader。

　　　　　　⑤ 注册成功的controller将选举结果写入到zookeeper中，同时其他controller也监听该信息，等leader故障后方便立即上位成为新的leader。

　　　　　　⑥ 加入leader挂掉了，controller监听到节点发送变化，开始拉取ISR信息，然后开始重新选择leader。

　　　　　　Kafka集群中有一个会被选举为controller leader，选举方式抢占式，谁先抢占到zookeeper的节点，谁就能成为leader。该controller负责Kafka集群broker的上下线，topic分区副本的分配和leader选举等工作。

　　　　　　LEO（log end offset）：每个副本的最后一个offset，即最新offset + 1。

　　　　　　HW（high watermark）：所有副本中最小的LEO。

　　　　　　　　leader发生故障后，kafka会从ISR中重新选举出新的leader

　　　　　　　　为保证多副本之前的数据一致性，其余的follower会先将自己高于HW的数据截掉，然后同步新leader的数据。

　　　　　　　　follower发生故障后会被临时踢出ISR。

　　　　　　　　此时leader和follower继续接收数据。

　　　　　　　　等follower恢复后读取保存在磁盘的HW，并将log文件中高于HW部分的数据截取丢弃，从HW位置开始同步leader的数据。

　　　　　　　　follower同步到LEO大于等于分区的HW时，kafka就将该副本重新加入到ISR中。

　　　　　　1）尽量将所有副本平均地分配到所有地broker上。

　　　　　　2）每个broker分配到地leader尽可能一样多。

　　　　　　3）leader和follower尽可能地分配到不同地broker上。

　　　　　　正常情况下，Kafka本身会尽可能地将leader分区均匀地分布到各个机器上，这样使得每个机器的读写吞吐量比较均匀。但是，如果当leader分区故障后，另一个follower就会迅速上位并且承担之前的leader的工作，读写请求的压力就会上升，造成集群负载不平衡。kafka允许的每个broker不平衡的比率默认为10%，如果超过了这个值就会触发leader分区平衡。

　　　　　　相关参数：

　　　　　　Kafka中topic是逻辑概念，但是分区partition是物理概念，生产者发送的数据都是存储在partition中的，每个partition中都有一个log文件，数据实际都是存储在log文件中，生产者生产的数据都会最佳到该log文件的末端。为防止log文件过大导致数据检索过慢，kafka将log文件进行切片，每个片成为segment，每个segment中包括“.index”文件，“.log”文件和“.timestamp”等文件。

　　　　　　log文件：保存实际数据的文件。

　　　　　　index文件：偏移量索引文件，文件为segment第一个数据offset值。index文件是一个稀疏索引，每往log文件中写入4kb的数据才向index文件中写入一条索引。

　　　　　　timestamp文件：时间戳索引文件，文件名为segment第一条数据的offset值。

　　　　　　Kafka提供的文件删除策略有delete和compact两种。

　　　　　　delete策略：将过期数据删除

　　　　　　　　log.cleanup.plocy = delete 所有数据启用删除策略

　　　　　　compact日志压缩：对于相同key的不同value值，只保留最后一个value值。压缩后的offset可能是不连续的。这种策略只适合某些特殊的场景，比如key保存的是用户id，value保存的用户信息，通过压缩策略就能将旧的用户数据删除，只保留新的用户信息。

　　　　　　1、Kafka是基于分布式集群，采用分区技术，数据读写并行度高。

　　　　　　2、读数据采用稀疏索引，消费者能够快速定位消费的数据。

　　　　　　3、顺序写磁盘，producer生产数据只是在log文件的末端追加数据，因此写磁盘的速度很快。

　　　　　　4、Kafka以来系统底层的pagecache技术和零拷贝技术，能够提高磁盘数据读写速度。

当 Kafka 中没有初始偏移量或当前偏移量在服务器中不存在如，数据被删除了），该如何处理？

earliest：自动重置偏移量到最早的偏移量。

latest：默认，自动重置偏移量为最新的偏移量。

none：如果消费组原来的（previous）偏移量存在，则向消费者抛异常。

anything：向消费者抛异常。

　　　　　　　　consumer主动拉取服务端的数据。Kafka采用这种方式。

　　　　　　　　优点：consumer根据自己的处理能力拉取服务端的数据。

　　　　　　　　缺点：当服务端没有数据时，consumer仍然不断从服务端拉取数据，会消耗consumer一定的资源。

　　　　　　　　服务端主动推消息给consumer。

　　　　　　　　优点：服务端只要有数据是才会推数据给consumer。

　　　　　　　　缺点：服务端发送消息的速率很难适应所有consumer处理消息的速率。

　　　　　　1）消费者通过offset拉取broker中指定位置的消息。offset则保存在系统主题中，系统主题保存在磁盘中，所以即使服务端出现故障或重启等能够按上次消费的位置开始消费。

　　　　　　2）一个消费者可以消费多个分区的数据。

　　　　　　3）每个分区的数据只能由消费者组的一个消费者消费。

　　　　　　1）消费者在消费消息时必须指定消费者组id（group.id），具有相同的group.id组成一个消费者组。

　　　　　　2）消费者组的消费者消费不同分区的数据。一个分区只能由一个消费者消费。

　　　　　　3）消费者组之间相互独立，互不影响。

　　　　　　4）消费者组的消费者数量应小于等于分区数量，若消费者数量大于分区数，多余的消费者则不会消费消息。

　　　　　　5）消费者消费消息后发送offset给服务端。

　　　　　　首先对同一个topic的分区对分区序号进行排序，然后消费者按照字母排序，利用分区数除以消费者数量得出每个消费者平均消费分区数，若剩余多的则按顺序消费。

　　　　　　注意：当topic数量比较多时，使用Range分区策略会导致排在前面的消费总是分配更多的消费者，造成数据倾斜。

　　　　　　再平衡策略：当某个消费者挂了，Kafka会将该消费者的任务全部交给正常消费的消费者。

　　　　　　先进行排序，然后消费依次轮询消费分区数据。

　　　　　　再平衡：触发再平衡后，将故障的消费者的任务轮询交给正常消费的消费者。

　　　　　　粘性分区是 Kafka 从 0.11.x 版本开始引入这种分配策略，首先会尽量均衡的放置分区到消费者上面，在出现同一消费者组内消费者出现问题的时候，会尽量保持原有分配的分区不变化。

　　　　　　再平衡：尽可能均衡的随机分成 0 和 1 号分区数据，分别由 1 号消费者或者 2 号消费者消费。

　　　　　　　Kafka0.9版本之前，consumer默认将offset保存在Zookeeper中。

　　　　　　　从0.9版本开始，consumer默认将offset保存在Kafka一个内置的topic中，该topic为__consumer_offsets。

　　　　　　分为自动提交和手动提交，默认为自动提交

　　　　　　　　enable.auto.commit：设置为true

　　　　　　　　auto.commit.interval.ms：自动提交offset的间隔时间，默认为5s

　　　　　　　　　enable.auto.commit ：设置为false。用户可以根据具体场景进行提交，更加灵活。手动提交可分为同步提交和异步提交两张方式。

　　　　　　当消费者第一次消费或者offset信息丢失，这是消费者该如何进行消费呢？Kafka提供了三种方式：earliest，latest和none

　　　　　　earliest：自动将offset设置为最开始的位置。

　　　　　　latest：自动将offset设置为最新的offset位置。

　　　　　　none：若没有找到之前的offset信息，则会抛出异常。

　　　　　　消费者根据指定时间（比如1天前）获取数据，通过broker的时间戳索引文件查到该时间对应的offset，然后用该offset获取对应的数据。

　　　　　　　先提交offset后处理数据，可能会造成漏消费消息。一般用户设置为手动异步提交offset会引起这种问题。

　　　　　　　场景：消费者offset提交方式设置为手动异步提交，消费者把offset提交出去，但是数据还未落盘，此时消费者挂掉，分区已经将最新的offset保存起来了。消费者重启后获取的offset则是最新提交的offset了，造成数据遗漏处理。

　　　　　　　　已经消费了消息，但是没提交offset。一般消费者设置为自动提交会有这种问题。

　　　　　　　　场景：消费者设置offset自动提交时间为5s，在第2s时消费者挂掉，在挂掉前已经处理部分数据，但offset没有提交到broker。消费者重启后拿到的offset则为旧的offset，之前处理过的数据又被拉取一遍，造成重复消费数据。

　　　　　　若要保证消费者消费数据时不漏消费，不重复消费，则需要将消费者端在消费消息和提交offset的操作看作是依次原子操作。

　　　　　　所谓数据积压是指broker中保存大量未消费的消息。若长时间未消费且触发了删除策略，那么这部分数据就会丢失。可能造成数据积压的原因有消费者的处理能力不足，分区数较少，可适当增加分区数量，使分数量等于消费者数量。也可以把消费者每次拉取数据的大小往上调。

　　　　Kafka凭借自身优秀的性能和海量数据的处理能力以及数据的可靠性保证，组件在大数据领域占据主流地位。