分布式系统面临的第一个问题就是数据分布，即将数据均匀地分布到多个存储节点。另外，为了保证可靠性和可用性，需要将数据复制多个副本，这就带来了多个副本之间的数据一致性问题。大规模分布式存储系统的重要目标就是节省成本，因而只能采用性价比较高的PC服务器。这些服务器性能很好，但是故障率很高，要求系统能够在软件层面实现自动容错。当存储节点出现故障时，系统能够自动检测出来，并将原有的数据和服务迁移到集群中其他正常工作的节点。

分布式系统中有两个重要的协议，包括Paxos选举协议以及两阶段提交协议。Paxos协议用于多个节点之间达成一致，往往用于实现总控节点选举。两阶段提交协议用于保证跨多个节点操作的原子性，这些操作要么全部成功，要么全部失败。理解了这两个分布式协议之后，学习其他分布式协议会变得相当容易。

异常

在分布式存储系统中，往往将一台服务器或者服务器上运行的一个进程称为一个节点，节点与节点之间通过网络互联。大规模分布式存储系统的一个核心问题在于自动容错。然而，服务器节点是不可靠的，网络也是不可靠的，本节介绍系统运行过程中可能会遇到的各种异常。

1.异常类型

（1）服务器宕机

引发服务器宕机的原因可能是内存错误、服务器停电等。服务器宕机可能随时发生，当发生宕机时，节点无法正常工作，称为“不可用”（unavailable）。服务器重启后，节点将失去所有的内存信息。因此，设计存储系统时需要考虑如何通过读取持久化介质（如机械硬盘，固态硬盘）中的数据来恢复内存信息，从而恢复到宕机前的某个一致的状态。进程运行过程中也可能随时因为core dump等原因退出，和服务器宕机一样，进程重启后也需要恢复内存信息。

（2）网络异常

引发网络异常的原因可能是消息丢失、消息乱序（如采用UDP方式通信）或者网络包数据错误。有一种特殊的网络异常称为“网络分区”，即集群的所有节点被划分为多个区域，每个区域内部可以正常通信，但是区域之间无法通信。例如，某分布式系统部署在两个数据中心，由于网络调整，导致数据中心之间无法通信，但是，数据中心内部可以正常通信。

设计容错系统的一个基本原则是：网络永远是不可靠的，任何一个消息只有收到对方的回复后才可以认为发送成功，系统设计时总是假设网络将会出现异常并采取相应的处理措施。

（3）磁盘故障

磁盘故障是一种发生概率很高的异常。磁盘故障分为两种情况：磁盘损坏和磁盘数据错误。磁盘损坏时，将会丢失存储在上面的数据，因而，分布式存储系统需要考虑将数据存储到多台服务器，即使其中一台服务器磁盘出现故障，也能从其他服务器上恢复数据。对于磁盘数据错误，往往可以采用校验和（checksum）机制来解决，这样的机制既可以在操作系统层面实现，又可以在上层的分布式存储系统层面实现。

（4）超时

由于网络异常的存在，分布式存储系统中请求结果存在“三态”的概念。在单机系统中，只要服务器没有发生异常，每个函数的执行结果是确定的，要么成功，要么失败。然而，在分布式系统中，如果某个节点向另外一个节点发起RPC（Remote Procedure Call）调用，这个RPC执行的结果有三种状态：“成功”、“失败”、“超时”（未知状态），也称为分布式存储系统的三态。

图3-1给出了RPC执行成功但超时的例子。服务器（Server）收到并成功处理完成客户端（Client）的请求，但是由于网络异常或者服务器宕机，客户端没有收到服务器端的回复。此时，RPC的执行结果为超时，客户端不能简单地认为服务器端处理失败。

当出现超时状态时，只能通过不断读取之前操作的状态来验证RPC操作是否成功。当然，设计分布式存储系统时可以将操作设计为“幂等”的，也就是说，操作执行一次与执行多次的结果相同，例如，覆盖写就是一种常见的幂等操作。如果采用这种设计，当出现失败和超时时，都可以采用相同的处理方式，即一直重试直到成功。

一致性

由于异常的存在，分布式存储系统设计时往往会将数据冗余存储多份，每一份称为一个副本（replica/copy）。这样，当某一个节点出现故障时，可以从其他副本上读到数据。可以这么认为，副本是分布式存储系统容错技术的唯一手段。由于多个副本的存在，如何保证副本之间的一致性是整个分布式系统的理论核心。

可以从两个角度理解一致性：第一个角度是用户，或者说是客户端，即客户端读写操作是否符合某种特性；第二个角度是存储系统，即存储系统的多个副本之间是否一致，更新的顺序是否相同，等等。

首先定义如下场景，这个场景包含三个组成部分：

●存储系统：存储系统可以理解为一个黑盒子，它为我们提供了可用性和持久性的保证。

●客户端A：客户端A主要实现从存储系统write和read操作。

●客户端B和客户端C：客户端B和C是独立于A，并且B和C也相互独立的，它们同时也实现对存储系统的write和read操作。

从客户端的角度来看，一致性包含如下三种情况：

●强一致性：假如A先写入了一个值到存储系统，存储系统保证后续A,B，C的读取操作都将返回最新值。当然，如果写入操作“超时”，那么成功或者失败都是可能的，客户端A不应该做任何假设。

●弱一致性：假如A先写入了一个值到存储系统，存储系统不能保证后续A,B，C的读取操作是否能够读取到最新值。

●最终一致性：最终一致性是弱一致性的一种特例。假如A首先写入一个值到存储系统，存储系统保证如果后续没有写操作更新同样的值，A，B，C的读取操作“最终”都会读取到A写入的最新值。“最终”一致性有一个“不一致窗口”的概念，它特指从A写入值，到后续A，B，C读取到最新值的这段时间。“不一致性窗口”的大小依赖于以下的几个因素：交互延迟，系统的负载，以及复制协议要求同步的副本数。

最终一致性描述比较粗略，其他常见的变体如下：

●读写（Read-your-writes）一致性：如果客户端A写入了最新的值，那么A的后续操作都会读取到最新值。但是其他用户（比如B或者C）可能要过一会才能看到。

●会话（Session）一致性：要求客户端和存储系统交互的整个会话期间保证读写一致性。如果原有会话因为某种原因失效而创建了新的会话，原有会话和新会话之间的操作不保证读写一致性。

●单调读（Monotonic read）一致性：如果客户端A已经读取了对象的某个值，那么后续操作将不会读取到更早的值。

●单调写（Monotonic write）一致性：客户端A的写操作按顺序完成，这就意味着，对于同一个客户端的操作，存储系统的多个副本需要按照与客户端相同的顺序完成。

从存储系统的角度看，一致性主要包含如下几个方面：

●副本一致性：存储系统的多个副本之间的数据是否一致，不一致的时间窗口等；

●更新顺序一致性：存储系统的多个副本之间是否按照相同的顺序执行更新操作。

一般来说，存储系统可以支持强一致性，也可以为了性能考虑只支持最终一致性。从客户端的角度看，一般要求存储系统能够支持读写一致性，会话一致性，单调读，单调写等特性，否则，使用比较麻烦，适用的场景也比较有限。

衡量指标

评价分布式存储系统有一些常用的指标，下面分别介绍。

（1）性能

常见的性能指标有：系统的吞吐能力以及系统的响应时间。其中，系统的吞吐能力指系统在某一段时间可以处理的请求总数，通常用每秒处理的读操作数（QPS,Query Per Second）或者写操作数（TPS,Transaction Per Second）来衡量；系统的响应延迟，指从某个请求发出到接收到返回结果消耗的时间，通常用平均延时或者99.9%以上请求的最大延时来衡量。这两个指标往往是矛盾的，追求高吞吐的系统，往往很难做到低延迟；追求低延迟的系统，吞吐量也会受到限制。因此，设计系统时需要权衡这两个指标。

（2）可用性

系统的可用性（availability）是指系统在面对各种异常时可以提供正常服务的能力。系统的可用性可以用系统停服务的时间与正常服务的时间的比例来衡量，例如某系统的可用性为4个9（99.99%），相当于系统一年停服务的时间不能超过365×24×60/10000=52.56分钟。系统可用性往往体现了系统的整体代码质量以及容错能力。

3）一致性

3.1.2节说明了系统的一致性。一般来说，越是强的一致性模型，用户使用起来越简单。笔者认为，如果系统部署在同一个数据中心，只要系统设计合理，在保证强一致性的前提下，不会对性能和可用性造成太大的影响。后文中笔者在Alibaba参与开发的OceanBase系统以及Google的分布式存储系统都倾向强一致性。

4）可扩展性

系统的可扩展性（scalability）指分布式存储系统通过扩展集群服务器规模来提高系统存储容量、计算量和性能的能力。随着业务的发展，对底层存储系统的性能需求不断增加，比较好的方式就是通过自动增加服务器提高系统的能力。理想的分布式存储系统实现了“线性可扩展”，也就是说，随着集群规模的增加，系统的整体性能与服务器数量呈线性关系。

性能分析

给定一个问题，往往会有多种设计方案，而方案评估的一个重要指标就是性能，如何在系统设计之初估算存储系统的性能是存储工程师的必备技能。性能分析用来判断设计方案是否存在瓶颈点，权衡多种设计方案，另外，性能分析也可作为后续性能优化的依据。性能分析与性能优化是相对的，系统设计之初通过性能分析确定设计目标，防止出现重大的设计失误，等到系统试运行后，需要通过性能优化方法找出系统中的瓶颈点并逐步消除，使得系统达到设计之初确定的设计目标。

性能分析的结果是不精确的，然而，至少可以保证，估算的结果与实际值不会相差一个数量级。设计之初首先分析整体架构，接着重点分析可能成为瓶颈的单机模块。系统中的资源（CPU、内存、磁盘、网络）是有限的，性能分析就是需要找出可能出现的资源瓶颈。本节通过几个实例说明性能分析方法。

1.生成一张有30张缩略图（假设图片原始大小为256KB）的页面需要多少时间？

●方案1：顺序操作，每次先从磁盘中读取图片，再执行生成缩略图操作，执行时间为：30×10ms（磁盘随机读取时间）+30×256K/30MB/s（假设缩略图生成速度为30MB/s）=560ms

●方案2：并行操作，一次性发送30个请求，每个请求读取一张图片并生成缩略图，执行时间为：10ms+256K/300MB/s=18ms

当然，系统实际运行的时候可能有缓存以及其他因素的干扰，这些因素在性能估算阶段可以先不考虑，简单地将估算结果乘以一个系数即为实际值。

2.1GB的4字节整数，执行一次快速排序需要多少时间？

Google的Jeff Dean提出了一种排序性能分析方法：排序时间=比较时间（分支预测错误）+内存访问时间。快速排序过程中会发生大量的分支预测错误，所以比较次数为2_{28}×log（2_{28}）≈2_{33}，其中，约1/2的比较会发生分支预测错误，所以比较时间为1/2×2_{33}×5ns=21s，另外，快速排序每次分割操作都需要扫描一遍内存，假设内存顺序访问性能为4GB/s，所以内存访问时间为28×1GB/4GB=7s。因此，单线程排序1GB 4字节整数总时间约为28s。

3.Bigtable系统性能分析

Bigtable是Google的分布式表格系统，它的优势是可扩展性好，可随时增加或者减少集群中的服务器，但支持的功能有限，支持的操作主要包括：

●单行操作：基于主键的随机读取，插入，更新，删除（CRUD）操作；

●多行扫描：扫描一段主键范围内的数据。Bigtable中每行包括多个列，每一行的某一列对应一个数据单元，每个数据单元包括多个版本，可以按照列名或者版本对扫描结果进行过滤。

假设某类Bigtable系统的总体设计中给出的性能指标为：

●系统配置：同一个机架下40台服务器（8核，24GB内存，10路15000转SATA硬盘）；

●表格：每行数据1KB，64KB一个数据块，不压缩。

a）随机读取（缓存不命中）：1KB/item×300item/s=300KB/s

Bigtable系统中每次随机读取需要首先从GFS中读取一个64KB的数据块，经过CPU处理后返回用户一行数据（大小为1KB）。因此，性能受限于GFS中ChunkServer（GFS系统中的工作节点）的磁盘IOPS以及Bigtable Tablet Server（Bigtable系统中的工作节点）的网络带宽。先看底层的GFS，每台机器拥有10块SATA盘，每块SATA盘的IOPS约为100，因此，每台机器的IOPS理论值约为1000，考虑到负载均衡等因素，将随机读取的QPS设计目标定为300，保留一定的余量。另外，每台机器每秒从GFS中读取的数据量为300×64KB=19.2MB，由于所有的服务器分布在同一个机架下，网络不会成为瓶颈。

b）随机读取（内存表）：1KB/item×20000items/s=20MB/s

Bigtable中支持内存表，内存表的数据全部加载到内存中，读取时不需要读取底层的GFS。随机读取内存表的性能受限于CPU以及网络，内存型服务的QPS一般在10W，由于网络发送小数据有较多overhead且Bigtable内存操作有较多的CPU开销，保守估计每个节点的QPS为20000，客户端和Tablet Server之间的网络流量为20MB/s。

c）随机写/顺序写：1KB/item×8000item/s=8MB/s

Bigtable中随机写和顺序写的性能是差不多的，写入操作需要首先将操作日志写入到GFS，接着修改本地内存。为了提高性能，Bigtable实现了成组提示技术，即将很多写操作凑成一批（比如512KB～2MB）一次性提交到GFS中。Bigtable每次写一份数据需要在GFS系统中重复写入3份到10份，当写入速度达到8000 QPS，即8MB/s后Tablet Server的网络将成为瓶颈。

d）扫描：1KB/item×30000item/s=30MB/s

Bigtable扫描操作一次性从GFS中读取大量的数据（比如512KB～2MB），GFS的磁盘IO不会成为瓶颈。另外，批量操作减少了CPU以及网络收发包的开销，扫描操作的瓶颈在于Tablet Server读取底层GFS的带宽，估计为30MB/s，对应30000 QPS。

如果集群规模超过40台，不能保证所有的服务器在同一个机架下，系统设计以及性能分析都会有所不同。性能分析可能会很复杂，因为不同情况下系统的瓶颈点不同，有的时候是网络，有的时候是磁盘，有的时候甚至是机房的交换机或者CPU，另外，负载均衡以及其他因素的干扰也会使得性能更加难以量化。只有理解存储系统的底层设计和实现，并在实践中不断地练习，性能估算才会越来越准。

数据分布

分布式系统区别于传统单机系统在于能够将数据分布到多个节点，并在多个节点之间实现负载均衡。数据分布的方式主要有两种，一种是哈希分布，如一致性哈希，代表系统为Amazon的Dynamo系统；另外一种方法是顺序分布，即每张表格上的数据按照主键整体有序，代表系统为Google的Bigtable系统。Bigtable将一张大表根据主键切分为有序的范围，每个有序范围是一个子表。

将数据分散到多台机器后，需要尽量保证多台机器之间的负载是比较均衡的。衡量机器负载涉及的因素很多，如机器Load值，CPU，内存，磁盘以及网络等资源使用情况，读写请求数及请求量，等等，分布式存储系统需要能够自动识别负载高的节点，当某台机器的负载较高时，将它服务的部分数据迁移到其他机器，实现自动负载均衡。

分布式存储系统的一个基本要求就是透明性，包括数据分布透明性，数据迁移透明性，数据复制透明性，故障处理透明性。本节介绍数据分布以及数据迁移相关的基础知识。

哈希分布

哈希取模的方法很常见，其方法是根据数据的某一种特征计算哈希值，并将哈希值与集群中的服务器建立映射关系，从而将不同哈希值的数据分布到不同的服务器上。所谓数据特征可以是key-value系统中的主键（key），也可以是其他与业务逻辑相关的值。例如，将集群中的服务器按0到N-1编号（N为服务器的数量），根据数据的主键（hash（key）%N）或者数据所属的用户id（hash（user_id）%N）计算哈希值，来决定将数据映射到哪一台服务器。

如果哈希函数的散列特性很好，哈希方式可以将数据比较均匀地分布到集群中去。而且，哈希方式需要记录的元信息也非常简单，每个节点只需要知道哈希函数的计算方式以及模的服务器的个数就可以计算出处理的数据应该属于哪台机器。然而，找出一个散列特性很好的哈希函数是很难的。这是因为，如果按照主键散列，那么同一个用户id下的数据可能被分散到多台服务器，这会使得一次操作同一个用户id下的多条记录变得困难；如果按照用户id散列，容易出现“数据倾斜”（data skew）问题，即某些大用户的数据量很大，无论集群的规模有多大，这些用户始终由一台服务器处理。

传统的哈希分布算法还有一个问题：当服务器上线或者下线时，N值发生变化，数据映射完全被打乱，几乎所有的数据都需要重新分布，这将带来大量的数据迁移。

一种思路是不再简单地将哈希值和服务器个数做除法取模映射，而是将哈希值与服务器的对应关系作为元数据，交给专门的元数据服务器来管理。访问数据时，首先计算哈希值，再查询元数据服务器，获得该哈希值对应的服务器。这样，集群扩容时，可以将部分哈希值分配给新加入的机器并迁移对应的数据。

另一种思路就是采用一致性哈希（Distributed Hash Table,DHT）算法。算法思想如下：给系统中每个节点分配一个随机token，这些token构成一个哈希环。执行数据存放操作时，先计算Key（主键）的哈希值，然后存放到顺时针方向第一个大于或者等于该哈希值的token所在的节点。一致性哈希的优点在于节点加入/删除时只会影响到在哈希环中相邻的节点，而对其他节点没影响。

●首先求出每个服务器的hash值，将其配置到一个0～2_{n}的圆环区间上；

●其次使用同样的方法求出待存储对象的主键哈希值，也将其配置到这个圆环上；

●然后从数据映射的位置开始顺时针查找，将数据分布到找到的第一个服务器节点。

增加服务节点5以后，某些原来分布到节点3的数据需要迁移到节点5，其他数据分布均保持不变。可以看出，一致性哈希算法在很大程度上避免了数据迁移。

为了查找集群中的服务器，需要维护每台机器在哈希环中位置信息，常见的做法如下。

（1）O（1）位置信息

每台服务器记录它的前一个以及后一个节点的位置信息。这种做法的维护的节点位置信息的空间复杂度为O（1），然而每一次查找都可能遍历整个哈希环中的所有服务器，即时间复杂度为O（N），其中，N为服务器数量。

（2）O（logN）位置信息

假设哈希空间为0～2_{n}（即N=2_{n}），以Chord系统为例，为了加速查找，它在每台服务器维护了一个大小为n的路由表（finger table），FT^{P}[i]=succ（p+2_{i-1}），其中p为服务器在哈希环中的编号，路由表中的第i个元素记录了编号为p+2_{i-1}的后继节点。通过维护O（logN）的位置信息，查找的时间复杂度改进为O（logN）。

（3）O（N）位置信息

Dynamo系统通过牺牲空间换时间，在每台服务器维护整个集群中所有服务器的位置信息，将查找服务器的时间复杂度降为O（1）。工程上一般都采用这种做法，Dynamo这样的P2P系统在每个服务器节点上都维护了所有服务器的位置信息，而带有总控节点的存储系统往往由总控节点统一维护。

一致性哈希还需要考虑负载均衡，增加服务节点node5后，虽然只影响到node5的后继，即node3的数据分布，但node3节点需要迁移的数据过多，整个集群的负载不均衡。一种自然的想法是将需要迁移的数据分散到整个集群，每台服务器只需要迁移1/N的数据量。为此，Dynamo中引入虚拟节点的概念，5.1节会详细讨论。

顺序分布

希散列破坏了数据的有序性，只支持随机读取操作，不能够支持顺序扫描。某些系统可以在应用层做折衷，比如互联网应用经常按照用户来进行数据拆分，并通过哈希方法进行数据分布，同一个用户的数据分布到相同的存储节点，允许对同一个用户的数据执行顺序扫描，由应用层解决跨多个用户的操作问题。另外，这种方式可能出现某些用户的数据量太大的问题，由于用户的数据限定在一个存储节点，无法发挥分布式存储系统的多机并行处理能力。

顺序分布在分布式表格系统中比较常见，一般的做法是将大表顺序划分为连续的范围，每个范围称为一个子表，总控服务器负责将这些子表按照一定的策略分配到存储节点上。如图3-3所示，用户表（User表）的主键范围为1～7000，在分布式存储系统中划分为多个子表，分别对应数据范围1～1000，1001～2000，……6001～7000。Meta表是可选的，某些系统只有根表（Root表）一级索引，在Root表中维护用户表的位置信息，即每个User子表在哪个存储节点上。为了支持更大的集群规模，Bigtable这样的系统将索引分为两级：根表以及元数据表（Meta表），由Meta表维护User表的位置信息，而Root表用来维护Meta表的位置信息。读User表时，需要通过Meta表查找相应的User子表所在的存储节点，而读取Meta表又需要通过Root表查找相应的Meta子表所在的存储节点。

顺序分布与B+树数据结构比较类似，每个子表相当于叶子节点，随着数据的插入和删除，某些子表可能变得很大，某些变得很小，数据分布不均匀。如果采用顺序分布，系统设计时需要考虑子表的分裂与合并，这将极大地增加系统复杂度。子表分裂指当一个子表太大超过一定阀值时需要分裂为两个子表，从而有机会通过系统的负载均衡机制分散到多个存储节点。子表合并一般由数据删除引起，当相邻的两个子表都很小时，可以合并为一个子表。一般来说，单个服务节点能够服务的子表数量是有限的，比如4000～10000个，子表合并的目的是为了防止系统中出现过多太小的子表，减少系统中的元数据。

负载均衡

分布式存储系统的每个集群中一般有一个总控节点，其他节点为工作节点，由总控节点根据全局负载信息进行整体调度。工作节点刚上线时，总控节点需要将数据迁移到该节点，另外，系统运行过程中也需要不断地执行迁移任务，将数据从负载较高的工作节点迁移到负载较低的工作节点。

工作节点通过心跳包（Heartbeat，定时发送）将节点负载相关的信息，如CPU，内存，磁盘，网络等资源使用率，读写次数及读写数据量等发送给主控节点。主控节点计算出工作节点的负载以及需要迁移的数据，生成迁移任务放入迁移队列中等待执行。需要注意的是，负载均衡操作需要控制节奏，比如一台全新的工作节点刚上线的时候，由于负载最低，如果主控节点将大量的数据同时迁移到这台新加入的机器，整个系统在新增机器的过程中服务能力会大幅下降。负载均衡操作需要做到比较平滑，一般来说，从新机器加入到集群负载达到比较均衡的状态需要较长一段时间，比如30分钟到一个小时。

负载均衡需要执行数据迁移操作。在分布式存储系统中往往会存储数据的多个副本，其中一个副本为主副本，其他副本为备副本，由主副本对外提供服务。迁移备副本不会对服务造成影响，迁移主副本也可以首先将数据的读写服务切换到其他备副本。整个迁移过程可以做到无缝，对用户完全透明。

假设数据分片D有两个副本D1和D2，分别存储在工作节点A1和A2，其中，D1为主副本，提供读写服务，D2为备副本。如果需要将D1从工作节点A1中迁移出去，大致的操作步骤如下：

1）将数据分片D的读写服务由工作节点A1切换到A2，D2变成主副本；

2）增加副本：选择某个节点，例如B节点，增加D的副本，即B节点从A2节点获取D的副本数据（D2）并与之保持同步；

3）删除工作节点A1上的D1副本。

复制

为了保证分布式存储系统的高可靠和高可用，数据在系统中一般存储多个副本。当某个副本所在的存储节点出现故障时，分布式存储系统能够自动将服务切换到其他的副本，从而实现自动容错。分布式存储系统通过复制协议将数据同步到多个存储节点，并确保多个副本之间的数据一致性。

同一份数据的多个副本中往往有一个副本为主副本（Primary），其他副本为备副本（Backup），由主副本将数据复制到备份副本。复制协议分为两种，强同步复制以及异步复制，二者的区别在于用户的写请求是否需要同步到备副本才可以返回成功。假如备份副本不止一个，复制协议还会要求写请求至少需要同步到几个备副本。当主副本出现故障时，分布式存储系统能够将服务自动切换到某个备副本，实现自动容错。

一致性和可用性是矛盾的，强同步复制协议可以保证主备副本之间的一致性，但是当备副本出现故障时，也可能阻塞存储系统的正常写服务，系统的整体可用性受到影响；异步复制协议的可用性相对较好，但是一致性得不到保障，主副本出现故障时还有数据丢失的可能。

复制的概述

分布式存储系统中数据保存多个副本，一般来说，其中一个副本为主副本，其他副本为备副本，常见的做法是数据写入到主副本，由主副本确定操作的顺序并复制到其他副本。

如图3-4所示，客户端将写请求发送给主副本，主副本将写请求复制到其他备副本，常见的做法是同步操作日志（Commit Log）。主副本首先将操作日志同步到备副本，备副本回放操作日志，完成后通知主副本。接着，主副本修改本机，等到所有的操作都完成后再通知客户端写成功。图3-4中的复制协议要求主备同步成功才可以返回客户端写成功，这种协议称为强同步协议。强同步协议提供了强一致性，但是，如果备副本出现问题将阻塞写操作，系统可用性较差。

假设所有副本的个数为N，且N＞2，即备副本个数大于1。那么，实现强同步协议时，主副本可以将操作日志并发地发给所有备副本并等待回复，只要至少1个备副本返回成功就可以回复客户端操作成功。强同步的好处在于如果主副本出现故障，至少有1个备副本拥有完整的数据，分布式存储系统可以自动地将服务切换到最新的备副本而不用担心数据丢失的情况。

与强同步对应的复制方式是异步复制。在异步模式下，主副本不需要等待备副本的回应，只需要本地修改成功就可以告知客户端写操作成功。另外，主副本通过异步机制，比如单独的复制线程将客户端修改操作推送到其他副本。异步复制的好处在于系统可用性较好，但是一致性较差，如果主副本发生不可恢复故障，可能丢失最后一部分更新操作。

强同步复制和异步复制都是将主副本的数据以某种形式发送到其他副本，这种复制协议称为基于主副本的复制协议（Primary-based protocol）。这种方法要求在任何时刻只能有一个副本为主副本，由它来确定写操作之间的顺序。如果主副本出现故障，需要选举一个备副本成为新的主副本，这步操作称为选举，经典的选举协议为Paxos协议，3.7.2节将专门进行介绍。

主备副本之间的复制一般通过操作日志来实现。操作日志的原理很简单：为了利用好磁盘的顺序读写特性，将客户端的写操作先顺序写入到磁盘中，然后应用到内存中，由于内存是随机读写设备，可以很容易通过各种数据结构，比如B+树将数据有效地组织起来。当服务器宕机重启时，只需要回放操作日志就可以恢复内存状态。为了提高系统的并发能力，系统会积攒一定的操作日志再批量写入到磁盘中，这种技术一般称为成组提交。

如果每次服务器出现故障都需要回放所有的操作日志，效率是无法忍受的，检查点（checkpoint）正是为了解决这个问题。系统定期将内存状态以检查点文件的形式dump到磁盘中，并记录检查点时刻对应的操作日志回放点。检查点文件成功创建后，回放点之前的日志可以被垃圾回收，以后如果服务器出现故障，只需要回放检查点之后的操作日志。

除了基于主副本的复制协议，分布式存储系统中还可能使用基于写多个存储节点的复制协议（Replicated-write protocol）。比如Dynamo系统中的NWR复制协议，其中，N为副本数量，W为写操作的副本数，R为读操作的副本数。NWR协议中多个副本不再区分主和备，客户端根据一定的策略往其中的W个副本写入数据，读取其中的R个副本。只要W+R＞N，可以保证读到的副本中至少有一个包含了最新的更新。然而，这种协议的问题在于不同副本的操作顺序可能不一致，从多个副本读取时可能出现冲突。这种方式在实际系统中比较少见，不建议使用。

一致性与可用性

来自Berkerly的Eric Brewer教授提出了一个著名的CAP理论：一致性（Consistency），可用性（Availability）以及分区可容忍性（Tolerance of network Partition）三者不能同时满足。笔者认为没有必要纠结CAP理论最初的定义，在工程实践中，可以将C、A、P三者按如下方式理解：

●一致性：读操作总是能读取到之前完成的写操作结果，满足这个条件的系统称为强一致系统，这里的“之前”一般对同一个客户端而言；

●可用性：读写操作在单台机器发生故障的情况下仍然能够正常执行，而不需要等待发生故障的机器重启或者其上的服务迁移到其他机器；

●分区可容忍性：机器故障、网络故障、机房停电等异常情况下仍然能够满足一致性和可用性。

分布式存储系统要求能够自动容错，也就是说，分区可容忍性总是需要满足的，因此，一致性和写操作的可用性不能同时满足。

如果采用强同步复制，保证了存储系统的一致性，然而，当主备副本之间出现网络或者其他故障时，写操作将被阻塞，系统的可用性无法得到满足。如果采用异步复制，保证了存储系统的可用性，但是无法做到强一致性。

存储系统设计时需要在一致性和可用性之间权衡，在某些场景下，不允许丢失数据，在另外一些场景下，极小的概率丢失部分数据时允许的，可用性更加重要。例如，Oracle数据库的DataGuard复制组件包含三种模式：

●最大保护模式（Maximum Protection）：即强同步复制模式，写操作要求主库先将操作日志（数据库的redo/undo日志）同步到至少一个备库才可以返回客户端成功。这种模式保证即使主库出现无法恢复的故障，比如硬盘损坏，也不会丢失数据。

●最大性能模式（Maximum Performance）：即异步复制模式，写操作只需要在主库上执行成功就可以返回客户端成功，主库上的后台线程会将重做日志通过异步的方式复制到备库。这种方式保证了性能及可用性，但是可能丢失数据。

●最大可用性模式（Maximum Availability）：上述两种模式的折衷。正常情况下相当于最大保护模式，如果主备之间的网络出现故障，切换为最大性能模式。这种模式在一致性和可用性之间做了一个很好的权衡，推荐大家在设计存储系统时使用。

容错

随着集群规模变得越来越大，故障发生的概率也越来越大，大规模集群每天都有故障发生。容错是分布式存储系统设计的重要目标，只有实现了自动化容错，才能减少人工运维成本，实现分布式存储的规模效应。

单台服务器故障的概率是不高的，然而，只要集群的规模足够大，每天都可能有机器故障发生，系统需要能够自动处理。首先，分布式存储系统需要能够检测到机器故障，在分布式系统中，故障检测往往通过租约（Lease）协议实现。接着，需要能够将服务复制或者迁移到集群中的其他正常服务的存储节点。

本节首先介绍Google某数据中心发生的故障，接着讨论分布式系统中的故障检测以及恢复方法。

可扩展性

通过数据分布，复制以及容错等机制，能够将分布式存储系统部署到成千上万台服务器。可扩展性的实现手段很多，如通过增加副本个数或者缓存提高读取能力，将数据分片使得每个分片可以被分配到不同的工作节点以实现分布式处理，把数据复制到多个数据中心，等等。

分布式存储系统大多都带有总控节点，很多人会自然地联想到总控节点的瓶颈问题，认为P2P架构更有优势。然而，事实却并非如此，主流的分布式存储系统大多带有总控节点，且能够支持成千上万台的集群规模。

另外，传统的数据库也能够通过分库分表等方式对系统进行水平扩展，当系统处理能力不足时，可以通过增加存储节点来扩容。

那么，如何衡量分布式存储系统的可扩展性，它与传统数据库的可扩展性又有什么区别？可扩展性不能简单地通过系统是否为P2P架构或者是否能够将数据分布到多个存储节点来衡量，而应该综合考虑节点故障后的恢复时间，扩容的自动化程度，扩容的灵活性等。

本节首先讨论总控节点是否会成为性能瓶颈，接着介绍传统数据库的可扩展性，最后讨论同构系统与异构系统增加节点时的差别。

总控节点

分布式存储系统中往往有一个总控节点用于维护数据分布信息，执行工作机管理，数据定位，故障检测和恢复，负载均衡等全局调度工作。通过引入总控节点，可以使得系统的设计更加简单，并且更加容易做到强一致性，对用户友好。那么，总控节点是否会成为性能瓶颈呢？

分为两种情况：分布式文件系统的总控节点除了执行全局调度，还需要维护文件系统目录树，内存容量可能会率先成为性能瓶颈；而其他分布式存储系统的总控节点只需要维护数据分片的位置信息，一般不会成为瓶颈。另外，即使是分布式文件系统，只要设计合理，也能够扩展到几千台服务器。例如，Google的分布式文件系统能够扩展到8000台以上的集群，开源的Hadoop也能够扩展到3000台以上的集群。当然，设计时需要减少总控节点的负载，比如Google的GFS舍弃了对小文件的支持，并且把对数据的读写控制权下放到工作机ChunkServer，通过客户端缓存元数据减少对总控节点的访问等。

如果总控节点成为瓶颈，例如需要支持超过一万台的集群规模，或者需要支持海量的小文件，那么，可以采用两级结构，如图3-6所示。在总控机与工作机之间增加一层元数据节点，每个元数据节点只维护一部分而不是整个分布式文件系统的元数据。这样，总控机也只需要维护元数据节点的元数据，不可能成为性能瓶颈。假设分布式文件系统（Distributed File System,DFS）中有100个元数据节点，每个元数据节点服务1亿个文件，系统总共可以服务100亿个文件。图3-6中的DFS客户端定位DFS工作机时，需要首先访问DFS总控机找到DFS元数据服务器，再通过元数据服务器找到DFS工作机。虽然看似增加了一次网络请求，但是客户端总是能够缓存DFS总控机上的元数据，因此并不会带来额外的开销。

异构系统

传统数据库扩容与大规模存储系统的可扩展性有何区别呢？为了说明这一问题，我们首先定义同构系统，如图3-8所示。

将存储节点分为若干组，每个组内的节点服务完全相同的数据，其中有一个节点为主节点，其他节点为备节点。由于同一个组内的节点服务相同的数据，这样的系统称为同构系统。同构系统的问题在于增加副本需要迁移的数据量太大，假设每个存储节点服务的数据量为1TB，内部传输带宽限制为20MB/s，那么增加副本拷贝数据需要的时间为1TB/20MB/s=50000s，大约十几个小时，由于拷贝数据的过程中存储节点再次发生故障的概率很高，所以这样的架构很难做到自动化，不适用大规模分布式存储系统。

大规模分布式存储系统要求具有线性可扩展性，即随时加入或者删除一个或者多个存储节点，系统的处理能力与存储节点的个数成线性关系。为了实现线性可扩展性，存储系统的存储节点之间是异构的。否则，当集群规模达到一定程度后，增加节点将变得特别困难。异构系统将数据划分为很多大小接近的分片，每个分片的多个副本可以分布到集群中的任何一个存储节点。如果某个节点发生故障，原有的服务将由整个集群而不是某几个固定的存储节点来恢复。

如图3-9所示，系统中有五个分片（A,B,C,D,E），每个分片包含三个副本，如分片A的三个副本分别为A1，A2以及A3。假设节点1发生永久性故障，那么可以从剩余的节点中任意选择健康的节点来增加A,B以及E的副本。由于整个集群都参与到节点1的故障恢复过程，故障恢复时间很短，而且集群规模越大，优势就会越明显。

分布式协议

分布式系统涉及的协议很多，例如租约，复制协议，一致性协议，其中以两阶段提交协议和Paxos协议最具有代表性。两阶段提交协议用于保证跨多个节点操作的原子性，也就是说，跨多个节点的操作要么在所有节点上全部执行成功，要么全部失败。Paxos协议用于确保多个节点对某个投票（例如哪个节点为主节点）达成一致。本节介绍这两个分布式协议。

两阶段提交协议

两阶段提交协议（Two-phase Commit，2PC）经常用来实现分布式事务，在两阶段协议中，系统一般包含两类节点：一类为协调者（coordinator），通常一个系统中只有一个；另一类为事务参与者（participants,cohorts或workers），一般包含多个。协议中假设每个节点都会记录操作日志并持久化到非易失性存储介质，即使节点发生故障日志也不会丢失。顾名思义，两阶段提交协议由两个阶段组成。在正常的执行过程中，这两个阶段的执行过程如下所述：

●阶段1：请求阶段（Prepare Phase）。在请求阶段，协调者通知事务参与者准备提交或者取消事务，然后进入表决过程。在表决过程中，参与者将告知协调者自己的决策：同意（事务参与者本地执行成功）或者取消（事务参与者本地执行失败）

●阶段2：提交阶段（Commit Phase）。在提交阶段，协调者将基于第一个阶段的投票结果进行决策：提交或者取消。当且仅当所有的参与者同意提交事务协调者才通知所有的参与者提交事务，否则协调者通知所有的参与者取消事务。参与者在接收到协调者发来的消息后将执行相应的操作。

例如，A组织B、C和D三个人去爬长城：如果所有人都同意去爬长城，那么活动将举行；如果有一人不同意去爬长城，那么活动将取消。用2PC算法解决该问题的过程如下：

1）首先A将成为该活动的协调者，B、C和D将成为该活动的参与者。

2）准备阶段：A发邮件给B、C和D，提出下周三去爬山，问是否同意。那么此时A需要等待B、C和D的回复。B、C和D分别查看自己的日程安排表。B、C发现自己在当日没有活动安排，则发邮件告诉A他们同意下周三去爬长城。由于某种原因，D白天没有查看邮件。那么此时A、B和C均需要等待。到晚上的时候，D发现了A的邮件，然后查看日程安排，发现下周三当天已经有别的安排，那么D回复A说活动取消吧。

3）此时A收到了所有活动参与者的邮件，并且A发现D下周三不能去爬山。那么A将发邮件通知B、C和D，下周三爬长城活动取消。此时B、C回复A“太可惜了”，D回复A“不好意思”。至此该事务终止。

通过该例子可以发现，2PC协议存在明显的问题。假如D一直不能回复邮件，那么A、B和C将不得不处于一直等待的状态。并且B和C所持有的资源一直不能释放，即下周三不能安排其他活动。当然，A可以发邮件告诉D如果晚上六点之前不回复活动就自动取消，通过引入事务的超时机制防止资源一直不能释放的情况。更为严重的是，假如A发完邮件后生病住院了，即使B、C和D都发邮件告诉A同意下周三去爬长城，如果A没有备份，事务将被阻塞，B、C和D下周三都不能安排其他活动。

两阶段提交协议可能面临两种故障：

●事务参与者发生故障。给每个事务设置一个超时时间，如果某个事务参与者一直不响应，到达超时时间后整个事务失败。

●协调者发生故障。协调者需要将事务相关信息记录到操作日志并同步到备用协调者，假如协调者发生故障，备用协调者可以接替它完成后续的工作。如果没有备用协调者，协调者又发生了永久性故障，事务参与者将无法完成事务而一直等待下去。

总而言之，两阶段提交协议是阻塞协议，执行过程中需要锁住其他更新，且不能容错，大多数分布式存储系统都采用敬而远之的做法，放弃对分布式事务的支持。

Paxos协议

Paxos协议用于解决多个节点之间的一致性问题。多个节点之间通过操作日志同步数据，如果只有一个节点为主节点，那么，很容易确保多个节点之间操作日志的一致性。考虑到主节点可能出现故障，系统需要选举出新的主节点。Paxos协议正是用来实现这个需求。只要保证了多个节点之间操作日志的一致性，就能够在这些节点上构建高可用的全局服务，例如分布式锁服务，全局命名和配置服务等。

为了实现高可用性，主节点往往将数据以操作日志的形式同步到备节点。如果主节点发生故障，备节点会提议自己成为主节点。这里存在的问题是网络分区的时候，可能会存在多个备节点提议（Proposer，提议者）自己成为主节点。Paxos协议保证，即使同时存在多个proposer，也能够保证所有节点最终达成一致，即选举出唯一的主节点。

大多数情况下，系统只有一个proposer，他的提议也总是会很快地被大多数节点接受。Paxos协议执行步骤如下：

1）批准（accept）：Proposer发送accept消息要求所有其他节点（acceptor，接受者）接受某个提议值，acceptor可以接受或者拒绝。

2）确认（acknowledge）：如果超过一半的acceptor接受，意味着提议值已经生效，proposer发送acknowledge消息通知所有的acceptor提议生效。

当出现网络或者其他异常时，系统中可能存在多个proposer，他们各自发起不同的提议。这里的提议可以是一个修改操作，也可以是提议自己成为主节点。如果proposer第一次发起的accept请求没有被acceptor中的多数派批准（例如与其他proposer的提议冲突），那么，需要完整地执行一轮Paxos协议。过程如下：

1）准备（prepare）：Proposer首先选择一个提议序号n给其他的acceptor节点发送prepare消息。Acceptor收到prepare消息后，如果提议的序号大于他已经回复的所有prepare消息，则acceptor将自己上次接受的提议回复给proposer，并承诺不再回复小于n的提议。

2）批准（accept）：Proposer收到了acceptor中的多数派对prepare的回复后，就进入批准阶段。如果在之前的prepare阶段acceptor回复了上次接受的提议，那么，proposer选择其中序号最大的提议值发给acceptor批准；否则，proposer生成一个新的提议值发给acceptor批准。Acceptor在不违背他之前在prepare阶段的承诺的前提下，接受这个请求。

3）确认（acknowledge）：如果超过一半的acceptor接受，提议值生效。Proposer发送acknowledge消息通知所有的acceptor提议生效。

Paxos协议需要考虑两个问题：正确性，即只有一个提议值会生效；可终止性，即最后总会有一个提议值生效。Paxos协议中要求每个生效的提议被acceptor中的多数派接受，并且每个acceptor不会接受两个不同的提议，因此可以保证正确性。Paxos协议并不能够严格保证可终止性。但是，从Paxos协议的执行过程可以看出，只要超过一个acceptor接受了提议，proposer很快就会发现，并重新提议其中序号最大的提议值。因此，随着协议不断运行，它会往“某个提议值被多数派接受并生效”这一最终目标靠拢。

Paxos与2PC

Paxos协议和2PC协议在分布式系统中所起的作用并不相同。Paxos协议用于保证同一个数据分片的多个副本之间的数据一致性。当这些副本分布到不同的数据中心时，这个需求尤其强烈。2PC协议用于保证属于多个数据分片上的操作的原子性。这些数据分片可能分布在不同的服务器上，2PC协议保证多台服务器上的操作要么全部成功，要么全部失败。
Paxos协议有两种用法：一种用法是用它来实现全局的锁服务或者命名和配置服务，例如Google Chubby以及Apache Zookeeper。另外一种用法是用它来将用户数据复制到多个数据中心，例如Google Megastore以及Google Spanner。

2PC协议最大的缺陷在于无法处理协调者宕机问题。如果协调者宕机，那么，2PC协议中的每个参与者可能都不知道事务应该提交还是回滚，整个协议被阻塞，执行过程中申请的资源都无法释放。因此，常见的做法是将2PC和Paxos协议结合起来，通过2PC保证多个数据分片上的操作的原子性，通过Paxos协议实现同一个数据分片的多个副本之间的一致性。另外，通过Paxos协议解决2PC协议中协调者宕机问题。当2PC协议中的协调者出现故障时，通过Paxos协议选举出新的协调者继续提供服务。

跨机房部署

在分布式系统中，跨机房问题一直都是老大难问题。机房之间的网络延时较大，且不稳定。跨机房问题主要包含两个方面：数据同步以及服务切换。跨机房部署方案有三个：集群整体切换、单个集群跨机房、Paxos选主副本。下面分别介绍。

1.集群整体切换

集群整体切换是最为常见的方案。如图3-10所示，假设某系统部署在两个机房：机房1和机房2。两个机房保持独立，每个机房部署单独的总控节点，且每个总控节点各有一个备份节点。当总控节点出现故障时，能够自动将机房内的备份节点切换为总控节点继续提供服务。另外，两个机房部署了相同的副本数，例如数据分片A在机房1存储的副本为A11和A12，在机房2存储的副本为A21和A22。在某个时刻，机房1为主机房，机房2为备机房。

机房之间的数据同步方式可能为强同步或者异步。如果采用异步模式，那么，备机房的数据总是落后于主机房。当主机房整体出现故障时，有两种选择：要么将服务切换到备机房，忍受数据丢失的风险；要么停止服务，直到主机房恢复为止。因此，如果数据同步为异步，那么，主备机房切换往往是手工的，允许用户根据业务的特点选择“丢失数据”或者“停止服务”。

如果采用强同步模式，那么，备机房的数据和主机房保持一致。当主机房出现故障时，除了手工切换，还可以采用自动切换的方式，即通过分布式锁服务检测主机房的服务，当主机房出现故障时，自动将备机房切换为主机房。

2.单个集群跨机房

上一种方案的所有主副本只能同时存在于一个机房内，另二种方案是将单个集群部署到多个机房，允许不同数据分片的主副本位于不同的机房，如图3-11所示。每个数据分片在机房1和机房2，总共包含4个副本，其中A1、B1、C1是主副本，A1和B1在机房1，C1在机房2。整个集群只有一个总控节点，它需要同机房1和机房2的所有工作节点保持通信。当总控节点出现故障时，分布式锁服务将检测到，并将机房2的备份节点切换为总控节点。

如果采用这种部署方式，总控节点在执行数据分布时，需要考虑机房信息，也就是说，尽量将同一个数据分片的多个副本分布到多个机房，从而防止单个机房出现故障而影响正常服务。

3.Paxos选主副本

在前两种方案中，总控节点需要和工作节点之间保持租约（lease），当工作节点出现故障时，自动将它上面服务的主副本切换到其他工作节点。

如果采用Paxos协议选主副本，那么，每个数据分片的多个副本构成一个Paxos复制组。如图3-12所示，B1、B2、B3、B4构成一个复制组，某一时刻B1为复制组的主副本，当B1出现故障时，其他副本将尝试切换为主副本，Paxos协议保证只有一个副本会成功。这样，总控节点与工作节点之间不再需要保持租约，总控节点出现故障也不会对工作节点产生影响。

Google后续开发的系统，包括Google Megastore以及Spanner，都采用了这种方式。它的优点在于能够降低对总控节点的依赖，缺点在于工程复杂度太高，很难在线下模拟所有的异常情况。