专栏首页sunsky解决KafKa数据存储与顺序一致性保证

解决KafKa数据存储与顺序一致性保证

“严格的顺序消费”有多么困难

下面就从3个方面来分析一下,对于一个消息中间件来说,”严格的顺序消费”有多么困难,或者说不可能。

发送端

发送端不能异步发送,异步发送在发送失败的情况下,就没办法保证消息顺序。

比如你连续发了1,2,3。 过了一会,返回结果1失败,2, 3成功。你把1再重新发送1遍,这个时候顺序就乱掉了。

存储端

对于存储端,要保证消息顺序,会有以下几个问题: (1)消息不能分区。也就是1个topic,只能有1个队列。在Kafka中,它叫做partition;在RocketMQ中,它叫做queue。 如果你有多个队列,那同1个topic的消息,会分散到多个分区里面,自然不能保证顺序。

(2)即使只有1个队列的情况下,会有第2个问题。该机器挂了之后,能否切换到其他机器?也就是高可用问题。

比如你当前的机器挂了,上面还有消息没有消费完。此时切换到其他机器,可用性保证了。但消息顺序就乱掉了。

要想保证,一方面要同步复制,不能异步复制;另1方面得保证,切机器之前,挂掉的机器上面,所有消息必须消费完了,不能有残留。很明显,这个很难!!!

接收端

对于接收端,不能并行消费,也即不能开多线程或者多个客户端消费同1个队列。

总结

从上面的分析可以看出,要保证消息的严格有序,有多么困难!

发送端和接收端的问题,还好解决一点,限制异步发送,限制并行消费。但对于存储端,机器挂了之后,切换的问题,就很难解决了。

你切换了,可能消息就会乱;你不切换,那就暂时不可用。这2者之间,就需要权衡了。

业务需要全局有序吗?

通过上面分析可以看出,要保证一个topic内部,消息严格的有序,是很困难的,或者说条件是很苛刻的。

那怎么办呢?我们一定要使出所有力气、用尽所有办法,来保证消息的严格有序吗?

这里就需要从另外一个角度去考虑这个问题:业务角度。正如在下面这篇博客中所说的: http://www.jianshu.com/p/453c6e7ff81c

实际情况中: (1)不关注顺序的业务大量存在; (2) 队列无序不代表消息无序。

第(2)条的意思是说:我们不保证队列的全局有序,但可以保证消息的局部有序。

举个例子:保证来自同1个order id的消息,是有序的!

下面就看一下在Kafka和RocketMQ中,分别是如何对待这个问题的:

Kafka中:发送1条消息的时候,可以指定(topic, partition, key) 3个参数。partiton和key是可选的。

如果你指定了partition,那就是所有消息发往同1个partition,就是有序的。并且在消费端,Kafka保证,1个partition只能被1个consumer消费。

或者你指定key(比如order id),具有同1个key的所有消息,会发往同1个partition。也是有序的。

RocketMQ: RocketMQ在Kafka的基础上,把这个限制更放宽了一步。只指定(topic, key),不指定具体发往哪个队列。也就是说,它更加不希望业务方,非要去要一个全局的严格有序。

Apache Kafka官方保证了partition内部的数据有效性(追加写、offset读);为了提高Topic的并发吞吐能力,可以提高Topic的partition数,并通过设置partition的replica来保证数据高可靠;

但是在多个Partition时,不能保证Topic级别的数据有序性。

因此,如果你们就像死磕kafka,但是对数据有序性有严格要求,那我建议:

  1. 创建Topic只指定1个partition,这样的坏处就是磨灭了kafka最优秀的特性。

所以可以思考下是不是技术选型有问题, kafka本身适合与流式大数据量,要求高吞吐,对数据有序性要求不严格的场景。

   2. 在Producer往Kafka插入数据时,控制同一Key分发到同一Partition,并且设置参数max.in.flight.requests.per.connection=1,也即同一个链接只能发送一条消息,如此便可严格保证Kafka消息的顺序

   3. 通过key, 一般会hash(某一属性)为key,来做若干个分组,这样只需在分组内严格有序即可,不牺牲并发性能。

再谈谈数据一致性保证:

一致性定义:若某条消息对client可见,那么即使Leader挂了,在新Leader上数据依然可以被读到 HW-HighWaterMark: client可以从Leader读到的最大msg offset,即对外可见的最大offset, HW=max(replica.offset) 对于Leader新收到的msg,client不能立刻消费,Leader会等待该消息被所有ISR中的replica同步后,更新HW,此时该消息才能被client消费,这样就保证了如果Leader fail,该消息仍然可以从新选举的Leader中获取。 对于来自内部Broker的读取请求,没有HW的限制。同时,Follower也会维护一份自己的HW,Folloer.HW = min(Leader.HW, Follower.offset)


数据存储 Topic 一类消息称为一个Topic

Topic逻辑结构 Topic可分为多个Parition; Parition内部保证数据的有序,按照消息写入顺序给每个消息赋予一个递增的offset; 为保证数据的安全性,每个Partition有多个Replica

多Parition的优点 并发读写,加快读写速度 多Partition分布式存储,利于集群数据的均衡 加快数据恢复的速率:当某台机器挂了,每个Topic仅需恢复一部分的数据,多机器并发

缺点 Partition间Msg无序,若想保证Msg写入与读取的序不变,只能申请一个Partition

Partition

Partition存储结构 每个Partition分为多个Segment 每个Segment包含两个文件:log文件和index文件,分别命名为start_offset.log和start_offset.index log文件包含具体的msg数据,每条msg会有一个递增的offset Index文件是对log文件的索引:每隔一定大小的块,索引msg在该segment中的相对offset和在log文件中的位置偏移量

根据offset查找msg的过程 根据msg的offset和log文件名中的start_offset,找到最后一个不大于msgoffset的segment,即为msg所在的segment; 根据对应segment的index文件,进一步查找msg在log文件中的偏移量 从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,并重建索引 优点 以segment为单位管理Partition数据,方便数据生命周期的管理,删除过期数据简单 在程序崩溃重启时,加快recovery速度,只需恢复未完全flush到磁盘的segment 通过命名中offset信息和index文件,大大加快msg查找时间,并且通过分多个Segment,每个index文件很小,查找速度更快

数据的同步

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

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

数据可靠性保证 当Producer向Leader发送数据时,可以通过acks参数设置数据可靠性的级别 0: 不论写入是否成功,server不需要给Producer发送Response,如果发生异常,server会终止连接,触发Producer更新meta数据; 1: Leader写入成功后即发送Response,此种情况如果Leader fail,会丢失数据 -1: 等待所有ISR接收到消息后再给Producer发送Response,这是最强保证

仅设置acks=-1也不能保证数据不丢失,当Isr列表中只有Leader时,同样有可能造成数据丢失。要保证数据不丢除了设置acks=-1, 还要保证ISR的大小大于等于2,具体参数设置: request.required.acks:设置为-1 等待所有ISR列表中的Replica接收到消息后采算写成功; min.insync.replicas: 设置为大于等于2,保证ISR中至少有两个Replica

Producer要在吞吐率和数据可靠性之间做一个权衡 数据一致性保证 一致性定义:若某条消息对client可见,那么即使Leader挂了,在新Leader上数据依然可以被读到 HW-HighWaterMark: client可以从Leader读到的最大msg offset,即对外可见的最大offset, HW=max(replica.offset) 对于Leader新收到的msg,client不能立刻消费,Leader会等待该消息被所有ISR中的replica同步后,更新HW,此时该消息才能被client消费,这样就保证了如果Leader fail,该消息仍然可以从新选举的Leader中获取。

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

与Kafka有几点明显不同: 数据分块,比如以64M为一个数据块; 流水线复制:每个数据块没有Leader和Follower之分,采用流水线的方式进行数据复制; 就近读取:为了减少读取时的网路IO,采用就近读取,加快读取速率

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • RPC详解

    RPC(Remote Procedure Call),即远程过程调用,是一个分布式系统间通信的必备技术,本文体系性地介绍了 RPC 包含的核心概念和技术,希望读...

    sunsky
  • 2017年的golang、python、php、c++、c、java、Nodejs性能对比[续]

      这个我用pypy 2.7确认了下,确实没那么差, 如果用numpy或其他版本python的话,性能更快。但pypy还不完善,pypy3在beta, 所以一...

    sunsky
  • 彻底搞懂etcd raft选举、数据同步

    etcd 是一个分布式的k/V存储系统。核心使用了RAFT分布式一致性协议。一致性这个概念,它是指多个服务器在状态达成一致,但是在一个分布式系统中,因为各种意外...

    sunsky
  • Curator之Master/Leader选举注意事项

    上篇博客《Zookeeper开源客户端Curator之Master/Leader选举》介绍了Leader选举的使用方法。这边博客主要说明一下在分布式定时任务选举...

    程序新视界
  • 分布式一致性协议 - ZAB

    学习ZAB,非常有必要聊聊它诞生的背景。因为在paxos的光芒下,还有必要折腾这样类似的算法吗?这个问题是我们初步了解ZAB关键。

    并发笔记
  • 大数据入门:ZooKeeper工作原理

    在大数据生态当中,分布式集群当中的一个重要组件,就是Zookeeper,作为集群运行的重要管理者,正如其名字“动物园管理员”所示,负责集群运行的诸多事宜。今天的...

    成都加米谷大数据
  • ZAB协议

    ZAB协议是为分布式协调服务Zookeeper专门设计的一种支持崩溃恢复的原子广播协议。在Zookeeper中,主要依赖ZAB协议来实现分布式数据一致性。

    shysh95
  • python 二进制文件的读写

    file.seek()方法标准格式是:seek(offset,whence=0) offset:开始的偏移量,也就是代表需要移动偏移的字节数 whenc...

    py3study
  • CCAI 2017 | 蚂蚁金服人工智能部技术总监李小龙:蚂蚁金服智能金融实践

    用户1737318
  • hbase 学习(十三)集群间备份原理

      集群建备份,它是master/slaves结构式的备份,由master推送,这样更容易跟踪现在备份到哪里了,况且region server是都有自己的WAL...

    岑玉海

扫码关注云+社区

领取腾讯云代金券