最新消息:深度思考

2.kafka的高可靠性实现方案及保障机制

kafka liuxuecheng 4559浏览 0评论

版权申明:转载请注明出处。
文章来源:大数据随笔

作为一款商用的高可靠实时消息系统,kafka使用了replication,ISR,leader选举等机制保证其可靠性。具体分析如下:

1.文件存储机制

前面提到消息在逻辑上以topic存在,而在物理存储上是partition,在实际存储中partition是作为目录存在于broker上。
假如server.properties中log.dirs和num.partitions配置如下:

log.dirs=/tmp/kafka-logs
num.partitions=4

同时创建一个名为my_test_topic的topic,则会在/tmp/kafka-logs/下面生成四个目录,分别为:

my_test_topic-0
my_test_topic-1
my_test_topic-2
my_test_topic-3

这四个目录对应就是my_test_topic的四个分区,由此也可以看到分区的命名规则为topic名称+以0开头的序号。
每个partition又包含多个segment文件,一个segment文件由.index文件和.log两个文件组成,分别为segment的索引
文件和数据文件,这两个文件的命名基于同一个topic下的所有partition,即每个partition下的所有segment文件命名
都不一样。具体的命名规则为:partition全局的第一个segment从0开始,后续每个segment文件名为上一个segment文件最后一条消息的offset值,数值大小为64位,20位数字字符长度,没有数字用0填充,如下:

00000000000000000000.index
00000000000000000000.log
00000000000000170410.index
00000000000000170410.log
00000000000000239430.index
00000000000000239430.log

index索引文件用来存储元数据,log文件用来存储消息,索引文件中的元数据指向数据文件中消息的物理偏移地址。
两者关系如下图示:
kafka-index-log
以index文件中(64 785)这组记录为例,它表示在00000000000000239430.log这个log文件中是第64个消息,
在全局为第239494个消息,即offset,它的物理偏移地址为785。consumer使用offset从partition读取消息依赖于上述关系。
举例说明:

假如要读取offset=239432的消息,使用二分查找定位到到00000000000000239430.index,然后根据偏移量定位到
消息的物理地址偏移量为10,然后读取Message239432。因为消息中记录了offset、magic、crc32、length等信息
所以可以直到在哪里结束读取。

2.消息发送机制

2.1消息发送模式

Kafka的发送模式由producer端的配置参数producer.type来设置,这个参数指定了在后台线程中消息的发送方式是同步的还是异步的,默认是同步的方式,即producer.type=sync。如果设置成异步的模式,即producer.type=async,可以是producer以batch的形式push数据,这样会极大的提高broker的性能,但是这样会增加丢失数据的风险。如果需要确保消息的可靠性,必须要将producer.type设置为sync。
对于异步模式,有如下四个参数可以设置:

参数 描述
queue.buffering.max.ms 默认值:5000。启用异步模式是,producer缓存消息的时间。比如设置成1000,
它会缓存1s的数据然后发送出去,这样可以极大的增加broker的吞吐量,但是会降低数据的失效性。
queue.buffering.max.message 默认值:10000。producer缓存队列里的最大缓存消息数,如果超过这个值,producer就会阻塞或者丢弃消息。
queue.enqueue.timeout.ms 默认值:-1。当达到上面参数时producer会阻塞等待的的时间。如果设置为0,
队列满时producer不会阻塞,消息会被直接丢弃。若设置为-1,producer会阻塞,不会丢弃消息。
batch.num.message 默认值:200。启用异步模式时,一个batch缓存消息的数量,当达到这个数值时producer才会发送。

2.2数据可靠性级别

当producer向leader发送数据时,可以通过request.required.acks参数来设置数据可靠性的级别。

参数设值 描述
1(默认) producer在数据replication的leader收到数据并且得到确认后发送下一条数据。如果leader宕机,消息丢失。
0 producer无需确认leader的回复而继续发送,这种情况下数据可靠性最低,但是传输效率最高。
-1 producer需要等待replication中所有的follower都确认收到数据后才发送下一条数据。可靠性最高,但是当只有leader而没有follower的时候和request.required.acks=1一样。

3.副本同步机制

kafka每个topic的partition有N(N>=1)个副本,其中N由replica factor这个配置项决定。kafka通过多个副本实现故障
的自动转移,当kafka集群中一个节点挂掉后可以保证服务仍然正常。多个副本中有一个副本为leader,其余为follower。
leader负责处理针对该partition的所有读写请求,follower能自动的从leader复制数据。下图是一个有4个broker、3个partition(同一个颜色深度为一个partition)、replica factor=3的kafka集群中leader和follower之间关系的示意图。
kafka replication
关于leader和follower之间的数据复制,考虑下面几个问题:
1.当leader挂掉后,会通过选举机制从follower中产生一个新的leader。加入之前的leader已有的数据条数为5,被选举
的follower只复制了其中三条,那么consumer从新的leader获取数据就会出现问题,这种情况下如何保证数据正确性?
2.若某个follower出现故障,导致复制数据出现异常,该如何保证系统可靠性?
针对上述问题,kafka引入了HW(HighWaterMark)和ISR(In Sync Replicas),下面分别介绍。

3.1 HW

和HW相关的还有LEO,LEO是LogEndOffset的缩写,标识每个partition的log最后一条message的位置。HW是指
consumer能够读到的此partition最新的位置。
kafka hw

3.2 ISR

ISR是指副本同步,副本对于kafka的吞吐率有一定的影响,但是极大的增强了可用性,默认情况下Kafka的replica数量为1,即每个partition都有一个唯一的leader,为了确保消息的可靠性,通常应用中将其值(由broker的参数offsets.topic.replication.factor指定)大小设置为大于1,比如3。所有的副本(replicas)统称为Assigned Replicas,即AR。ISR是AR中的一个子集,由leader维护ISR列表,follower从leader同步数据有一些延迟(包括延迟时间replica.lag.time.max.ms和延迟条数replica.lag.max.messages两个维度, 当前最新的版本0.10.x中只支持replica.lag.time.max.ms这个维度),任意一个超过阈值都会把follower剔除出ISR, 存入OSR(Outof-Sync Replicas)列表,新加入的follower也会先存放在OSR中,AR=ISR+OSR。

3.3基于HW和ISR的消息同步

一个partition中取ISR中最小的LEO作为HW,consumer最多只能消费到HW所在的位置。另外每个replica都有HW,leader和follower各自负责更新自己的HW的状态。对于leader新写入的消息,consumer不能立刻消费,leader会等待该消息被所有ISR中的replicas同步后更新HW,此时消息才能被consumer消费。这样就保证了如果leader所在的broker失效,该消息仍然可以从新选举的leader中获取。对于来自内部broKer的读取请求,没有HW的限制。
kafka-replica-sync
(1) follower和leader数据一致,HW和LEO也一致,无需复制。
(2)leader接收到producer写入的数据,此时LEO移动,两个follower开始复制数据。
(3)follower1完全复制了消息,follower2只复制了一部分,HW移动一位。
(4)follower2完全复制消息,HW和LEO一致。
由此可见,Kafka的复制机制既不是完全的同步复制,也不是单纯的异步复制。事实上,同步复制要求所有能工作的follower都复制完,这条消息才会被commit,这种复制方式极大的影响了吞吐率。而异步复制方式下,follower异步的从leader复制数据,数据只要被leader写入log就被认为已经commit,这种情况下如果follower都还没有复制完,落后于leader时,突然leader宕机,则会丢失数据。而Kafka的这种使用ISR的方式则很好的均衡了确保数据不丢失以及吞吐率。

3.4文件截断机制

kafka文件截断
如上图,某个topic的某partition有三个副本,分别为A、B、C。A作为leader肯定是LEO最高,B紧随其后,C机器由于配置比较低,网络比较差,故而同步最慢。这个时候A机器宕机,这时候如果B成为leader,假如没有HW,在A重新恢复之后会做同步(makeFollower)操作,在宕机时log文件之后直接做追加操作,而假如B的LEO已经达到了A的LEO,会产生数据不一致的情况,所以使用HW来避免这种情况。 A在做同步操作的时候,先将log文件截断到之前自己的HW的位置,即3,之后再从B中拉取消息进行同步。
如果失败的follower恢复过来,它首先将自己的log文件截断到上次checkpointed时刻的HW的位置,之后再从leader中同步消息。leader挂掉会重新选举,新的leader会发送“指令”让其余的follower截断至自身的HW的位置然后再拉取新的消息。

4.consumer消息保证

有三种级别的消息保证:
* At most once: 消息可能会丢,但绝不会重复传输
* At least once:消息绝不会丢,但可能会重复传输
* Exactly once:每条消息肯定会被传输一次且仅传输一次
(1)先commit offset再消费消息,如果在offset被commit之后但消息没有被消费时consumer宕机,则会丢失消息,实现at most once。
(2)先消费消息,再commit offset,如果在offset被commit之前consumer宕机,当重启时会重复消费。实现at least once。
(3)要实现exactly once则需要对commit offset和消息消费加上事务,或者对at least once加上去重机制。

转载请注明:大数据随笔 » 2.kafka的高可靠性实现方案及保障机制

与本文相关的文章

发表我的评论
取消评论

表情

Hi,您需要填写昵称和邮箱!

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址