KafKa数据存储与数据一致性保证

你的名字 2022-07-15 03:27 180阅读 0赞

http://yonghuiyang.github.io/2015/12/04/kafka\_data\_store/

## 数据存储 ##

### Topic ###

一类消息称为一个Topic  
[![Kafka\_store][Kafka_store]][Kafka_store]

#### Topic逻辑结构 ####

1.  Topic可分为多个Parition;
2.  Parition内部保证数据的有序，按照消息写入顺序给每个消息赋予一个递增的offset;
3.  为保证数据的安全性，每个Partition有多个Replica

#### 多Parition的优点 ####

1.  并发读写，加快读写速度
2.  多Partition分布式存储，利于集群数据的均衡
3.  加快数据恢复的速率：当某台机器挂了，每个Topic仅需恢复一部分的数据，多机器并发

#### 缺点 ####

1.  Partition间Msg无序，若想保证Msg写入与读取的序不变，只能申请一个Partition

### Partition ###

[![Kafka\_partition][Kafka_partition]][Kafka_partition]

#### Partition存储结构 ####

1.  每个Partition分为多个Segment
2.  每个Segment包含两个文件:log文件和index文件，分别命名为start\_offset.log和start\_offset.index
3.  log文件包含具体的msg数据，每条msg会有一个递增的offset
4.  Index文件是对log文件的索引：每隔一定大小的块，索引msg在该segment中的相对offset和在log文件中的位置偏移量

[![log\_index][log_index]][log_index]

#### 根据offset查找msg的过程 ####

1.  根据msg的offset和log文件名中的start\_offset,找到最后一个不大于msgoffset的segment，即为msg所在的segment；
2.  根据对应segment的index文件，进一步查找msg在log文件中的偏移量
3.  从log文件的偏移量开始读取解析msg，比较msgoffset，找到所要读取的msg

#### Partition recovery过程 ####

每个Partition会在磁盘记录一个RecoveryPoint, 记录已经flush到磁盘的最大offset。当broker fail 重启时，会进行loadLogs。首先会读取该Partition的RecoveryPoint，找到包含RecoveryPoint的segment及以后的segment, 这些segment就是可能没有完全flush到磁盘segments。然后调用segment的recover，重新读取各个segment的msg，并重建索引

#### 优点 ####

1.  以segment为单位管理Partition数据，方便数据生命周期的管理，删除过期数据简单
2.  在程序崩溃重启时，加快recovery速度，只需恢复未完全flush到磁盘的segment
3.  通过命名中offset信息和index文件，大大加快msg查找时间，并且通过分多个Segment，每个index文件很小，查找速度更快

## 数据的同步 ##

[![Kafka\_replica][Kafka_replica]][Kafka_replica]

### 数据流 ###

1.  Partition的多个replica中一个为Leader，其余为follower
2.  Producer只与Leader交互，把数据写入到Leader中
3.  Followers从Leader中拉取数据进行数据同步
4.  Consumer只从Leader拉取数据

> ISR:所有不落后的replica集合, 不落后有两层含义：距离上次FetchRequest的时间不大于某一个值或落后的消息数不大于某一个值，Leader失败后会从ISR中选取一个Follower做Leader

### 数据可靠性保证 ###

当Producer向Leader发送数据时，可以通过acks参数设置数据可靠性的级别

1.  0: 不论写入是否成功，server不需要给Producer发送Response，如果发生异常，server会终止连接，触发Producer更新meta数据；
2.  1: Leader写入成功后即发送Response，此种情况如果Leader fail，会丢失数据
3.  \-1: 等待所有ISR接收到消息后再给Producer发送Response，这是最强保证

仅设置acks=-1也不能保证数据不丢失，当Isr列表中只有Leader时，同样有可能造成数据丢失。要保证数据不丢除了设置acks=-1, 还要保证ISR的大小大于等于2，具体参数设置：

1.  request.required.acks：设置为-1 等待所有ISR列表中的Replica接收到消息后采算写成功；
2.  min.insync.replicas: 设置为大于等于2，保证ISR中至少有两个Replica

Producer要在吞吐率和数据可靠性之间做一个权衡

### 数据一致性保证 ###

一致性定义：若某条消息对client可见，那么即使Leader挂了，在新Leader上数据依然可以被读到

1.  HW-HighWaterMark: client可以从Leader读到的最大msg offset，即对外可见的最大offset， HW=max(replica.offset)
2.  对于Leader新收到的msg，client不能立刻消费，Leader会等待该消息被所有ISR中的replica同步后，更新HW，此时该消息才能被client消费，这样就保证了如果Leader fail，该消息仍然可以从新选举的Leader中获取。

对于来自内部Broker的读取请求，没有HW的限制。同时，Follower也会维护一份自己的HW，Folloer.HW = min(Leader.HW, Follower.offset)

### HDFS数据组织 ###

[![Kafka\_replica][Kafka_replica 1]][Kafka_replica 1]  
与Kafka有几点明显不同：

1.  数据分块，比如以64M为一个数据块；
2.  流水线复制：每个数据块没有Leader和Follower之分，采用流水线的方式进行数据复制；
3.  就近读取：为了减少读取时的网路IO，采用就近读取，加快读取速率

[Kafka_store]: http://yonghuiyang.github.io/images/kafka_store.png
[Kafka_partition]: http://yonghuiyang.github.io/images/kafka_partition.png
[log_index]: http://yonghuiyang.github.io/images/kafka_log_index.png
[Kafka_replica]: http://yonghuiyang.github.io/images/kafka_replica.png
[Kafka_replica 1]: http://yonghuiyang.github.io/images/hadoop_data_flow.png