【IT168 专稿】本文根据简怀兵老师在2018年5月12日【第九届中国数据库技术大会(DTCC)】现场演讲内容整理而成。
讲师简介：
简怀兵，YY Research Lab存储负责人。10+年后端技术、数据库内核、分布式系统设计研发&架构经验，PTimeDB作者。先后供职于腾讯、唯品会、YY等大型互联网公司。在分布式存储系统的设计研发和工程化实践方面，有较丰富经验。
摘要：
本次分享将从实战经验出发，详细介绍从0开始开发一个强一致/高可用/高性能的分布式Log存储系统的过程和挑战;在分享分布式系统设计开发的同时，会结合工程实践，阐述如何将分布式Log存储系统作为构建其它分布式系统(分布式Cache/分布式Key-value/分布式MQ/分布式数据库)的“基础”，以及如何成为数据抽象中的关键一环。
分享大纲：
·研发背景
·设计思路
·关键实现
·应用形态
正文演讲：
我今天和大家分享的内容很简单，只有四点。第一点是我们研发分布式Log系统的背景是什么，第二是遇到问题之后的设计思路，第三是在落地过程中我们主要解决了哪些问题，第四是有了统一的日志抽象以后，我们的应用形态是什么样?
研发背景
大型互联网公司的业务系统，一般都非常庞大，同时也非常复杂。为什么会复杂呢?这里面既有历史原因、设计原因，也有业务快速发展的原因，而我们希望不同子系统之间的耦合不要那么重，这是我们的诉求之一。
很多快速成长起来的系统都有一个共同点就是很乱，连接很乱，交互也很乱。所以，我们就在思考能不能有一个统一的抽象，把不同的业务系统或者说存储系统之间解耦。
分布式Cache、分布式KV、分布式MQ、分布式数据库等都是大型互联网公司技术栈中常用的组件，而这些系统中又都有比较通用的相同部分，所以我们希望把通用的部分抽象出来，使其成为所有分布式系统的“积木”，这是我们的第二个诉求。
上面这两个核心的诉求，如何去满足呢?我们思路的来源，相信在座的很多同学都曾看过上图这篇博文，博文中有句话特别关键，大意是：数据库、NoSQL存储、KV存储、replication、paxos、Hadoop、版本控制等等，这些系统都有一个很本质的东西 — log。
不要把log想得太复杂，它就是data。依据上文中的思路，我们来分析一下这两个诉求，我们有不同的业务系统和存储系统，那么这个统一的抽象是什么呢?就是log，它可以看做是数据流，MySQL中的数据写到统一的log中，不同的系统都可以去消费，例如实时订阅系统、大数据、elasticsearch等等。
第二个诉求是希望分布式日志系统成为所有其它常见分布式系统的积木，实现这个诉求的理论支撑就是RSM(Replicated State Machine)。
上图中有三个服务节点，某个服务节点在第一步收到请求之后，第二步是通过一致性协议模块，在不同的服务节点之间达成共识并将请求以log的方式持久化到存储介质中，第三步是把log回放到状态机中，最后一步回复给客户端(请求发起端)。
通过这样一个简单的RSM模型，有了统一的log抽象以后，我们有很多文章可以做。
RSM本身是一个简单抽象的概念，所以我们具体延伸一下。RSM第一种延伸应用如上图所示，共有三个节点，我们把它认为是服务节点，下面有统一的log抽象。假设客户端发起请求到master节点，master节点执行请求通常会产生log，log代表的是这次请求对存储系统状态带来的变化，存到统一log中再通过其它手段回放到不同的Slave中去。这就是典型的passive replication。
还有一种模式，当请求来了直接写到统一的log中，不同的节点直接从log回放到本地database中，这里的database指的是通用的存储系统，这种方式叫做active replication。
左边模式中的请求可能是nondeterministic的，所以必须通过master的执行将这个请求转换为deterministic的log，回放到其它slave中，来保证集群中数据的一致性。
讲完延伸应用之后，我们再来看看之前的两个诉求。
解决第一个诉求比较简单的方式就是面向log的架构设计。国外有很多人提出了这个概念，但是在国内还比较小众，如果我们所有东西都基于log，那么我们可以做很多东西。如应用到DB中，无论是MySQL binlog还是InnoDB redolog，将其存储到统一的log集群中，这个log可以用来做什么呢?第一是做异构DB的复制，有了redolog之后，就可以回放到其它DB中，例如Redis、PostgreSQL等等。做数据库备份恢复的时候，有了统一的log之后可以通过位点去追binlog。数仓、离线处理、Hadoop、实时流计算、Flink、Spark、ES或者是实时监控都可以基于log来做数据同步和消费。
第二个诉求其实前文的配图中已经有说明，唯一不同的是在上图这种方案中，我们把状态机替换成了Redis，这就快速开发出了一个基于一致性协议的分布式KV系统。所以我们总结一下，第二个诉求的关键点是我们要实现不同的状态机。
总结一下，我们最终的诉求是需要一个分布式的log存储系统，这个系统有三个最基本的要求：强一致、高可用，以及在满足前两个要求的基础上尽可能高性能。
关键实现
走到这个环节，我们已经明确了目标 —— 强一致、高可用、高性能的分布式log系统，如何实现呢?这里面有一些关键的问题。
上图是分布式log系统的抽象架构，客户端对分布式集群生产数据或消费数据，在系统设计的时候，我们做了些基本的语义抽象，例如topic、partition概念，逻辑上和kafka很像，其中每个partition都有多个副本，上图中我只画了三个副本，同一个partition的副本之间使用raft进行数据复制。如果简单来理解的话，大多数情况下都有一个稳定的leader和两个follower。
有了统一的架构抽象以后，这里面还涉及到几个问题，第一个是分布式系统当中的一致性问题，这是做存储或者分布式系统当中比较经典的问题了。我按照一致性协议的时间先后排了一下序，从最早的Viewstamped Replication到后面的Paxos，其中 Paxos又有非常多的变种，multi-Paxos，Fast-Paxos，Mencius、Epaxos，接下来是Zookeeper里的ZAB协议，最后是Raft算法。
为什么会选择Raft算法?其实有两大原因，第一是Raft算法工程化程度比较高，第二是和我们所属公司的技术栈相关，贴近公司etcd等其它组件。
之前大家都会觉得强一致就意味着性能差，其实这个看法并不严谨。比较负责任地说，在IDC里以及同城的IDC之间强一致的成本是完全可以接受的，不过跨城之间的成本确实会很高。
对于实现强一致，我有三点想和大家分享，首先强一致的理论早在40年前就已经出现了，所以我们真正要思考的是如何去提升性能?
第一个解决方法是batch，说白了就是多个请求去分摊跑一次一致性协议的开销(其中主要部分通常是网络传输环节)，我画了了一个简单的图，大家一看就懂了。注意，batch和group commit不同，group commit主要是为了分摊存储介质定位数据的开销。
第二个解决方法是Pipeline，一个请求发起之后，没必要等到这个请求完全执行完毕之后再发起下一个请求，可以同时发起第二个请求，只要保证发起的请求和得到的结果之间是匹配的。
除了上面两个解决方法，还有一个方法就是优化pipeline的流程。传统从leader到follower的复制流程是：请求到达leader后先刷盘，刷完之后再把请求发送到follower上去刷盘，刷盘以后回复给leader，leader收到之后保证在某一个三副本的系统当中，至少有两个副本是持久化的。现在很多开源系统早期都是采用这种方式，也是最常使用的一种方式。
我们简单优化了pipeline的流程，在leader刷盘之前先把数据发送出去，当副本回复到达leader的时候，leader本地已经刷完盘了，从上图来看，通过优化我们节约了一次刷盘的开销。
在做NoSQL存储系统时通常会遇到几个问题，第一是IO放大(读写放大)，第二是读写之间或不同租户之间的隔离问题。
举例来说，Kafka是一个优秀的消息系统，当所有的消费者都消费比较新的数据，那么不会有太大影响，因为数据可能都在page cache中，如果消费者落后的数据比较多，例如去消费三天之前的数据，那么这些数据可能已经从page cache刷到存储介质中，通常要从存储介质去读区。我们把读取操作分为两种：一种叫tailing read，读的是比较新的数据;第二种叫catch-up read，读的是比较久之前的数据。
如何解决这两个问题呢?写放大的解决方法是，我们认为log就是data，我们不会先写预写日志，再从预写日志回放到SST Table中。我们写日志时会建两份索引，指到这一份data当中的不同offset。
至于隔离问题，我们把老的数据(温/冷数据)放到不同的磁盘上去，在我们的环境中，读写的磁盘是分离的，冷数据移到其它磁盘上，保证做catch-up read的时候，服务请求是打到另外一块磁盘上去的。
开发一个分布式系统的最大问题是什么呢?其实是测试，你看即使是像谷歌、微软这样级别的公司都是直到现在才有比较多的分布式系统，用一致性算法实现分布式系统的过程中，工程化难度特别高。
讲到测试，那么就有三个概念是绕不过的，fault、failure和error，它们三者之间其实是有某种比较复杂的关系，比如说硬件出错了会导致系统软件层面的报错，报错之后，用户调用时会得到error。所以，在分布式系统中做测试要考虑各方面的原因，例如：随机去kill掉分布式系统中的一个节点，那么这个节点可能是leader，也可能是follower节点。某些测试用例需要kill集群中多数派的节点，甚至是所有节点全部干点等等，我们需要去构建各种各样的情景。
在网络分区情况下，系统的表现是很难去验证的，所以我们在测试框架中注入了模拟网络的分区、网络延时。其中网络延时包含定点延时、定点丢包以及随机丢包、随机延时。
除此之外还有IO问题。线上运营过大规模系统的同学应该都有这样的经历，平时测试的时候都很好，但是一上线就会有各种问题。我们是如何做的呢?我们在不同的节点之间不仅做了系统级别的异常注入和诊断测试，还在代码中插入了同步点，可以在关键路径上触发同步点，让逻辑走进去。
解决完上述问题后，下面和大家分享一个我们自己开发的工具，这个工具有三部分组成，最上面是driver，驱动每个case发起和最终执行，下面的agent与要测试的分布式系统中每个节点都是匹配的，它会去接受driver发行的请求，例如杀掉节点，模拟节点上的磁盘抖动和延迟。Agent会和下面engine之间的交互，如果做了同步点插入，engine通常会侵入到业务逻辑代码中，系统级的磁盘IO异常和网络IO抖动是不需要有代码侵入的。
FIU是 Fault Injection Utility的简称，第一步发起请求，设置一个case，例如A节点上网络抖动，B节点上磁盘抖动，C节点上CPU紧张，设置好之后才是发起真正要跑的case。
刚才我们讲了，分布式系统的论文读起来貌似很简单，但是真正落地或者工程化却不是那么容易了。我们做了很多非常有意义的工程特性，下面和大家分享一下。
第一个是fencing token，通过租约或者其它方式来选主，其实都有一个预设前提：不同节点的时钟同步偏差不大，这样某个节点在租约期内可以被降级或转换状态。如果遭遇GC停顿后还认为自己是leader，再去以leader的身份做事情，就会出现比较严重的问题了。所以，我们会有一个兜底方案，存储系统本身去做fencing，拒绝上一任leader的读写请求，也就是fencing Token。
第二个是leader transfer，因为这个系统会作为其他分布式系统的基础，我会让log系统和分布式系统的节点尽量都在IDC内部或都在本机，再去调度集群的leader，并把它部署在某个节点上。简单讲就是保证log和log消费者在同IDC甚至同机器，这样两者之间的通讯代价很小。
存储log的时候我们没有采用主流的方式，而是通过写一份日志，建不同的索引来达到目标。
ZooKeeper默认是既可以读leader，也可以读follower，但现在很多主流的强一致系统为了保持线性一致性，都是在leader上去读，我们开发了一个SDK支持在follower上去消费，并以这种方式对消费者提供不同一致性级别。传统leader/follower模式的强一致系统，leader比较容易成为瓶颈，我们其实还实现了一种在多个follower之间通过quorum和log index来达到强一致读的方案，具体在后续我们单独分享出来。
Raft pre-vote，就是在一个集群遭遇网络分区时，通常一个节点比如leader节点被分区以后，多数派会选出一个新的leader，但旧的leader会不停的尝试，等网络恢复以后，你会发现他尝试了很多次，这时他就会和正常多数派集群当中的节点来争抢leader，但如果它再参与选举的话，会中断整个集群对外的服务能力，为了使不可服务时间尽可能短，我们会用pre-vote来减少这个时间窗口。
Leader lease就是在遭遇网络分区的情况下，如何保证整个系统对外大多数时间都是一个leader?我们采用的方式是租约。
上图是一个和Kafka最新版本的简单测试，Kafka配置的是强刷盘模式，所有样本都是一样的，我们的测试是在同城IDC之间、RTT大约两毫秒左右的三副本集群测试。
经过实测，这个分布式日志系统TPS大概会提升至130%左右，Latency会下降30%左右。因为我们的强一致是在Leader上读的，没有开启follower读，所以在读方面没有太明显的变化。如果是在读写混合的场景下，我们写的优势会更大。
应用形态
应用形态其实前文已经提过了，如果要实现分布式的强一致Redis，那么只有一个难点就是如何借助Redis的bgsave实现snapshot功能。
第二，基于分布式log系统实现MQ。它采取的思路还是比较容易理解的，上面的消息和下面的存储分离开来，相当于是一种计算和存储相分离的架构。不同的是，上面的消息处理节点是无状态的，下面的log存储节点是有状态的。
第三，做数据库内核复制其实基本都是一个思路——DataBase State Machine。我们实现了一个强一致的MySQL自治集群，其中的难点包括Binlog如何高效Replay到Slave;Binlog多点冗余;Proxy的无状态等等。
