kafka基本结构和特性

怼烎@ 2023-06-29 13:58 24阅读 0赞

## 基本结构 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2MDQyOTM4_size_16_color_FFFFFF_t_70][]

为了使得kafka的吞吐率可以线性提高，物理上把一个topic拆分为一个或多个分区partition，每个分区partition在物理上对应一个文件夹，该文件夹下存储这个partition的所有消息和索引文件

每个日志文件都是一个`log entry`序列，消息体就保存在序列中，每条消息都有一个当前Partition下唯一的64字节的offset，它指明了这条消息的起始位置，这个`log entry`并非由一个文件构成，而是分成多个segment，每个segment以该segment第一条消息的offset命名并以“.kafka”为后缀

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2MDQyOTM4_size_16_color_FFFFFF_t_70 1][]

因为每条消息都被append到该Partition中，属于顺序写磁盘，因此效率非常高（经验证，顺序写磁盘效率比随机写内存还要高，这是Kafka高吞吐率的一个很重要的保证）

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2MDQyOTM4_size_16_color_FFFFFF_t_70 2][]

对于传统的message queue而言，一般会删除已经被消费的消息，而Kafka集群会保留所有的消息，无论其被消费与否。当然，因为磁盘限制，不可能永久保留所有数据（实际上也没必要），因此Kafka提供两种策略删除旧数据。

## 消息传递机制 ##

对于JMS实现,消息传输担保非常直接:有且只有一次(exactly once).在kafka中稍有不同:

1) at most once: 最多一次,这个和JMS中"非持久化"消息类似.发送一次,无论成败,将不会重发.

2) at least once: 消息至少发送一次,如果消息未能接受成功,可能会重发,直到接收成功.

3) exactly once: 消息只会发送一次.

*      at most once处理顺序: ①消费者fetch消息,②然后保存offset,③然后处理消息;

当client保存offset之后,但是在消息处理过程中出现了异常,导致部分消息未能继续处理.那么此后"未处理"的消息将不能被fetch到,这就是"at most once".

*      at least once处理顺序: ①消费者fetch消息,②然后处理消息,③然后保存offset.

如果消息处理成功之后,但是在保存offset阶段zookeeper异常导致保存操作未能执行成功,这就导致接下来再次fetch时可能获得上次已经处理过的消息,这就是"at least once"，原因offset没有及时的提交给zookeeper，zookeeper恢复正常还是之前offset状态.

exactly once: kafka中并没有严格的去实现(基于2阶段提交,事务),我们认为这种策略在kafka中是没有必要的.

通常情况下"at-least-once"是我们首选.(相比at most once而言,重复接收数据总比丢失数据要好).

## 生产上实际常用的配置 ##

消费端关闭自动提交，配置ackMode = RECORD即每处理一条commit一次 ，并做好业务上的幂等。

发送方请设置autoflush=false，acks=1

若消费逻辑比较耗时，则可减少每次最大拉取消息的数量，调低max.poll.records的值，例如max.poll.records=10，改值默认需要300秒内单线程处理完成，请根据业务耗时自行设置合理的值**。****若存在调用外部接口等不确定因素，请起线程池异步处理。**

## kafka的数据丢失原因 ##

*  如果auto.commit.enable=true，当consumer fetch了一些数据但还没有完全处理掉的时候，刚好到commit interval出发了提交offset操作，接着consumer crash掉了。这时已经fetch的数据还没有处理完成但已经被commit掉，因此没有机会再次被处理，数据丢失
 *  网络负载很高或者磁盘很忙写入失败的情况下，没有自动重试重发消息。没有做限速处理，超出了网络带宽限速
 *  如果磁盘坏了，会丢失已经落盘的数据
 *  单批数据的长度超过限制会丢失数据，报kafka.common.MessageSizeTooLargeException异常

一：生产者和broker阶段消息丢失

Kafka 消息发送分同步 (sync)、异步 (async) 两种方式，默认使用同步方式，可通过 producer.type 属性进行配置；

通过 request.required.acks 属性进行配置：值可设为 0, 1, -1(all)    -1 和 all 等同

0 代表：不等待 broker 的 ack，这一操作提供了一个最低的延迟，broker 一接收到还没有写入磁盘就已经返回，当 broker 故障时有可能丢失数据，如果加了重试会好些；

1 代表：producer 等待 broker 的 ack，partition 的 leader 落盘成功后返回 ack，如果在 follower 同步成功之前 leader 故障，那么将会丢失数据；

\-1 代表：producer 等待 broker 的 ack，partition 的 leader 和 follower 全部落盘成功后才返回 ack，数据一般不会丢失，延迟时间长但是可靠性高；但是这样也不能保证数据不丢失，比如当 ISR 中只有 leader 时( ISR 中的成员由于某些情况会增加也会减少，最少就只剩一个 leader)，这样就变成了 acks = 1 的情况；

三种机制，性能依次递减 (producer吞吐量降低)，数据健壮性则依次递增

二：生产者和broker阶段消息丢失解决方案

①producer.type = sync（同步）, Request.required.acks =  -1（all），unclean.leader.election.enable : false，对kafka进行限速， 其次启用重试机制，重试间隔时间设置长一些

producer发送消息完，等待follower同步完再返回，如果异常则重试。这是副本的数量可能影响吞吐量，最大不超过5个，一般三个足够了。不允许选举ISR以外的副本作为leader

②配置min.insync.replicas>1

当producer将acks设置为“all”(或“-1”)时，min.insync。副本指定必须确认写操作成功的最小副本数量。如果不能满足这个最小值，则生产者将引发一个异常(要么是NotEnoughReplicas，要么是NotEnoughReplicasAfterAppend)。

当一起使用时，min.insync.replicas和ack允许执行更大的持久性保证。一个典型的场景是创建一个复制因子为3的主题，设置min.insync复制到2个，用“all”配置发送。将确保如果大多数副本没有收到写操作，则生产者将引发异常

③失败的offset单独处理

producer发送消息，会自动重试，遇到不可恢复异常会抛出，这时可以捕获异常记录到数据库或缓存，进行单独处理

三：消费者消息丢失

另外一个就是使用高级消费者存在数据丢失的隐患: 消费者读取完成，高级消费者 API 的 offset 已经提交，但是还没有处理完成Spark Streaming 挂掉，此时 offset 已经更新，无法再消费之前丢失的数据. 解决办法使用低级消费者

## 数据重复原因 ##

*  生产者阶段重复场景

根本原因

生产者的消息没有收到正确的broker响应，导致producer重试，producer发出一条消息，broke落盘以后因为网络等种种原因发送端得到一个发送失败的响应或者网络中断，然后producer收到一个可恢复的Exception重试消息导致消息重复。

场景

acks = -1 的情况下，数据发送到 leader 后 ，部分 ISR 的副本同步，leader 此时挂掉。比如 follower1 和 follower2 都有可能变成新的 leader, producer 端会得到返回异常，producer 端会重新发送数据，数据可能会重复

重试过程

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2MDQyOTM4_size_16_color_FFFFFF_t_70 3][]

1. new KafkaProducer()后创建一个后台线程KafkaThread扫描RecordAccumulator中是否有消息；  
2. 调用KafkaProducer.send()发送消息，实际上只是把消息保存到RecordAccumulator中；  
3. 后台线程KafkaThread扫描到RecordAccumulator中有消息后，将消息发送到kafka集群；  
4. 如果发送成功，那么返回成功；  
5. 如果发送失败，那么判断是否允许重试。如果不允许重试，那么返回失败的结果；如果允许重试，把消息再保存到RecordAccumulator中，等待后台线程KafkaThread扫描再次发送

*  生产者发送重复解决方案

①启动kafka的幂等性

要启动kafka的幂等性，无需修改代码，默认为关闭，需要修改配置文件:enable.idempotence=true 同时要求 ack=all 且 retries>1，密等原理类似oracle数据库的sequence主键

②ack=0，不重试

可能会丢消息，适用于吞吐量指标重要性高于数据丢失，例如：日志收集

*  消费者重复消费

根本原因

数据消费完没有及时提交offset到broke.

场景

在高阶消费者中，offset 采用自动提交的方式, 自动提交时，假设 1s 提交一次 offset 的更新，设当前 offset = 10，当消费者消费了 0.5s 的数据，offset 移动了 15，由于提交间隔为 1s，因此这一 offset 的更新并不会被提交，这时候我们写的消费者挂掉，重启后，消费者会去 ZooKeeper 上获取读取位置，获取到的 offset 仍为10，它就会重复消费. 解决办法使用低级消费者

*  消费者重复解决方案

消息可以使用唯一id标识（密等）  
生产者（ack=all 代表至少成功发送一次)  
消费者 （offset手动提交，业务逻辑成功处理后，提交offset）  
落表（主键或者唯一索引的方式，避免重复数据）  
业务逻辑处理（选择唯一主键存储到Redis或者mongdb中，先查询是否存在，若存在则不处理；若不存在，先插入Redis或Mongdb,再进行业务逻辑处理）

## kafka的特性 ##

Kafka是Apache下的一个子项目，是一个高性能跨语言分布式发布/订阅消息队列系统，而Jafka是在Kafka之上孵化而来的，即Kafka的一个升级版。

具有以下特性：

1、快速持久化，可以在O(1)的系统开销下进行消息持久化；

2、高吞吐，在一台普通的服务器上既可以达到10W/s的吞吐速率；

3、完全的分布式系统，Broker、Producer（生产）、Consumer（消费）都原生自动支持分布式，自动实现负载均衡，同时保证每个Partition内的消息顺序传输；

4、同时支持离线数据处理和实时数据处理

5、Scale out：支持在线水平扩展，一般broker数量越多，集群吞吐率越高

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2MDQyOTM4_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200113101449348.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2MDQyOTM4,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2MDQyOTM4_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200113103524694.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2MDQyOTM4,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2MDQyOTM4_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20200113102525260.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2MDQyOTM4,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2MDQyOTM4_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20200114160554996.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2MDQyOTM4,size_16,color_FFFFFF,t_70